zhuxiangyi opened a new pull request, #9216:
URL: https://github.com/apache/paimon/pull/9216
### 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.
--
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]
