tharvey5 opened a new pull request, #26715:
URL: https://github.com/apache/flink/pull/26715
## What is the purpose of the change
Currently, Apache Flink's does not support storage partition join, which can
lead to unnecessary data shuffles in batch mode. This PR implements a basic
version to support that.
## Brief change log
This pull request introduces a new optimizer configuration option
table.optimizer.storage-partition-join-enabled and query planer changes to
detect when both sides of a join are partitioned by the join keys and
compatible, allowing it to apply a storage partition join strategy. This avoids
unnecessary shuffles by leveraging the source's partitioning.
Key changes include:
Addition of the SupportsPartitioning interface for table sources to expose
partitioning information.
Implementation of KeyGroupedPartitioning to represent partitioning schemes.
Integration of partitioning awareness in the batch physical sort-merge join
rule to conditionally use the storage partition join when enabled and
applicable.
[for Testing] Serialization and deserialization utilities for partitioning
metadata.
[for Testing] Extension of the test values table factory to support
partitioning.
Comprehensive unit and integration tests verifying the new join strategy and
its configuration.
This enhancement is currently applicable only in batch mode and requires the
source tables to be partitioned by the join keys.
## Verifying this change
* Added test util for serialization and deserialization of partitioning
metadata, so we can create a test table with a KeyGroupPartition.
* Added integration tests (TestStoragePartitionJoin) that verify the
optimizer plan changes when the storage partition join is enabled or disabled.
* Verified that existing tests pass and that the new join strategy is
correctly applied only when the configuration is enabled and partitioning is
compatible.
* Manual verification of execution plans to confirm the absence of
unnecessary shuffles when storage partition join is enabled.
* Verified unit test in table-planner module result is the same as before
the change:
`[ERROR] Tests run: 8671, Failures: 4, Errors: 0, Skipped: 1 `
We tested that before the change test failures is also 4:
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: (**yes** / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no /
don't know)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't
know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs
/ not documented)
--
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]
