jordepic commented on PR #4658: URL: https://github.com/apache/datafusion-comet/pull/4658#issuecomment-5209755822
Thanks @comphead — all nine items are addressed and pushed. Details, including one place where the review's expectation turned out to be version-dependent: **F1 (metrics double-registration)** — fixed. The write's custom metrics are now registered on `IcebergCommitExec` only. Worth noting why that's the complete fix: every Iceberg write custom metric (`addedDataFiles`, `addedRecords`, …) is a commit-report metric delivered via `Write.reportDriverMetrics()` on the driver; Iceberg's write-side `DataWriter`s don't override `currentMetricsValues()`, so the executor-side `CustomMetrics.updateMetrics` plumbing on the writer could never move a metric and has been removed along with the writer's now-unused `write` field. `IcebergWriteExec` keeps only `numOutputRows`. A new test pins single registration and the committed `addedDataFiles` value. **PERF-1 (per-row closure + `Method.invoke`)** — fixed, and the reflection is gone entirely rather than hoisted. `DataWriter.write(metadata, row)` exists at compile time on every Spark 4.x profile, so the dispatch now goes through a new `IcebergDataWriterShim`: the `spark-4.x` shim calls `writer.write(metadata, row)` directly; the `spark-3.x` shim throws (unreachable — operation-coded rows only exist on 4.0+ and `IcebergReplaceDataShim` returns no dispatch info on 3.x). No per-row allocation, no reflective dispatch. **eh-1 (serialization outside the abort guard)** — fixed. `serializeMessage(writer.commit())` now runs inside `tryWithSafeFinallyAndFailureCallbacks`, so a serialization failure after a successful task commit triggers `writer.abort()`, which deletes the already-finalised files via `SparkCleanupUtil`. (Stock Spark has the same exposure — task-result serialization happens outside `DataWritingSparkTask`'s guard — so the split path is now strictly safer on this edge, not just equal.) **Test gaps** — all six added to `CometIcebergWriteActionSuite` (suite is now 30 tests): - **primitives**: BOOLEAN/TINYINT/SMALLINT/INT/BIGINT/FLOAT/DOUBLE/DECIMAL(18,4)/STRING/BINARY/DATE/TIMESTAMP round-trip, including an all-NULL row. - **NULLs**: an all-NULL row plus per-column NULLs on the standard fixture. - **nested types**: STRUCT/ARRAY/MAP round-trip including NULL struct fields and NULL containers. - **MERGE variants**: matched-DELETE-only (all versions), and `WHEN NOT MATCHED BY SOURCE` with update + delete legs (gated to Spark 3.5+: Spark 3.4 pairs with Iceberg 1.5.x, whose extensions parser rejects the clause). - **hidden partition transforms**: `bucket(4, id)` + `truncate(2, region)` in one table, and a `days(ts)` table asserting the expected partition count. - **CTAS / RTAS / DataFrameWriterV2** — this is the one where I'd push back on the review's framing: "they should fall through to Spark's own path" is only true on Spark 3.4. On Spark 3.5+ the staged CTAS/RTAS execs run their inner append as its own `AppendData` `QueryExecution` (`V2CreateTableAsSelectBaseExec.writeToTable` → `session.sessionState.executePlan`), which `IcebergWriteStrategy` intercepts like any other append — so CTAS/RTAS actually *do* write through the split operators there, and the first draft of the test failed for exactly that reason. The test now pins both behaviours per version (fall-through on 3.4, split-operator write on 3.5+, correct rows and exactly one snapshot on all versions), and the user guide documents it. Verified locally: full suite green on spark-3.4, spark-3.5, and spark-4.1 profiles, plus scalafix/spotless/scalastyle. -- 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]
