Support NOT MATCHED BY SOURCE in the GPU MERGE command on Databricks 17.3 [databricks] - #15884
Support NOT MATCHED BY SOURCE in the GPU MERGE command on Databricks 17.3 [databricks]#15884jtwynne wants to merge 18 commits into
Conversation
…17.3 [databricks] MERGE statements with WHEN NOT MATCHED BY SOURCE clauses fell back to the CPU on Databricks 17.3 (NVIDIA#8415). The fallback costs more than the clause itself: the DBR 17.3 CPU merge uses the row-index-set algorithm, whose scan (hidden metadata columns, BitmapAggregator) and rewrite phases the plugin cannot accelerate, so the whole merge runs on the CPU. Extend the shared GPU merge-join processor to evaluate the not-matched- by-source conditions on target rows without a source match. The existing projection series already implements "first matching clause wins", so the target-only rows now go through it with the no-op copy as the default instead of being copied unconditionally. In the DBR 17.3 GpuMergeIntoCommand, follow the OSS Delta 2.3 shape: findTouchedFiles uses a right outer join and no target-only data skipping when such clauses are present, writeAllChanges builds the clause conditions and outputs (update, delete, CDC images) for the new clause type, the insert-only shortcut is disabled when the clauses are present, the row-based fallback processor handles the clauses the same way, the commit records the clause predicates, and the per-clause-type update and delete metrics reported by the CPU command are added. Remove the veto in the DBR 17.3 MergeIntoCommandMetaShim; the deletion vector veto is unchanged. Enable the existing not-matched-by-source integration test on Databricks 17.3, turn the DBR 17.3 fallback test into a test that asserts the GPU merge processor is in the plan, and add a composite null-safe key case. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
…rows [databricks] In GpuRapidsProcessDeltaMergeJoinIterator.processProjectionSeries the closeOnExcept guard wrapped the batch that splitBatchAndClose had already closed, while the not-matched remainder produced by the split was left open. When a clause projection throws (for example the CheckOverflowInTableWrite error path exercised by test_delta_merge_check_overflow_in_table_write_error on Databricks 17.3) the remainder leaked and the cleaner thread reported the joined batch's columns as leaked device column vectors. Guard the not-matched remainder instead of the consumed input. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
Greptile SummaryAdds GPU support for Databricks 17.3 MERGE statements containing
Confidence Score: 5/5The PR appears safe to merge, with the previously reported matched-delete metric overcount corrected and no new actionable findings. The current implementation compensates both aggregate and matched-delete counters for duplicate unconditional deletes, and the added coverage exercises the new clause paths, duplicate-match rules, NULL semantics, CDC output, helper-column collisions, schema evolution, and row tracking. Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
A[Source and target rows] --> B[Join using MERGE condition]
B --> C{Row category}
C -->|Matched pair| D[First matching WHEN MATCHED clause]
C -->|Source only| E[First matching WHEN NOT MATCHED clause]
C -->|Target only| F[First matching WHEN NOT MATCHED BY SOURCE clause]
D --> G[Update, delete, or copy]
E --> H[Insert or discard]
F --> I[Update, delete, or copy]
G --> J[Apply CDC, metrics, and row tracking]
H --> J
I --> J
J --> K[Write rewritten Delta files and commit]
Reviews (14): Last reviewed commit: "Merge branch 'main' into db173-merge-not..." | Re-trigger Greptile |
…databricks] An unconditional MATCHED DELETE is the only merge that allows several source rows to match the same target row. The delete counters are incremented once per joined pair and the duplicate count is subtracted afterwards, but the Databricks 17.3 command only compensated numTargetRowsDeleted, so the new numTargetRowsMatchedDeleted metric was committed with the overcount. Apply the same compensation to the matched-delete counter, as OSS Delta and the delta-24x command do. Add test_delta_merge_delete_only_duplicate_source_metrics_db173, which merges three source rows per key with an unconditional MATCHED DELETE and checks that both counters in the commit's operationMetrics equal the number of target rows deleted, on the CPU and on the GPU. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
|
Good catch! OSS Delta and the delta-24x command compensate both counters and I carried over only the first. Fixed in |
| // Target rows without a source match are handled by the NOT MATCHED BY SOURCE clauses. | ||
| // A target row that satisfies none of the clause conditions is copied unchanged. | ||
| val targetNotMatchedBatches = closeOnExcept(targetMatchBatch) { _ => | ||
| processProjectionSeries(targetNoMatchBatch, |
There was a problem hiding this comment.
processProjectionSeries partitions rows using predicate and predicate.not(). If a nullable NOT MATCHED BY SOURCE condition evaluates to NULL, both cuDF filter masks exclude the row because NOT NULL remains NULL. The row consequently reaches neither the clause output nor noopCopyOutput, silently removing an unmatched target row. Could we coalesce each condition to false before splitting and add a nullable-condition test? SQL MERGE should preserve the target row when no clause condition evaluates to true.
There was a problem hiding this comment.
Gotcha! And it looks like it isn't specific to the new clause. Every condition goes through the same split, so a WHEN MATCHED AND s.flag with a NULL flag was deleting the target row on the GPU where the CPU copies it. Fixed in ded963e, one level below these lines in splitBatchAndClose (the one place every clause type passes through) rather than just on the not-matched-by-source conditions: NULLs in the evaluated condition get replaced with false before the two filters (masks without NULLs are used as-is), so the row falls through to the next clause or the default the same way the CPU row processor does.
Tests: test_delta_merge_nullable_matched_conditions (all Delta versions; matched update and delete clauses plus an insert clause with NULL conditions, expected rows spelled out) and test_delta_merge_nullable_not_matched_by_source_condition (OSS 4.1 and 17.3+; adds a NOT MATCHED BY SOURCE condition that's NULL for target rows with a NULL column, GPU processor asserted in the plan). Without the fix the first one loses the four matched rows with a NULL flag and the second also loses the two target-only rows with a NULL column.
| .withColumn(FILE_NAME_COL, input_file_name()) | ||
| val joinToFindTouchedFiles = | ||
| sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), "inner") | ||
| sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), joinType) |
There was a problem hiding this comment.
DBR 16.0+ uses both the ON condition and WHEN MATCHED conditions when determining whether multiple source rows ambiguously match one target row. This join is subsequently grouped by target row ID without filtering through the matched-clause conditions, so it appears to retain the older OSS ON-only behavior. For example, two source rows may satisfy ON while only one satisfies WHEN MATCHED AND source.apply_update; DBR 17.3 CPU should accept that merge, while this path reports multiple matches. Since this PR makes queries containing NOT MATCHED BY SOURCE GPU-eligible, those queries no longer fall back to CPU. Could we align the duplicate-match predicate with DBR 17.3 and add a CPU/GPU parity test for this case?
There was a problem hiding this comment.
Yes good call! The 17.3 command has been using the OSS ON-only count for every merge it runs, this PR just makes it easier to hit. Fixed in eb8f690.
findTouchedFilesnow counts, per target row, the pairs that satisfy a WHEN MATCHED condition (no condition = true) and only raises the error when more than one does. The unconditional-delete exception is unchanged.- The join-based write needs one extra step the CPU row-index merge doesn't. When a target row has several joined pairs and at most one of them takes an action,
writeAllChangeskeeps one pair per target row (the one that takes the action if there is one) using arow_numberwindow over the target and source row ids. The pairs it drops are matched source rows that don't take an action, so they don't get inserted or applied, and a row with no applying pair gets copied once. The window only runs whenfindTouchedFilesactually found rows like that.
I checked what DBR 17.3 CPU does first (default and enableLowShuffle=false) and turned those shapes into the parity tests: test_delta_merge_duplicate_source_rows_matched_conditions_db173 (six accepted shapes, including no WHEN MATCHED clause and a conditional delete; table and per-clause counters have to match the CPU, GPU processor asserted in the plan) and test_delta_merge_duplicate_source_rows_ambiguous_error_db173 (three shapes both engines reject with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE). Ran them on a 17.3 cluster with the rest of the merge tests: 17 passed, 3 skipped, 0 failed, no leak reports.
|
Thanks for this PR @jtwynne |
…ks 17.3 [databricks] Databricks Runtime 16.0 and later report several source rows matching the same target row as an error only when more than one of them satisfies the ON condition and a WHEN MATCHED clause condition (an undefined condition is implicitly true). Source rows that match on ON alone take no action: they are not inserted, they do not flag the row for the NOT MATCHED BY SOURCE clauses, and they do not make it ambiguous. The Databricks 17.3 GPU command still counted every joined pair, the OSS ON-only rule, so a merge the CPU accepts failed on the GPU; with NOT MATCHED BY SOURCE merges no longer falling back, that gap now reaches more queries. findTouchedFiles counts, per target row, the joined pairs and the pairs that take a WHEN MATCHED action, and raises the multiple-match error only for the second count (the unconditional-delete exception is unchanged). When some target rows have several joined pairs of which at most one takes an action, writeAllChanges keeps one pair per target row, preferring the one that takes the action, with a row_number window over target and source row ids; the dropped pairs are matched source rows that take no action, and when none does the surviving pair copies the target row unchanged. The window only runs when such rows exist, so merges without duplicate matches are unchanged. Add test_delta_merge_duplicate_source_rows_matched_conditions_db173 (six accepted shapes: one or no effective match with and without NOT MATCHED BY SOURCE, no WHEN MATCHED clause, a conditional delete, three matches with one effective; the tables and the per-clause row counters must match the CPU) and test_delta_merge_duplicate_source_rows_ambiguous_error_db173 (three shapes both engines reject: two effective matches, each row taking a different clause, a target-only condition). Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
…tabricks] A MERGE clause condition that evaluates to NULL is false: the row moves on to the next clause or to the default action, which copies a target row unchanged or skips a source row. The shared GPU merge processor splits each batch with the condition and with its negation through cuDF filters, and a NULL mask entry passes neither (NOT NULL is NULL), so a row with a NULL condition was silently dropped from the output. For matched rows that turned a "WHEN MATCHED AND s.flag" with a NULL flag into a deleted target row; with NOT MATCHED BY SOURCE clauses the same happens to target-only rows. The CPU row processor evaluates such a condition as false. Replace NULLs in the evaluated condition with false before splitting, in the one place every clause type goes through; masks without NULLs are used as they are. Add test_delta_merge_nullable_matched_conditions (all Delta versions: matched update and delete clauses and an insert clause whose conditions are NULL for some rows; expected rows spelled out, CPU and GPU tables compared) and test_delta_merge_nullable_not_matched_by_source_condition (OSS Delta 4.1 and Databricks 17.3+: the same plus a NOT MATCHED BY SOURCE condition that is NULL for target rows with a NULL column, GPU processor asserted in the plan). Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
| rawJoinedDF | ||
| .withColumn(DUPLICATE_MATCH_RANK_COL, row_number().over(onePairPerTargetRow)) | ||
| .filter(col(DUPLICATE_MATCH_RANK_COL) === lit(1)) | ||
| .drop(DUPLICATE_MATCH_RANK_COL, TARGET_ROW_ID_COL, SOURCE_ROW_ID_COL) |
There was a problem hiding this comment.
This helper is added with the fixed name _duplicate_match_rank_:
Dataset.withColumn replaces an existing column with the same name. If the source has a column named _duplicate_match_rank_, this projection replaces it with the window result, and the subsequent drop removes it completely. MERGE action expressions still reference the original source attribute, so the GPU
processor later fails to bind that expression.
I reproduced this on DBR 17.3 with two source rows matching the same target row, where only one satisfies the WHEN MATCHED condition:
-- source:
-- (k=1, apply=true, _duplicate_match_rank_='chosen')
-- (k=1, apply=false, _duplicate_match_rank_='ignored')
-- (k=4, apply=true, _duplicate_match_rank_='inserted')
MERGE INTO target t
USING source s
ON t.k = s.k
WHEN MATCHED AND s.apply THEN
UPDATE SET t.v = s._duplicate_match_rank_
WHEN NOT MATCHED THEN
INSERT (k, v) VALUES (s.k, s._duplicate_match_rank_)The DBR CPU command succeeds, producing chosen for key 1 and inserted for key 4. The GPU command instead aborts with:
Couldn't find _duplicate_match_rank_#140 in
[k#138, apply#139, _source_row_present_#779,
k#476, v#477, _target_row_present_#781]
This duplicate pattern is supported by DBR 16+ because duplicate-match detection considers both the ON condition and WHEN MATCHED conditions:
https://docs.databricks.com/aws/en/delta/merge
_duplicate_match_rank_ is also a valid Databricks identifier: ordinary identifiers may contain letters, digits, and underscores:
https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-identifiers
Databricks advises avoiding underscore-prefixed names because they may be used for pseudo-columns, but it does not reserve this name or reject such columns.
We could rename the constant to another fixed string with a more specific prefix like cudf_spark_delta_merge_duplicate_match_rank to make the collision less likely. But we also can make this robust:
- A helper name generated to be absent from
rawJoinedDFunder Spark's configured name resolver. - A Catalyst
Aliaswith a freshExprId, followed by removal of that exact generated attribute instead of removal by string name.
There was a problem hiding this comment.
Fixed in 108cf7c: each of the three helper names is generated to be absent from the joined sides under the session's resolver (conf.resolver, so case sensitivity follows the session), and the helpers are dropped by those generated names. The fixed names stay the first candidate, so plans without a clash are unchanged. Added test_delta_merge_duplicate_source_rows_helper_column_names_db173: your repro plus a _source_row_id_ column on the source and a _target_row_id_ column on the target, all three read by the clause actions; the table has to match the CPU and the GPU processor has to be in the plan.
This one made me step back and look at what else the port inherited without checking, since the last few rounds were all places where DBR and OSS Delta differ. I went through the DBR 15.0 to 17.3 release notes for MERGE changes and held them against the command: the 16.x multiple-match rule is the earlier commit, source materialization goes through DBR's MergeIntoMaterializeSource, the 17.0 non-deterministic action values are only evaluated once in the write pass, DV tables are vetoed. Nothing else in the code, but two holes in the tests, both in the same commit and green on 17.3: the duplicate-match, helper-name and NULL not-matched-by-source tests now run with CDF on too (the change rows come out of the same window), and test_delta_merge_not_matched_by_source_schema_evolution_db173 covers MERGE WITH SCHEMA EVOLUTION with the new clause. One more thing this made me check: 17.2 turned row tracking on by default for new managed tables, and the merge command had no row tracking test where UPDATE and DELETE do (see the end of the next paragraph).
The older helpers had the same fixed-name shape (_row_id_ and _file_name_ in findTouchedFiles, the two presence flags, the CDC de-dup row ids), here as in OSS Delta and the other shims, so rather than leave them for a follow-up I moved them onto the same generator in bbfabf8. The two control columns needed something else: the processors find _row_dropped_ by name in the output and the command dropped both by name, so a user column with that name was read as the control column or dropped from the written data. The command now passes the position it knows (right after the target output columns, with or without CDC) through an optional field on RapidsProcessDeltaMergeJoin that defaults to the old lookup for the other shims, and drops the control columns by attribute, only when CDC put them in the output. Tests: test_delta_merge_internal_column_names_db173 (_row_id_ and _file_name_ as user columns on both sides, with and without CDF), test_delta_merge_control_column_names_gpu_db173 (_row_dropped_ and _incr_row_count_ as user columns, GPU only against spelled-out rows, because the DBR CPU command rejects user columns named like its presence flags or its _row_dropped_; the presence-flag names are generated the same way and stay untested for that reason) and test_delta_merge_delete_only_duplicate_cdc_internal_column_names_db173 for the CDC delete de-dup row ids. The CDC de-dup test also tripped over something older: the delete de-dup with an insert clause aliases an untyped null for the row id the other side lacks, the insert output carries it as a long, and the GPU processor can't concatenate the two, so that literal is typed now (same code in every shim, never reached by a GPU-processor test before). The row tracking question is answered too, and it was a real one: test_delta_merge_preserves_row_tracking_db173, the merge counterpart of the UPDATE and DELETE ones, showed the command committing with the preserved-row-tracking tag while every rewritten row came back with a fresh id. 2a1669b fixes it the way the GPU UPDATE does: the two materialized columns ride through the join (copied rows keep both, updated rows keep the id and reset the version, inserts get nulls), with the field metadata the Delta writer keys on. I first read the OSS 3.3 and 4.x commands as having the same gap and opened #15906 for it, but they inherit OSS Delta's ClassicMergeExecutor.writeAllChanges, which does preserve the columns, so I have re-scoped that issue to the test the OSS shims are missing. Those are now #15907, run on Spark 3.5.6 / Delta 3.3.0, 4.0.0 / Delta 4.0.1 and 4.1.1 / Delta 4.1.0 on a T4 box (three passed each); kept separate because they are OSS-only and this PR is the Databricks shim.
…databricks] The de-duplication of non-effective duplicate matches attached three helper columns under fixed names: the target and source row ids and the row_number rank. Dataset.withColumn replaces an existing column of the same name, so a source or target column called _duplicate_match_rank_, _target_row_id_ or _source_row_id_ was replaced by the helper and then dropped with it, and the clause expressions that referenced the user column failed to bind in the GPU merge processor. The Databricks CPU command accepts such columns. Pick each helper name so that no column of the joined sides resolves to it under the session's resolver, and drop the helpers by those generated names. The fixed names stay the first choice, so plans without a clash are unchanged. Add test_delta_merge_duplicate_source_rows_helper_column_names_db173: a source with columns named _duplicate_match_rank_ and _source_row_id_ and a target with _target_row_id_, two source rows matching one target row with one WHEN MATCHED condition satisfied, the clause actions reading those columns; the table must equal the CPU's and the GPU processor must be in the plan. Run this test, the duplicate-match accepted cases and the NULL not-matched-by-source condition test with change data feed on as well, so the change rows produced through the de-duplication window are compared to the CPU's too. Add test_delta_merge_not_matched_by_source_schema_evolution_db173: MERGE WITH SCHEMA EVOLUTION adds a source column while a conditional NOT MATCHED BY SOURCE clause updates target-only rows; the updated and the copied target-only rows get NULL in the new column. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
…m/jtwynne/cudf-spark into db173-merge-not-matched-by-source
The sixth commit generated the names of the three helpers the duplicate-match de-duplication attaches. The older helpers of the Databricks 17.3 GPU merge command had the same shape: the row id and file name attached to the target in findTouchedFiles, the presence flag attached to each side of the join in writeAllChanges, and the row ids of the CDC delete de-duplication. Dataset.withColumn replaced a user column of the same name, and a name present on both sides was ambiguous after the join. The two control columns are different: the processors locate _row_dropped_ by name in the output, and the command dropped both control columns by name, so a user column named _row_dropped_ or _incr_row_count_ was either read as the control column or dropped from the written data. Generate every attached helper name with uniqueColumnName. Pass the position of the row-dropped control column to the processors as the command knows it (right after the target output columns, with or without CDC), with an optional field on RapidsProcessDeltaMergeJoin that defaults to the previous by-name lookup for the other shims. Drop the control columns by attribute, only when CDC put them in the output. The CDC delete de-duplication with an insert clause also aliased an untyped null literal for the row id that the other side of the join does not have. The insert output carries that row id as a long, so the GPU processor could not concatenate the two outputs. The literal is typed as a long now. Add test_delta_merge_internal_column_names_db173 (with and without CDF): source and target columns named _row_id_, _file_name_, _row_dropped_ and _incr_row_count_, read by the update, insert and not-matched-by-source actions and the not-matched-by-source condition; the expected rows are spelled out and the GPU processor must be in the plan. The Databricks CPU command rejects user columns named like the presence flags, so a parity test cannot use those two names. Add test_delta_merge_delete_only_duplicate_cdc_internal_column_names_db173 for the CDC delete de-duplication with user columns named _target_row_id_ and _source_row_id_. Add test_delta_merge_preserves_row_tracking_db173, the merge counterpart of the UPDATE and DELETE row tracking tests: the row ids of the rows that existed before the merge survive on both engines, the commit version moves only for the updated rows, and the inserted row gets a fresh id. The ids are checked per row because the join-based merge lays out files differently from the CPU. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
The command committed with the preserved-row-tracking tag but did not carry the target's materialized row id and commit version columns through the join, so every rewritten row came back with a fresh row id and the current commit version while the commit claimed the ids were preserved. The Databricks CPU merge keeps the ids of copied and updated rows. Attach the two materialized columns to the target side with the same helper the GPU UPDATE command uses and treat them like the CPU does: copied rows keep both, updated rows (matched and not matched by source) keep the id and reset the commit version so the writer assigns the new one, inserted rows get both null. The processors rebuild their output attributes from the output schema, and the Delta writer recognises the two columns by the field metadata the helper put on them, so the schema carries that metadata. Without row tracking the helper adds nothing and the plan is unchanged. test_delta_merge_preserves_row_tracking_db173, added by the previous commit as expected to fail, now passes: the row ids of the rows that existed before survive on both engines, the commit version moves only for the updated rows, and the inserted row gets a fresh id. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
… skip it on Databricks test_delta_merge_preserves_row_tracking drove the CPU and GPU sessions by hand with its own plan capture. It now runs through assert_gpu_and_cpu_writes_are_equal_collect, the path the UPDATE and DELETE tests already take: for a Delta test the helper runs the GPU side under assert_rapids_delta_write, which asserts the GPU Delta write that only the GPU merge command produces, then reads both tables back on the CPU with the row tracking columns and compares them engine to engine (the inserted row's id masked, since the file layout decides it). The merge's result row and the per-engine row id and commit version checks stay. The capability predicate includes Databricks 17.3, but the 17.3 GPU merge on main regenerates row ids (NVIDIA#15884 fixes that and carries the 17.3 test), so the MERGE test is skipped on Databricks with that reason. The UPDATE and DELETE tests keep running on 17.3. Signed-off-by: Thomas Wynne <jtwynne3@gmail.com>
4 participants
Contributes to #8415. This covers the Databricks 17.3 shim (the 14.3 shim has the same command shape and can follow in a separate PR if needed).
Description
On Databricks 17.3 a MERGE with a WHEN NOT MATCHED BY SOURCE clause falls back to the CPU with "notMatchedBySourceClauses not supported on GPU". The fallback costs more than the clause: the 17.3 CPU merge is the row-index-set algorithm, whose scan (file-in-scan id and row-index metadata columns) and rewrite phases the plugin cannot accelerate, so the whole merge runs on the CPU with only the shuffles on the GPU. I hit this on an SCD type 2 pipeline where every dimension merge carries the clause for delete detection.
After this change a MERGE with the clause runs as
GpuMergeIntoCommandon Databricks 17.3 whenspark.rapids.sql.command.MergeIntoCommand/MergeIntoCommandEdgeare enabled, the same way merges without the clause already do. No new configuration. The deletion vector veto is unchanged, so tables with persistent deletion vectors still fall back.How it is done:
GpuRapidsProcessDeltaMergeJoinExec) now evaluates the not-matched-by-source conditions on target rows that have no source match.processProjectionSeriesalready implements "first matching clause wins, otherwise the default", so the target-only rows go through it with the no-op copy as the default instead of being copied unconditionally. The tworequire(...isEmpty)guards become a length check, and the all-empty batch case is handled since the series can now produce nothing.GpuMergeIntoCommandfollows the shape the delta-23x and delta-24x commands in this repo already have (they carry the full implementation but stay vetoed because of the shared processor): findTouchedFiles uses a right outer join and no target-only data skipping when the clauses are present; writeAllChanges builds conditions and outputs for the new clause type, including the CDC pre and post images, throughupdateOutput,deleteOutput,insertOutputand aclauseOutputdispatch; the insert-only shortcut is disabled when the clauses are present and the matched-only right outer join rule is keyed on "no insert clauses" as in OSS Delta; the row-based fallback processor handles the clauses the same way; the commit records the not-matched-by-source predicates; and the per-clause-type metrics the CPU command reports (numTargetRowsMatchedUpdated,numTargetRowsNotMatchedBySourceUpdated,numTargetRowsMatchedDeleted,numTargetRowsNotMatchedBySourceDeleted) are added, incremented through the sameAnd(metric, metric)pattern delta-24x uses.MergeIntoCommandMetaShimis removed.test_delta_merge_check_overflow_in_table_write_errorreports six leaked device column vectors.processProjectionSeriesguarded the batch thatsplitBatchAndClosehad already consumed and left the not-matched remainder open when a clause projection throws. It is pre-existing and independent of the first commit, but it is in the function the port relies on, so it is fixed here: the guard now covers the remainder. With the fix the test runs with no leak reports.numTargetRowsMatchedDeleted, as OSS Delta and delta-24x do for both counters.row_numberwindow over target and source row ids, preferring the pair that takes the action), so source rows that matched on ON alone are neither inserted nor applied and a row with no applying pair is copied once. The window only runs for merges that have such rows. The intended behaviour was checked against DBR 17.3 CPU with and without low-shuffle merge before writing the tests.splitBatchAndClosefiltered each batch with the condition and with its negation, and a NULL passes neither cuDF filter, so such rows were silently dropped: aWHEN MATCHED AND s.flagwith a NULL flag deleted the target row, and a NULL NOT MATCHED BY SOURCE condition dropped the target-only row. NULLs in the evaluated condition are now replaced with false before the split (masks without NULLs are used as they are), so the row falls through to the next clause or the default like the CPU row processor. This is in the shared processor and applies to every shim that uses it.Dataset.withColumnreplaces a same-named user column, so a source or target column called_duplicate_match_rank_,_target_row_id_or_source_row_id_was replaced and dropped and the clause expressions failed to bind. Each helper name is now generated to be absent from the joined sides under the session's resolver and dropped by that generated name; the fixed names stay the first candidate. The same commit runs the duplicate-match, helper-name and NULL not-matched-by-source tests with change data feed on as well, so the change rows written through the de-duplication window are compared to the CPU's too._row_dropped_by name in the output and the command dropped both control columns by name, so a user column with either name was read as the control column or dropped from the written data. The command now passes the control column's position (right after the target output columns, with or without CDC) through an optional field onRapidsProcessDeltaMergeJointhat defaults to the previous by-name lookup for the other shims, and drops the control columns by attribute, only when CDC put them in the output. The new CDC de-duplication test also found that the delete de-duplication with an insert clause aliased an untyped null literal for the row id the other side lacks, while the insert output carries it as a long, so the GPU processor could not concatenate the two outputs (pre-existing, same code in every shim); the literal is typed now. A merge row tracking test, the counterpart of the UPDATE and DELETE ones, is added in the same commit, marked as expected to fail because of the next item.Known limit, same as OSS Delta: with the clause present, findTouchedFiles records every target file, so the join-based merge rewrites the whole table. For a full-refresh source against a deep history table that is much more data than the row-index merge rewrites, so this does not replace a GPU low-shuffle merge for 17.3 (#11079); it makes the command GPU-owned so such a merge has somewhere to plug in.
Follow-ups not in this PR:
delta-spark350db143can take the same change; the delta-23x and delta-24x metas could drop their veto now that the shared processor accepts the clauses (delta-23x wraps the wrong lists for the fallback check atGpuMergeIntoCommand.scala:998-999, which would need fixing first).Other shims that use the shared processor are unaffected by the clause change: their commands still pass empty clause lists, and an empty series takes the default projection, which is the previous no-op copy. They do pick up the NULL-condition fix of the fifth commit.
Areas worth a close look:
targetOutputColsbefore the CDC de-duplication row ids, so the control-column position still counts them; the update expressions get(row id, null)from the Databricks helper and the insert expressions get two nulls, the copy and delete outputs usetargetOutputColsas they are. The GPU scan still vetoes the row-tracking metadata fields, so on a row-tracked table the target read runs on the CPU and the rest of the merge on the GPU, as for UPDATE.rowDroppedColumnIndexis the number of target output columns (row-id helpers included) at the point the schema is built, which is where every clause output appendsROW_DROPPED_COL, with or without CDC. The shared processor uses it only when set; the other shims pass nothing and keep the by-name lookup.splitBatchAndClose, so it also applies to the two presence predicates, which are never NULL, and to every matched and not-matched condition on every shim.And(incrTotal, incrByType); both sides are the metric UDF, soGpuAnd's side-effect path evaluates the right side on the rows where the left side is true, which is all of them.updateOutput,deleteOutput,insertOutput,clauseOutput) so matched and not-matched-by-source clauses share the update and delete expression builders; the matched and insert outputs are unchanged apart from the metric expression.Tests:
test_delta_merge_not_matched_by_source(with and without CDF) now also runs on Databricks 17.3.test_delta_merge_not_matched_by_source_db173_fallbackis replaced bytest_delta_merge_not_matched_by_source_db173, which exercises every row path (matched update, insert, not-matched-by-source update, untouched target rows) with the CPU expression bridge disabled and assertsGpuRapidsProcessDeltaMergeJoinExecis in the captured plan, so it fails if the command falls back.test_delta_merge_not_matched_by_source_null_safe_keysadds a composite null-safe key with a boolean flag, the SCD type 2 shape.test_delta_merge_not_matched_by_source_fallbackis skipped on 17.3 like on OSS Delta 4.1.test_delta_merge_delete_only_duplicate_source_metrics_db173merges three source rows per key with an unconditional MATCHED DELETE and checks thatnumTargetRowsDeletedandnumTargetRowsMatchedDeletedin the commit's operationMetrics both equal the rows deleted, on the CPU and the GPU (added after review: the matched-delete counter was not compensated for duplicate matches).test_delta_merge_duplicate_source_rows_matched_conditions_db173(added after review): six merges in which several source rows match a target row on ON but at most one satisfies a WHEN MATCHED condition (with and without NOT MATCHED BY SOURCE, with no WHEN MATCHED clause, with a conditional delete, with three matches), each with and without change data feed; the table (and the change rows) and the per-clause row counters must equal the CPU's, and the GPU processor must be in the plan.test_delta_merge_duplicate_source_rows_ambiguous_error_db173(added after review): three merges both engines must reject withDELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE(two applying rows, each row taking a different clause, a target-only condition satisfied twice).test_delta_merge_nullable_matched_conditions(added after review, all Delta versions): matched update and delete clauses and an insert clause whose conditions are NULL for some rows; the expected rows are spelled out and the CPU and GPU tables must both equal them.test_delta_merge_duplicate_source_rows_helper_column_names_db173(added after review): the reviewer's reproduction, a source with columns named_duplicate_match_rank_and_source_row_id_, a target with_target_row_id_, two source rows matching one target row with one WHEN MATCHED condition satisfied and the clause actions reading those columns, with and without change data feed; table equal to the CPU's, GPU processor in the plan.test_delta_merge_internal_column_names_db173(seventh commit, with and without CDF): source and target columns named_row_id_and_file_name_, read by the update and insert actions and by the not-matched-by-source condition; the expected rows are spelled out and the GPU processor is asserted in the plan. The presence-flag and control-column names are generated the same way but the Databricks CPU command rejects user columns with those names (missing or ambiguous attribute), so a parity test cannot use them.test_delta_merge_control_column_names_gpu_db173(seventh commit, with and without CDF, GPU only for the reason above): user columns named_row_dropped_and_incr_row_count_on both sides, read by every clause type; spelled-out rows, the GPU processor in the plan, and with CDF the change-row histogram (two pre-images, two post-images, one insert).test_delta_merge_delete_only_duplicate_cdc_internal_column_names_db173(seventh commit): an unconditional MATCHED DELETE with duplicate source matches, an insert clause and CDF on, with user columns named_target_row_id_and_source_row_id_read by the insert action; this is the test that reached the untyped null literal.test_delta_merge_preserves_row_tracking_db173(seventh commit): the merge counterpart oftest_delta_update_preserves_row_tracking_db173and the DELETE one. Every clause type touches a row-tracked target; the row ids of the rows that existed before survive on both engines, the commit version moves only for the updated rows, and the inserted row gets a fresh id. The ids are checked per row because the join-based merge lays out files differently from the CPU, which the log comparison of the UPDATE and DELETE tests does not tolerate.test_delta_merge_not_matched_by_source_schema_evolution_db173:MERGE WITH SCHEMA EVOLUTIONadding a source column, with a conditional NOT MATCHED BY SOURCE update; the updated and the copied target-only rows get NULL in the new column, the expected rows are spelled out, and the GPU processor is asserted in the plan.test_delta_merge_nullable_not_matched_by_source_condition(added after review, OSS Delta 4.1 and Databricks 17.3+): the same plus a NOT MATCHED BY SOURCE condition that is NULL for target rows with a NULL column, with and without change data feed; GPU processor asserted in the plan. Without the fix the first test loses the four matched rows with a NULL flag and the second additionally the two target-only rows with a NULL column.Validated on a Databricks 17.3 LTS ML cluster (Standard_NC16as_T4_v3, single node), built with
jenkins/databricks/build.sh:-k not_matched_by_source: 5 passed, 3 skipped (the two fallback parametrizations and the 14.3-only union test).delta_lake_merge_test.py: 93 passed, 0 failed, 4 skipped, 78 xfailed (the deletion vector parametrizations under [FEA] Need Deletion Vector read support for Databricks 17.3 #12042), 4 xpassed (test_delta_merge_disabled_fallbackwith DV on and a conf that already forces the CPU path, pre-existing).WAS LEAKEDlines.WAS LEAKEDlines; wholedelta_lake_merge_test.pyagain: 103 passed, 0 failed, same skipped and xfailed set as before.WAS LEAKEDlines.WAS LEAKEDlines.WAS LEAKEDlines.WAS LEAKEDlines.The Databricks build only runs on a Databricks cluster, so I could not run the plugin's scala unit tests for the 17.3 shim locally; happy to rerun anything the Databricks CI flags.
Performance Testing
Production-shaped merge from my SCD type 2 reproduction on public TPC-DS data (null-safe key, wide rows, materialized source, delete-detect clause), one job cluster per leg at the same seed.
Environment
spark.rapids.sql.enableddifferResults
GpuMergeIntoCommandAll three legs rewrote the same 2,987,856 unchanged rows, so the comparison is on equal work. The driver log of the first leg shows
MergeIntoCommandEdgeandRapidsProcessDeltaMergeJoinExecwill run on GPU and no fallback.The merged table from the GPU command is identical to the CPU command's table on the same plugin-on cluster: same row count, same current, closed and flagged counts, same xor of xxhash64 over key, rowhash and flags. Against the plugin-off control the row set is the same and 117 of 2.9 M rows differ only in generated double attributes, which the GPU rounds differently at data generation time before the merge runs.
Checklists
Documentation
Testing
(Please provide the names of the existing tests in the PR description.)
Performance