andygrove commented on PR #5318: URL: https://github.com/apache/datafusion-comet/pull/5318#issuecomment-5555352046
Thanks for this, and thanks @sunchao for the earlier rounds. I went through the current head (`e8af41e48`) and all four of the previous P2 findings look genuinely addressed: `run_group` physically shrinks the working batch between instructions, `project` goes through `cast_and_stamp_schema`, the instruction groups are real case-class fields so `prepareSubqueries` can find scalar subqueries, and the cardinality set reserves against `estimate_memory_size` at rehash boundaries. I also checked the null-row-id handling against `UnsafeRow.setNullAt` semantics and it is correct, and confirmed the `MergeRows.Instruction` API (`condition`, `outputs`, `ROW_ID`) is identical in 3.5.9 and 4.1.0. Nice piece of work. ## Relationship to #5122 The scoping is right. #5122 covers four operators (`MergeRowsExec`, `ReplaceDataExec`, `WriteDeltaExec`, and `InsertOnlyMergeExec` on 4.2+), and this takes only the first, independent of native writes, which is exactly what the issue proposed. Two things worth naming: - #5122's motivating workload is Iceberg CDC upsert. Testing against Spark's own `InMemoryRowLevelOperationTableCatalog` is a good call for pinning the Spark-core contract, but it leaves the actual motivating case untested with the flag on. See point 7 below. - #5122 notes that full acceleration needs #5121 (DSv2 writes). Until that lands the writer stays on the JVM, so this moves the `ColumnarToRow` boundary up one node rather than removing it. That is the crux of point 5. `InsertOnlyMergeExec` is correctly out of scope here, but on Spark 4.2 an insert-only MERGE plans that instead of `MergeRowsExec`, so this operator will silently not engage. A follow-up issue under #5122 would be better than expanding this PR. @peterxcli asked to be tagged for review on the issue thread. ## Findings **1. Rebase and get a green CI run.** The branch conflicts with `main` in two places. `operator.proto` field 120 is now `IcebergWrite iceberg_write = 120`, so `MergeRows` needs to move to 121, and `operators/mod.rs` conflicts too. `main` also moved to DataFusion 55.0 and Arrow 59.2 in `75fdddc92`. `gh pr checks` currently reports no checks at all on this head. A 2400-line change touching the planner, the proto, error mapping and three Spark profiles really needs a full build before it can be evaluated. **2. `CometMergeRows.scala` is byte-for-byte identical in `spark-3.5/` and `spark-4.x/`.** I diffed them and the only difference is the path. Same for `ShimCometMergeRows.scala` and the three-line `ShimSparkErrorConverter` case. That is 174 lines of real fallback logic that has to be kept in sync by hand across four Spark profiles. `ShimCometWindowGroupLimit` sets a precedent for duplication, but that is a 20-line shim, not a serde. The `spark-4.1+` source root with the `spark-none` placeholder is the existing pattern for a shared "3.5 and later" root. Could the serde move there, leaving only the thin class-registration shim per version? **3. `getSupportLevel` returns `Compatible(None)` while three divergences are documented.** The one that stands out is metrics. On Spark 4.1+, `MergeRowsExec` publishes eight per-clause counters (`numTargetRowsCopied`, `numTargetRowsInserted`, `numTargetRowsUpdated`, `numTargetRowsDeleted`, and the four matched / not-matched-by-source breakdowns). Enabling this operator replaces all of them with generic Comet metrics, and 4.1 is the default build profile. Should this be `Incompatible(Some(...))` with a matching `getIncompatibleReasons()` so it surfaces on the compat page? Implementing the counters may not be much work either: the proto already has a per-instruction message and `run_group` already computes `fire.true_count()` per instruction, so a `Context` enum on `MergeInstruction` plus per-instruction counts would cover it. Also, the doc says "Spark 4.x's `MergeRowsExec` exposes eight metrics" but 4.0.1 has none. They arrived in 4.1 alongside the `Context` field on `Keep`. **4. Output row order diverges from Spark and is not on the compat page.** `run_group` emits rows grouped by the instruction that produced them, and `process_batch` concatenates matched, then not-matched, then not-matched-by-source. Spark emits in input row order. The Rust doc comment argues this is safe because `DistributionAndOrderingUtils` places the required repartition and sort above `MergeRows`, which holds for a connector that declares one. A V2 table that declares neither ends up with a different physical row order in the written files than Spark produces. That belongs on `compatibility/operators.md` alongside the other two caveats, since it is what a user would actually notice from an unordered `SELECT *`. **5. Could you add some benchmark numbers?** The rationale is that MERGE falls back today, but with the writer still on the JVM this moves the `ColumnarToRow` up one node rather than removing it, and `process_batch` adds a `concat_batches` full copy per input batch on top of a `filter_record_batch` per instruction. A before and after on a realistic CDC-shaped MERGE would tell us whether the operator earns the flag now, or whether the win really arrives with #5121. **6. Test gaps.** `WHEN NOT MATCHED BY SOURCE` never executes end to end. It appears only in the equality and canonicalization test, which builds the plan but never runs the query, so `not_matched_by_source_instructions` is covered by Rust unit tests only. Copy-on-write `WHEN MATCHED THEN DELETE` is also uncovered at the SQL level, only the delta variant is tested, and multi-clause first-match-wins with conditions is likewise Rust-only. It would also be worth one test that the default (flag off) falls back cleanly and matches Spark, since that is the path every user is on today. Separately, the four `withSQLConf(CometConf.COMET_ENABLED.key -> "true", ...)` calls can drop the `COMET_ENABLED` entry, `CometTestBase` already enables Comet by default. **7. Iceberg MERGE coverage.** The suite scaladoc points at `CometIcebergWriteActionSuite` for Iceberg coverage, but nothing there enables `mergeRows`, so with the flag on there is no Iceberg MERGE test anywhere. Given that is the workload #5122 is about, and given how the native Iceberg writer behaves differently depending on whether its child converted natively, one Iceberg MERGE test with the flag on would be worth adding. **8. Why is the output schema derived from the projections rather than `output_types`?** `merge.output_types` is serialized on the wire but only read in the branch where every instruction is a `Discard`, which effectively never happens since `RewriteMergeIntoTable` appends a catch-all `Keep`. The live path goes through `ExpandExec::build_schema`. Since `project` already funnels every column through `cast_and_stamp_schema`, always building the schema from Spark's declared `output_types` would make the native output schema provably equal to what the JVM expects across FFI, and would delete the two-branch derivation. If the derived types are preferred in order to avoid a lossy safe cast on, say, a decimal, that reasoning is worth a comment. Otherwise the field is dead weight. **9. Comment density in `merge_rows.rs`.** A good chunk of the 1320 lines is prose, and some of it is genuinely valuable, the `UnsafeRow.setNullAt` research and the cardinality-versus-instruction ordering caveat especially. But the twelve-line comment on the `baseline` field, the twelve lines on `SEEN_HASH_TABLE_SLACK_BYTES`, and several blocks that pre-argue with a hypothetical reviewer put this well above the density of the neighbouring operators, and that kind of prose rots first. A trim pass down to the non-obvious facts would help. **10. `ShimCometMergeRows` creates a new `org.apache.comet.rules.shims` package.** Every sibling shim, including `ShimCometWindowGroupLimit`, `ShimCometStreaming` and `ShimSubqueryBroadcast`, lives in `org.apache.comet.shims`. Was the new package deliberate? -- 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]
