[core] Tag diff query at postpone table should check bucket number - #9048
[core] Tag diff query at postpone table should check bucket number#9048yuzelin wants to merge 5 commits into
Conversation
| TableSchema schema = schemaManager.latest().get(); | ||
| if (!schema.primaryKeys().isEmpty() && schema.numBuckets() == BucketMode.POSTPONE_BUCKET) { | ||
| Map<BinaryRow, Integer> startBucketNumbers = | ||
| PostponeUtils.getKnownNumBuckets(reader, start.id()); |
There was a problem hiding this comment.
[P1] Preserve tag snapshots after snapshot expiration
start and end may come from Tag.trimToSnapshot(), which is what allows tag reads to continue after the original snapshots have been expired. Passing only start.id() here eventually calls reader.withSnapshot(long) and reloads snapshot-N from the main snapshot directory. Once that file is expired, this validation throws before the tag diff can run.
I reproduced this with a bucket = -2 table: create TAG1/TAG2, retain only the latest snapshot, then query the diff between the tags; it fails with snapshot-1 does not exist. Please pass the Snapshot object through getKnownNumBuckets and call reader.withSnapshot(snapshot). It would also be useful to add a regression test that expires the source snapshots while retaining the tags.
| Map<BinaryRow, Integer> startBucketNumbers = | ||
| PostponeUtils.getKnownNumBuckets(reader, start.id()); | ||
| Map<BinaryRow, Integer> endBucketNumbers = | ||
| PostponeUtils.getKnownNumBuckets(reader, end.id()); |
There was a problem hiding this comment.
[P2] Avoid scanning both snapshots twice
These two calls traverse the active manifest entries for the start and end snapshots, and readIncrementalDiff immediately plans both snapshots again. Therefore every postpone-bucket diff whose bucket counts are unchanged performs four manifest traversals instead of two; without a manifest cache this also duplicates the remote manifest reads.
Could the bucket counts be derived from the beforeFiles and afterFiles already collected by readIncrementalDiff, or otherwise be returned from the same planning pass?
Purpose
ncremental diff queries compare data between two tags bucket by bucket. For postpone-bucket tables, the configured bucket value remains -2, while the actual bucket number of a partition may change between the two snapshots. In that case, directly performing the bucket-based diff may produce incorrect results.
So this PR collects the actual bucket number of each partition from both snapshots and check if they are equal. If not, fall back to Spark except query.
Tests
IncrementalTableTest
TableValuedFunctionsTest