[Fix][Connectors-v2] fix dynamic bucket for paimon sink #9595
Conversation
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
Hisoka-X
reviewed
Jul 21, 2025
...r-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSink.java
Outdated
Show resolved
Hide resolved
Contributor
There was a problem hiding this comment.
Pull Request Overview
This PR refactors Paimon's dynamic bucketing mechanism to address potential data duplication issues that may arise after modifying the parallelism level. The fix implements a new channel computer and bucket assigner factory to ensure consistent data distribution across different parallelism configurations.
- Introduces new bucket assignment logic with
RowAssignerChannelComputerandPaimonBucketAssignerFactory - Adds comprehensive test cases for dynamic bucket functionality with different parallelism settings
- Replaces the previous bucket assignment approach to prevent data duplication when parallelism changes
Reviewed Changes
Copilot reviewed 19 out of 21 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| PaimonSinkWriter.java | Implements new dynamic bucket assignment logic using factory pattern and synchronized access |
| RowAssignerChannelComputer.java | New channel computer for consistent row distribution across parallel tasks |
| PaimonBucketAssignerFactory.java | Factory class to manage bucket assigners per table and task |
| PaimonBucketAssigner.java | Refactored to use HashBucketAssigner instead of SimpleHashBucketAssigner |
| Test configuration files | Multiple new test configurations for different parallelism scenarios |
| PaimonSinkDynamicBucketIT.java | Enhanced test suite with MySQL integration and parallelism validation |
...on/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
Outdated
Show resolved
Hide resolved
...n-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonSinkDynamicBucketIT.java
Outdated
Show resolved
Hide resolved
| // When multiple threads call assigner.assign() simultaneously, they can | ||
| // corrupt the internal hash map structure, leading to the | ||
| // ArrayIndexOutOfBoundsException during rehashing operations | ||
| synchronized (bucketAssigner) { |
There was a problem hiding this comment.
Synchronizing on the bucketAssigner instance may cause performance bottlenecks. Consider using a more granular locking mechanism or concurrent data structures to reduce contention.
...rg/apache/seatunnel/connectors/seatunnel/paimon/sink/bucket/PaimonBucketAssignerFactory.java
Outdated
Show resolved
Hide resolved
hawk9821
force-pushed
the
paimon_dynamic_bucket
branch
2 times, most recently
from
80962c7 to
7b55968
Compare
hawk9821
force-pushed
the
paimon_dynamic_bucket
branch
from
7b55968 to
584d704
Compare
hawk9821
force-pushed
the
paimon_dynamic_bucket
branch
from
b8e5eec to
ef0d544
Compare
Hisoka-X
reviewed
Jul 28, 2025
...rg/apache/seatunnel/connectors/seatunnel/paimon/sink/bucket/PaimonBucketAssignerFactory.java
Outdated
Show resolved
Hide resolved
Hisoka-X
reviewed
Jul 28, 2025
…che/seatunnel/connectors/seatunnel/paimon/sink/bucket/PaimonBucketAssignerFactory.java Co-authored-by: Jia Fan <[email protected]>
Hisoka-X
approved these changes
Jul 31, 2025
corgy-w
approved these changes
Jul 31, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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 of this pull request
Refactor Paimon's dynamic bucketing mechanism to address potential data duplication issues that may arise after modifying the parallelism level.
Does this PR introduce any user-facing change?
How was this patch tested?
PaimonSinkDynamicBucketIT#testWriteForDifferentParallelism
Check list
New License Guide