Describe the bug
With spark.comet.iceberg.write.enabled=true and adaptive query execution disabled, every Iceberg copy-on-write row-level operation (DELETE, UPDATE, MERGE) fails with:
java.lang.ClassCastException: class org.apache.spark.sql.vectorized.ColumnarBatch
cannot be cast to class org.apache.spark.sql.catalyst.InternalRow
at scala.collection.Iterator$$anon$9.next(Iterator.scala:584)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at ...WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext
at org.apache.spark.sql.comet.execution.shuffle.CometBypassMergeSortShuffleWriter.write(CometBypassMergeSortShuffleWriter.java:158)
Converting IcebergWriteExec into CometIcebergWriteExec rewrites the write subtree and loses the ColumnarToRow that Spark had inserted between Iceberg's columnar BatchScan (IcebergCopyOnWriteScan) and the whole-stage codegen stage above it. At runtime the codegen stage receives ColumnarBatch and casts it to InternalRow.
Executed plan with spark.comet.iceberg.write.enabled=false (works):
IcebergWrite
+- *(2) CometColumnarToRow
+- CometSort
+- CometColumnarExchange hashpartitioning(dep, 10), REBALANCE_PARTITIONS_BY_COL, CometColumnarShuffle
+- *(1) Project
+- *(1) Filter
+- *(1) BroadcastHashJoin
:- *(1) ColumnarToRow <-- present
: +- BatchScan IcebergCopyOnWriteScan
Same query with spark.comet.iceberg.write.enabled=true (fails):
CometIcebergWrite
+- CometSort
+- CometColumnarExchange hashpartitioning(dep, 10), REBALANCE_PARTITIONS_BY_COL, CometColumnarShuffle
+- *(1) Project
+- *(1) Filter
+- *(1) BroadcastHashJoin
:- BatchScan IcebergCopyOnWriteScan <-- ColumnarToRow gone
With AQE enabled the failure disappears, because transitions are re-inserted when each stage is materialised. Every existing Comet Iceberg suite runs with AQE on, which is why this was never caught.
Steps to reproduce
Spark 4.1.3, Iceberg 1.11.0. Session config:
spark.comet.scan.icebergNative.enabled = true
spark.comet.write.iceberg.splitOperator.enabled = true
spark.comet.iceberg.write.enabled = true
spark.sql.adaptive.enabled = false
CREATE TABLE cat.db.t (id INT, dep STRING, amount DOUBLE)
USING iceberg PARTITIONED BY (dep)
TBLPROPERTIES (
'format-version'='2',
'write.delete.mode'='copy-on-write',
'write.distribution-mode'='none',
'write.spark.fanout.enabled'='false');
INSERT INTO cat.db.t VALUES (1,'hr',1.0),(2,'hr',2.0),(3,'eng',3.0),(4,'eng',4.0),(5,'ops',5.0);
-- src is any small view, e.g. SELECT explode(array(1,3)) AS id
DELETE FROM cat.db.t WHERE id IN (SELECT id FROM src);
DELETE, UPDATE and MERGE all fail the same way, on the table's main branch and on a named branch.
Isolated with a config matrix on that DELETE:
| split operator |
native write |
localTableScan |
result |
| on |
on |
on |
ClassCastException |
| on |
on |
off |
ClassCastException |
| on |
off |
on |
ok |
| on |
off |
off |
ok |
| off |
off |
off |
ok |
So the split-operator plan alone does not trigger it and spark.comet.exec.localTableScan.enabled is irrelevant. It needs the native write flag plus AQE off.
Expected behavior
The copy-on-write operation succeeds with the native writer regardless of whether AQE is enabled, or the operator declines conversion rather than producing a plan that violates the columnar contract.
Additional context
Found by turning the two Iceberg write flags on by default in #5677 and reading the Iceberg Spark SQL job logs. This is the dominant failure in Iceberg's spark-extensions suites across Iceberg 1.8.1, 1.9.1, 1.10.0 and 1.11.0.
Note that Iceberg's ExtensionsTestBase sets ADAPTIVE_EXECUTION_ENABLED to RANDOM.nextBoolean() per session, so the number of extension tests this takes down swings a lot between runs of identical code (99 failures in one run and 31 in the next, on the same commit's behaviour). That randomisation should be kept in mind when reading those job results.
Part of #5649.
Describe the bug
With
spark.comet.iceberg.write.enabled=trueand adaptive query execution disabled, every Iceberg copy-on-write row-level operation (DELETE, UPDATE, MERGE) fails with:Converting
IcebergWriteExecintoCometIcebergWriteExecrewrites the write subtree and loses theColumnarToRowthat Spark had inserted between Iceberg's columnarBatchScan (IcebergCopyOnWriteScan)and the whole-stage codegen stage above it. At runtime the codegen stage receivesColumnarBatchand casts it toInternalRow.Executed plan with
spark.comet.iceberg.write.enabled=false(works):Same query with
spark.comet.iceberg.write.enabled=true(fails):With AQE enabled the failure disappears, because transitions are re-inserted when each stage is materialised. Every existing Comet Iceberg suite runs with AQE on, which is why this was never caught.
Steps to reproduce
Spark 4.1.3, Iceberg 1.11.0. Session config:
DELETE, UPDATE and MERGE all fail the same way, on the table's main branch and on a named branch.
Isolated with a config matrix on that DELETE:
So the split-operator plan alone does not trigger it and
spark.comet.exec.localTableScan.enabledis irrelevant. It needs the native write flag plus AQE off.Expected behavior
The copy-on-write operation succeeds with the native writer regardless of whether AQE is enabled, or the operator declines conversion rather than producing a plan that violates the columnar contract.
Additional context
Found by turning the two Iceberg write flags on by default in #5677 and reading the Iceberg Spark SQL job logs. This is the dominant failure in Iceberg's
spark-extensionssuites across Iceberg 1.8.1, 1.9.1, 1.10.0 and 1.11.0.Note that Iceberg's
ExtensionsTestBasesetsADAPTIVE_EXECUTION_ENABLEDtoRANDOM.nextBoolean()per session, so the number of extension tests this takes down swings a lot between runs of identical code (99 failures in one run and 31 in the next, on the same commit's behaviour). That randomisation should be kept in mind when reading those job results.Part of #5649.