andygrove opened a new issue, #5259: URL: https://github.com/apache/datafusion-comet/issues/5259
## Describe the bug [#4658](https://github.com/apache/datafusion-comet/pull/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](https://github.com/apache/datafusion-comet/pull/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 https://github.com/apache/datafusion-comet/pull/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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
