test: audit CI failures from enabling the Iceberg write split-operator plan by default - #5255
test: audit CI failures from enabling the Iceberg write split-operator plan by default#5255andygrove wants to merge 6 commits into
Conversation
Iceberg's V2 writes run as a single `V2ExistingTableWriteExec` command that both
writes files and commits. Spark's `InsertAdaptiveSparkPlan` treats it as a leaf,
so AQE never sees the data sub-query inside it and Comet's columnar rules can't
convert the scans / shuffles / sorts that feed the write.
This injects `IcebergWriteStrategy` ahead of Spark's `DataSourceV2Strategy` (via
`experimentalMethods.extraStrategies`, which is prepended to the planner's
strategy list). For Iceberg `AppendData`, `OverwriteByExpression`, and
`OverwritePartitionsDynamic` it rewrites the single command into two operators:
IcebergCommitExec(batchWrite, refreshCache) <- committer (driver-side)
└── AdaptiveSparkPlanExec <- AQE bubble
└── IcebergWriteExec(batchWrite) <- writer (UnaryExecNode)
└── <data sub-query> <- now visible to AQE / Comet
The writer emits one row per Spark task carrying the Java-serialised
`WriterCommitMessage`; the committer collects them and calls
`BatchWrite.commit(messages)` -- the same call Iceberg-Java makes internally --
then runs Spark's post-write cache refresh.
Two design points keep this stable under AQE, which re-runs the planner on each
materialised stage's `logicalLink`:
- The writer's child is wrapped in `IcebergWriteLogical`, a stable logical
anchor, so each re-plan re-emits only the writer rather than vanishing it or
re-firing the surrounding logical write and duplicating the commit.
- The `BatchWrite` is materialised once (`Write.toBatch()` mints a fresh
instance per call) and shared between committer and writer, so commit-time
validation sees the same scan / emitted-file state the writer used.
`WriteDelta` (merge-on-read) is intentionally not intercepted: its per-task
`DeltaWriter` is row-dispatched and the native writer can't emit position-delete
files, so the split plan would add planning complexity for no acceleration.
Copy-on-write DELETE / UPDATE / MERGE (`ReplaceData`) is handled in a follow-up
commit.
Guarded by `spark.comet.write.iceberg.splitOperator.enabled` (off by default);
when off, writes go straight through Iceberg-Java unchanged. File writing still
runs through Iceberg's JVM writer -- native Parquet write lands in a later
commit.
Adds copy-on-write row-level DML to the split-operator plan. Spark lowers a CoW `DELETE` / `UPDATE` / `MERGE` on a V2 table into a `ReplaceData` logical node; `IcebergWriteStrategy` now matches it (refreshing `originalTable`'s cache) and routes it through the same committer + writer pair as appends/overwrites. The writer gains a per-row dispatch path. On Spark 4.x a `ReplaceData` rewrite prefixes each row with an operation code and carries a `ReplaceDataProjections`; `IcebergWriteExec` applies the row / metadata projections per code (WRITE / WRITE_WITH_METADATA) before handing rows to the `DataWriter`. The two-arg `DataWriter.write(metadata, row)` is looked up reflectively since it is 4.x-only. On Spark 3.4 / 3.5 the rewritten stream is already post-projection, so the shim returns `None` and the writer keeps its plain `write(row)` loop. `ReplaceDataProjections` does not exist on 3.4 / 3.5, so `ReplaceDataDispatchInfo` mirrors it as a version-neutral carrier, populated by the per-version `IcebergReplaceDataShim`. Iceberg 1.5.2 (Spark 3.4) lacks native row-level operations and instead routes UPDATE / MERGE through its own `ReplaceIcebergData` logical node. It has the same field shape as Spark's `ReplaceData` and is matched by FQCN via reflection, so the main module takes no compile dependency on iceberg-spark-extensions.
Flips spark.comet.write.iceberg.splitOperator.enabled from false to true so the split-operator plan is exercised by default, and updates the Iceberg writes user guide to match.
|
cc @jordepic - let's see what breaks and we can file an epic for follow on work |
|
Thanks @andygrove ! I'm following along :) |
|
I have catalogued the CI failures from this audit run in #5259. Run: https://github.com/apache/datafusion-comet/actions/runs/30920419488 8 failing jobs, 4 distinct root causes:
Buckets 3 and 4 are behaviour gaps rather than stale test expectations, so they need fixing before the flag can default to One caveat on reading these results: |
|
Gonna see if I can get all 4 of these patched up. Thanks for the help guys! |
Audit sweep results — CI run 30920419488Full matrix completed on Group 1 — Iceberg's own
|
| Iceberg | Spark | Scala | Java | iceberg-spark |
iceberg-spark-extensions |
iceberg-spark-runtime |
|---|---|---|---|---|---|---|
| 1.8.1 | 3.4.3 | 2.13 | 11 | pass | FAIL | pass |
| 1.9.1 | 3.5.9 | 2.13 | 17 | pass | FAIL | pass |
| 1.10.0 | 3.5.9 | 2.13 | 17 | pass | FAIL | pass |
| 1.11.0 | 4.1.3 | 2.13 | 17 | pass | FAIL | pass |
The failure is version-independent, and it is confined to extensions. That is a useful narrowing:
iceberg-spark (core read/write paths, DDL/DML, pushdown) and iceberg-spark-runtime (shaded-JAR
smoke test) are green on all four versions — so plain appends and overwrites survive the split
operator. iceberg-spark-extensions is where the row-level copy-on-write UPDATE/DELETE/MERGE
tests and the stored procedures (rewrite_data_files, rewrite_manifests, expire_snapshots, ...)
live, which is exactly the surface ReplaceData interception and procedure-driven writes touch.
I do not yet have the per-test breakdown for these four jobs — see "Blocked on" below.
Group 2 — Comet's own scans bucket: fails on every profile except Spark 4.2
| Runner | Profile | Result |
|---|---|---|
| Linux | Spark 3.4, JDK 11, Scala 2.12 | FAIL |
| Linux | Spark 3.5, JDK 17, Scala 2.13 | FAIL |
| Linux | Spark 4.0, JDK 21 | FAIL |
| Linux | Spark 4.1, JDK 17 | FAIL |
| Linux | Spark 4.2, JDK 17 | pass |
| macOS | Spark 4.0, JDK 17, Scala 2.13 | FAIL |
Root cause confirmed locally on Spark 3.5 / Scala 2.13 / JDK 17 —
org.apache.comet.CometIcebergRewriteActionSuite, 4 of 5 tests:
binPack rewrite reads each file group via CometIcebergNativeScansort rewrite runs scan, exchange, and sort natively in Cometsingle-column zOrder rewrite runs scan, native exchange, and sort natively in CometbinPack rewrite applies positional and equality deletes during compaction (MOR)
All four fail on Expected at least one captured plan with AppendData but got none. Iceberg's
rewrite_data_files action writes its compacted output through AppendData; the split operator
replaces that with IcebergWrite under IcebergCommit, so the hasNode("AppendData") filter at
CometIcebergRewriteActionSuite.scala:207 and :269 matches nothing. The rewrite itself is
correct — the row-preservation, file-count, and delete-application assertions all pass; only the
plan-shape expectation is stale. Control run: the same suite is 5/5 green with the flag back at
false, so this PR is the cause.
Spark 4.2 passing is a red herring rather than a signal that 4.2 is unaffected: that profile pulls
iceberg-spark-runtime-4.0 1.10.0 (spark/pom.xml:318-324) because no 4.2 runtime is published, so
the rewrite-action tests most likely cancel there instead of running. Worth confirming rather than
assuming.
Everything else is green
All other pr_build buckets (exec, shuffle, expressions) on all six profiles, Spark SQL Tests (Spark 3.5), Spark SQL Tests (Spark 4.1), the Rust tests, and every lint job passed.
CometIcebergWriteActionSuite (19 tests, added by #4658) passes locally with the new default.
Proposed follow-on work
- Teach
CometIcebergRewriteActionSuiteto accept either write shape. Replace the bare
hasNode("AppendData")filter with one that also matchesIcebergWrite, so the suite is
correct with the feature on or off. Small and well understood — this clears all of Group 2. - Triage
iceberg-spark-extensions. The real work. Needs the per-test failure list before it
can be scoped; it may be more stale plan-shape expectations, or genuine gaps inReplaceData
/ procedure handling. - Confirm the Spark 4.2 pass is test-cancellation rather than real coverage.
- Reconsider the config category. The entry is
CATEGORY_TESTING; if it ships on by default
it probably belongs underCATEGORY_EXECso it appears in the user-facing config table rather
than under "Development & Testing Settings".
Blocked on
The Iceberg reusable workflow uploads only the native library — no test reports — so the
iceberg-spark-extensions failure detail exists only in the job logs, and those are not reachable
from my environment (the run's log/artifact blob host is not on my egress allowlist). If someone can
paste the failing test names from any one of the four iceberg-spark-extensions jobs, that is enough
to scope item 2. Separately, it may be worth having
iceberg_spark_test_reusable.yml upload the Gradle test reports on failure, so this is
self-service next time.
|
@andygrove I'm running iceberg-spark-extensions suite locally now. Would you mind kicking off the suite in the meantime? Thanks for your help on the detailed audit here. |
3 participants
Which issue does this PR close?
Relates to #4322. Follow-up experiment on top of #4658.
Rationale for this change
This is a CI audit sweep, not a merge candidate. It is intentionally left red.
#4658 adds Comet's split-operator plan for Iceberg V2 writes behind
spark.comet.write.iceberg.splitOperator.enabled, defaulting tofalse. With the feature off,no existing test exercises it, so we have no visibility into what breaks once it is on.
This PR rebases #4658 onto current
mainand flips that default totrue, so the full CImatrix — Comet's own suites, the Spark SQL suites, and the Iceberg Spark suites — runs against
the split-operator plan. The point is to enumerate every failure the feature causes so we can
scope the follow-on work needed before it can ship enabled.
What changes are included in this PR?
Commits 1-5 are #4658 rebased onto
main(only conflict was an additive one inIcebergReflection.ClassNames).Commit 6 is the actual change here:
spark.comet.write.iceberg.splitOperator.enableddefaultfalse->truedocs/source/user-guide/latest/iceberg-writes.mdupdated to describe the new defaultNo test expectations were adjusted — the whole point is to let them fail and be catalogued.
Known failures so far
Reproduced locally (Spark 3.5, Scala 2.13, JDK 17), all four from a single root cause:
org.apache.comet.CometIcebergRewriteActionSuite— 4 of 5 tests fail:binPack rewrite reads each file group via CometIcebergNativeScansort rewrite runs scan, exchange, and sort natively in Cometsingle-column zOrder rewrite runs scan, native exchange, and sort natively in CometbinPack rewrite applies positional and equality deletes during compaction (MOR)Each fails with
Expected at least one captured plan with AppendData but got none. Iceberg'srewrite_data_filesaction writes its compacted output throughAppendData, which the splitoperator now replaces with
IcebergWriteunderIcebergCommit. The rewrite itself is correct(the row-preservation and file-count assertions pass); only the plan-shape expectation is stale.
The same suite passes 5/5 with the flag back at
false, confirming this PR is the cause.Likely fix for follow-on work: teach the suite to accept either write shape, rather than
matching
AppendDataalone.How are these changes tested?
By CI, deliberately.
CometIcebergWriteActionSuite(19 tests, added by #4658) passes locallywith the new default. Iceberg 1.11 / Spark 4.1 runs on every PR, so the Iceberg Spark suites
will report against the split-operator plan without needing the
run-iceberg-testslabel; addthat label if we want 1.8 / 1.9 / 1.10 coverage too.