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]

Reply via email to