Native Iceberg write drops a ColumnarToRow transition when AQE is disabled

[andygrove]

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.