Skip to content

[spark] Fix bucketed scan reporting for unsupported timestamp precision - #9216

Open
zhuxiangyi wants to merge 1 commit into
apache:masterfrom
zhuxiangyi:spark/fix-spj-timestamp-precision
Open

[spark] Fix bucketed scan reporting for unsupported timestamp precision#9216
zhuxiangyi wants to merge 1 commit into
apache:masterfrom
zhuxiangyi:spark/fix-spj-timestamp-precision

Conversation

@zhuxiangyi

Copy link
Copy Markdown
Contributor

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 of
TIMESTAMP(6) are indistinguishable in the transform reported through
SupportsReportPartitioning: the bound bucket function derives its canonical name from
Spark types, and TransformExpression.isSameFunction compares exactly that canonical
name 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 stored
compactly when the precision is <= 3 (Timestamp.isCompact, AbstractBinaryWriter#writeTimestamp)

  • so the same value hashes differently and ends up in a different bucket.

Joining such tables with spark.sql.sources.v2.bucketing.enabled=true therefore lets
Spark 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 DDL
TIMESTAMP is precision 6.

The fix skips reporting the bucket transform for these tables, using the same
BucketFunction.supportsTable check the write side already performs in
PaimonSparkWriter. The asymmetry between the two paths - the write side guarded, the
read 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 are
now excluded as well. Removing the restriction entirely requires BucketFunction to see
the real Paimon type instead of reconstructing it from Spark's StructType, which is the
existing todo above supportsType:

todo: find a way get the correct paimon type in BucketFunction, then remove this checker

Tests

Added BucketedTableQueryTest."Query on a bucketed table - join - bucket key of an unsupported timestamp precision". Spark DDL cannot express a parameterized timestamp, so
the two tables are created through the table API.

  • Without the fix the test fails with Correct Answer - 8 vs Spark Answer - 4, and the
    physical plan contains no Exchange at all.
  • With the fix the shuffle is kept and the join returns all 8 rows.

Full paimon-spark-ut run: 806 passed, 11 failed. The 11 failures are all
LuminaVectorIndexTest and reproduce identically on an unmodified checkout - they come
from 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.

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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant