[spark] Fix bucketed scan reporting for unsupported timestamp precision - #9216
Open
zhuxiangyi wants to merge 1 commit into
Open
[spark] Fix bucketed scan reporting for unsupported timestamp precision#9216zhuxiangyi wants to merge 1 commit into
zhuxiangyi wants to merge 1 commit into
Conversation
Spark's timestamp precision is fixed to 6, so a bucket key of TIMESTAMP(3) and one of TIMESTAMP(6) are indistinguishable in the transform reported through SupportsReportPartitioning: the bound bucket function derives its canonical name from Spark types, and Spark compares exactly that canonical name to decide whether two scans are co-partitioned. Paimon, however, lays those two out differently in BinaryRow - a timestamp is stored compactly when the precision is <= 3 - so the same value hashes differently and lands in a different bucket. Joining such tables with spark.sql.sources.v2.bucketing.enabled therefore let Spark drop a shuffle that was actually required, silently returning fewer rows. Skip reporting the bucket transform for these tables, which is the same BucketFunction.supportsTable check the write side already performs in PaimonSparkWriter. Tables whose bucket key carries a precision other than 6 now fall back to a shuffle instead of producing wrong results.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
A bucketed scan can report a partitioning that does not match how Paimon actually
assigned the buckets, which makes Spark drop a shuffle it still needs and silently
return fewer rows.
Spark's timestamp precision is fixed to 6, so a bucket key of
TIMESTAMP(3)and one ofTIMESTAMP(6)are indistinguishable in the transform reported throughSupportsReportPartitioning: the bound bucket function derives its canonical name fromSpark types, and
TransformExpression.isSameFunctioncompares exactly that canonicalname to decide whether two scans are co-partitioned. The other two guards do not help
here either, because both tables report the same number of partitions and the same
partition values (the bucket ids
0..n-1).Paimon, however, lays those two out differently in
BinaryRow- a timestamp is storedcompactly when the precision is
<= 3(Timestamp.isCompact,AbstractBinaryWriter#writeTimestamp)Joining such tables with
spark.sql.sources.v2.bucketing.enabled=truetherefore letsSpark pair bucket 0 with bucket 0 across two tables whose bucket 0 holds a different set
of rows. Nothing fails, the result is just incomplete. This is easy to hit in a mixed
stack, since Flink pipelines commonly declare
TIMESTAMP(3)while a Spark DDLTIMESTAMPis precision 6.The fix skips reporting the bucket transform for these tables, using the same
BucketFunction.supportsTablecheck the write side already performs inPaimonSparkWriter. The asymmetry between the two paths - the write side guarded, theread side not - is what let this through.
Trade-off worth noting: this is deliberately conservative. Two tables that are both
TIMESTAMP(3)are encoded identically and could safely skip the shuffle, but they arenow excluded as well. Removing the restriction entirely requires
BucketFunctionto seethe real Paimon type instead of reconstructing it from Spark's
StructType, which is theexisting
todoabovesupportsType:Tests
Added
BucketedTableQueryTest."Query on a bucketed table - join - bucket key of an unsupported timestamp precision". Spark DDL cannot express a parameterized timestamp, sothe two tables are created through the table API.
Correct Answer - 8vsSpark Answer - 4, and thephysical plan contains no
Exchangeat all.Full
paimon-spark-utrun: 806 passed, 11 failed. The 11 failures are allLuminaVectorIndexTestand reproduce identically on an unmodified checkout - they comefrom the Lumina native library not loading in this environment
(
org.aliyun.lumina.LuminaNative.<clinit>), not from this change.API and Format
No public API or format change.
Documentation
No documentation change - this restores the intended behaviour rather than adding a
feature.