Iceberg write split-operator plan: CI failures when enabled by default

[andygrove]

Describe the bug

#4658 adds Comet's split-operator plan for Iceberg V2 writes (IcebergWriteExec + IcebergCommitExec) behind spark.comet.write.iceberg.splitOperator.enabled, defaulting to false. Because the feature ships off, nothing in CI exercises it.

#5255 is a throwaway audit PR: #4658 rebased onto main with that default flipped to true, so the full matrix runs against the split-operator plan. This issue catalogues what it broke, so the work can be scoped before the feature can be enabled by default.

Audit run: https://github.com/apache/datafusion-comet/actions/runs/30920419488

8 jobs failed, and every failure falls into one of four buckets. All four are caused by the same underlying change: the write is no longer a single V2TableWriteExec/AppendData node, it is IcebergCommit -> IcebergWrite -> <child>.

Failure summary

#	Symptom	Failing tests	Jobs
1	Expected at least one captured plan with AppendData but got none	CometIcebergRewriteActionSuite, 4 tests	Comet [scans] on Spark 3.4 / 3.5 / 4.0 / 4.1
2	ClassCastException: IcebergCommitExec cannot be cast to V2TableWriteExec	TestSystemFunctionPushDownInRowLevelOperations, 7-12 tests	Iceberg extensions on 1.8 / 1.9 / 1.10 / 1.11
3	Snapshot summary should contain merge metric: spark.merge-into.num-target-rows-copied	TestCopyOnWriteMergeMetrics, 36 parameterizations	Iceberg extensions 1.11
4	Cached table should reflect session schema changes	TestCachedTableRefresh.testCachedTableWithSessionSchemaChangeAddColumn, 2 parameterizations	Iceberg extensions 1.11

Note that PR Build (Linux) / Spark 4.2, JDK 17 [scans] passing is not a signal: Iceberg is not on the 4.2 classpath, so all the Iceberg suites cancel there.

---

1. CometIcebergRewriteActionSuite matches on AppendData (4 tests, all Spark versions)

• binPack rewrite reads each file group via CometIcebergNativeScan
• sort rewrite runs scan, exchange, and sort natively in Comet
• single-column zOrder rewrite runs scan, native exchange, and sort natively in Comet
• binPack rewrite applies positional and equality deletes during compaction (MOR)

List() was empty Expected at least one captured plan with AppendData but got none.
--- plan 0 ---
== Physical Plan ==
IcebergCommit (4)
+- IcebergWrite (3)
   +- * CometColumnarToRow (2)
      +- CometIcebergNativeScan (1)

Iceberg's rewrite_data_files action writes its compacted output through AppendData, which the split operator replaces. The rewrite itself is still correct: the row-preservation and file-count assertions in these same tests pass, only the plan-shape filter finds nothing.

The plan filter is at spark/src/test/scala/org/apache/comet/CometIcebergRewriteActionSuite.scala:207 and :269. It should accept either write shape rather than matching AppendData alone.

2. Iceberg's own tests cast the plan root to V2TableWriteExec (1.8 / 1.9 / 1.10 / 1.11)

java.lang.ClassCastException: class org.apache.spark.sql.comet.IcebergCommitExec cannot be cast to
class org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec
    at TestSystemFunctionPushDownInRowLevelOperations.executeAndCollectFunctionCalls(...:316)

Affects the copy-on-write delete tests on every version (bucket-in-predicate, bucket-eq-predicate, years, months, days, hours, truncate), plus the copy-on-write update and merge variants on 1.9 / 1.10 / 1.11. 7 failures on 1.8, 12 on each of 1.9 / 1.10 / 1.11.

executeAndCollectFunctionCalls reaches into the executed plan and casts it to V2TableWriteExec to pull out the write's query. With the split operator the root is IcebergCommitExec, which is not a V2TableWriteExec.

Two possible directions:

• Patch the test in dev/diffs/iceberg/{1.8.1,1.9.1,1.10.0,1.11.0}.diff to unwrap IcebergCommitExec (none of the four diffs touch this file today).
• Or make the split-operator plan present a V2TableWriteExec-compatible root, so third-party code that pattern-matches on the standard Iceberg write shape keeps working. This is the more interesting question for the feature generally: any external tooling that inspects the physical plan of an Iceberg write sees a shape it does not recognize.

3. Merge metrics missing from the snapshot summary (1.11, 36 failures)

java.lang.AssertionError: [Snapshot summary should contain merge metric: spark.merge-into.num-target-rows-copied]
Expecting actual:
  {"added-data-files"="2", ..., "spark.app.id"="local-...", "total-records"="3"}
to contain key:
  "spark.merge-into.num-target-rows-copied"
    at TestMergeMetrics.assertMergeMetric(TestMergeMetrics.java:299)

All six TestCopyOnWriteMergeMetrics methods fail across all six parameterizations: testMergeMetricsWithAllClauses, testMergeMetricsWithMatchedDelete, testMergeMetricsWithMatchedUpdate, testMergeMetricsWithMultipleUpdatesAndDeletes, testMergeMetricsWithNotMatchedBySourceDelete, testMergeMetricsWithNotMatchedBySourceUpdate. The merge-on-read equivalents pass.

Unlike buckets 1 and 2 this is not a stale test expectation, it is missing behaviour. Iceberg derives the spark.merge-into.* snapshot summary properties from the write's driver-side metrics at commit time. Splitting write from commit appears to lose them: the commit operator does not carry the MergeRows metrics forward into the commit's summary properties. Anything downstream that reads snapshot summaries (auditing, incremental processing) would silently see incomplete metadata.

4. TestCachedTableRefresh (1.11, 2 failures)

java.lang.AssertionError: [Cached table should reflect session schema changes]
Expected size: 2 but was: 1 in:
[[1, 100]]
    at TestCachedTableRefresh.testCachedTableWithSessionSchemaChangeAddColumn(...:211)

Fails for testhive/SparkCatalog and spark_catalog/SparkSessionCatalog. Attribution is a little less direct than for the other three, but the Iceberg 1.11 extensions job passes on other PRs against the same main (e.g. https://github.com/apache/datafusion-comet/actions/runs/30915735530), so it is caused by this change rather than pre-existing. Worth a closer look, since a stale cached table after a schema change is a correctness concern, not a test-shape concern.

---

Follow-on work

1. Relax the AppendData plan assertions in CometIcebergRewriteActionSuite to accept the split-operator shape (test-only).
2. Decide whether Iceberg's V2TableWriteExec cast should be patched in the Iceberg diffs, or whether the split-operator plan should preserve a compatible root node (design question, affects any third-party plan inspection).
3. Propagate merge/write metrics from IcebergWriteExec through to the commit so spark.merge-into.* snapshot summary properties are still written (behaviour gap).
4. Root-cause TestCachedTableRefresh.testCachedTableWithSessionSchemaChangeAddColumn.

Items 1, 3, and 4 look like prerequisites for turning the flag on by default. Item 2 needs a decision before either fix is worth writing.

Steps to reproduce

Check out #5255, which is #4658 rebased with spark.comet.write.iceberg.splitOperator.enabled defaulted to true. Alternatively, set that config to true on top of #4658 and run CometIcebergRewriteActionSuite plus the Iceberg Spark extensions suites.

Expected behavior

The Iceberg write split-operator plan passes the existing Comet and Iceberg test suites when enabled, so the flag can eventually default to true.

Additional context

Relates to #4322. Depends on #4658.