zhuxiangyi opened a new pull request, #10055:
URL: https://github.com/apache/paimon/pull/10055
### Purpose
Bug fix: a data-evolution `MERGE INTO ... ON target._ROW_ID = source.col`
fails with `FAILED_EXECUTE_UDF` as soon as the source holds a row id that no
longer exists in the table, instead of treating that source row as not matched.
**Why.** When the merge condition is a plain equality on the target's
`_ROW_ID`, `MergeIntoPaimonDataEvolutionTable` takes a shortcut to find the
touched files: rather than joining target and source, it maps every value of
`source.col` to the file whose row-id range contains it
(`findRelatedFirstRowIds`). That mapping was `floorBinarySearch` over the
sorted live first row ids, a helper written for the target's own `_ROW_ID`
values, which are always inside some file. Applied to source values it has two
problems:
- a value below the smallest live first row id throws
`IllegalArgumentException("Value N is less than the first element in the sorted
sequence")`, which surfaces as `FAILED_EXECUTE_UDF` and aborts the whole
statement;
- a value beyond the last range floors to the last file, which is then
rewritten for nothing.
The first one is easy to hit in practice. Row ids are stable but not
permanent: `INSERT OVERWRITE` of a partition, dropping a partition, or a
compaction that reassigns row ids all remove ranges. A source captured from an
earlier snapshot (`INSERT INTO source SELECT _ROW_ID, ... FROM target`) then
carries ids the table no longer has. Since the oldest data is what gets
overwritten or expired first, the missing ids are typically the smallest ones,
i.e. exactly the throwing case:
```
INSERT INTO target VALUES (1, 10, 'p1'), (2, 20, 'p1'); -- row ids 0, 1
INSERT INTO target VALUES (3, 30, 'p2'), (4, 40, 'p2'); -- row ids 2, 3
INSERT INTO source SELECT _ROW_ID, b + 100 FROM target;
INSERT OVERWRITE target PARTITION (dt = 'p1') VALUES (5, 50), (6, 60); --
row ids 4, 5; 0 and 1 are gone
MERGE INTO target USING source ON target._ROW_ID = source.rid
WHEN MATCHED THEN UPDATE SET b = source.b
WHEN NOT MATCHED THEN INSERT ...
```
```
org.apache.spark.SparkException: [FAILED_EXECUTE_UDF] Failed to execute user
defined function (... (bigint) => bigint)
Caused by: java.lang.IllegalArgumentException: Value 0 is less than the
first element in the sorted sequence.
at
...MergeIntoPaimonDataEvolutionTable$.floorBinarySearch(MergeIntoPaimonDataEvolutionTable.scala:1522)
at
...MergeIntoPaimonDataEvolutionTable.$anonfun$findRelatedFirstRowIds$1(MergeIntoPaimonDataEvolutionTable.scala:1243)
```
The same statement with `ON target._ROW_ID = source.rid + 0`, which does not
qualify for the shortcut and goes through the join, succeeds and treats those
rows as not matched. The shortcut is an optimisation and must not change the
answer.
**What.** Keep the exclusive end of every row-id range next to its first row
id (`rowIdRangeEnds`, aligned with `firstRowIds`; column-group files of one
range share both), and map a source value to a file only when it lies inside
`[firstRowId, end)`. Values outside every range map to no file and are filtered
out before the distinct/collect, so they simply do not contribute a touched
file. `addFirstRowId`, which runs on the target's own `_ROW_ID`, keeps the
strict floor search.
**Benefit.**
- The shortcut now gives the same result as the join path for any source
content: stale ids, ids from another table, negative or out-of-range values all
behave as non-matching rows and fall through to `WHEN NOT MATCHED` if present.
- Values beyond the last range no longer drag the last file into the rewrite.
### Tests
`RowTrackingTestBase` (run as `RowTrackingTest` on Spark 3.5, 67 tests, plus
`BlobUpdateTest` and `DataEvolutionDeletionTest`):
- New `merge into with _ROW_ID shortcut ignores source row ids outside the
table`: builds the source from a snapshot, overwrites the oldest partition,
adds `-1` and `1000` to the source, then merges on `_ROW_ID` with `WHEN
MATCHED` / `WHEN NOT MATCHED`. Asserts that the touched files were found
without a join (the shortcut was taken), that the live rows 2 and 3 are
updated, the overwritten partition is untouched, and the four dangling ids are
inserted through `WHEN NOT MATCHED`. Fails on master with the
`IllegalArgumentException` above.
- Existing `merge into table with data-evolution with _ROW_ID shortcut`
still passes; its source ids `6` and `8` now map to no file instead of the last
one, with the same result.
`spotless:check` and `checkstyle:check` pass on both modules.
### API and Format
No changes.
### Documentation
No changes.
--
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]
