[core] Tag diff query at postpone table should check bucket number

[yuzelin]

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

[JingsongLi]

[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.

[JingsongLi]

[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?

[JingsongLi]

[P1] Continue validating the remaining partitions

When the first intersecting partition has the same actual bucket count, this return exits the entire method, so a later partition whose bucket count changed is never checked. With a bucket = -2 partitioned table, I reproduced this using unchanged common partitions plus one partition rescaled from 1 to 2 buckets between the tags: no InconsistentTagBucketException was thrown, and the Spark EXCEPT fallback would therefore be skipped. Replacing this with continue makes the regression pass. Please also add a multi-partition test covering this ordering.

[JingsongLi]

[P1] Keep the assertion in sync with the new message

This targeted test currently fails because the assertion still expects The real bucket number..., while checkRescaleBucketForIncrementalDiffQuery now throws The bucket number of two snapshots are different (1, 2).... Running mvn -pl paimon-core -Pfast-build -DwildcardSuites=none -Dtest=IncrementalTableTest#testPostponeBucketNumberChangedInIncrementalDiff test fails at this assertion. Please update the expectation, or preserve the previous exception message.