[Feature] Add asynchronous bucket refresh mechanism for dynamic bucket mode
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
## Problem
In dynamic bucket mode, when buckets fill up and reach their target capacity, the assigner cannot detect that previously full buckets may have become available again
after compaction or data deletion.
This causes the system to continuously create new buckets even when existing buckets have free space, leading to:
- Unnecessary bucket growth
- Increased metadata overhead
- Suboptimal resource utilization
## Root Cause
The `PartitionIndex` maintains an in-memory map (`nonFullBucketInformation`) of buckets with available space. However:
1. When a bucket fills up, it's removed from this map
2. Compaction or deletions may free space in these "full" buckets
3. The assigner has no mechanism to rediscover these freed buckets
4. Result: System keeps creating new buckets indefinitely
## Proposed Solution
Add an **asynchronous bucket refresh mechanism** that:
### Trigger Conditions
- Refresh triggers when a bucket's row count reaches `(target-row-num - threshold)`
- Example: With `target-row-num=1,000,000` and `threshold=100,000`, refresh triggers at 900,000 rows
- This provides early warning before running out of available buckets
### Architecture
- **Async Execution**: Uses `CompletableFuture` with dedicated thread pool
- **Thread Pool**: Unbounded queue with 4 core / 12 max threads (I/O bound operations)
- **Per-Partition**: Each partition refreshes independently (handles data skew)
- **Timeout**: 5-minute timeout per refresh operation
- **Thread-Safe**: Uses `putIfAbsent()` to avoid race conditions
### Memory Efficiency
- Queue is unbounded but safe: ~400 bytes per task
- With 500 partitions: max ~200KB memory footprint
- Tasks are infrequent (24h default interval) and process quickly (30-60s)
### Configuration Options
**`dynamic-bucket.empty-bucket-threshold`** (default: `-1`)
- Threshold in rows before triggering refresh
- When bucket reaches `(target-row-num - threshold)`, trigger refresh
- Set to `-1` to disable feature (backward compatible)
- Recommended: 10-20% of `dynamic-bucket.target-row-num`
**`dynamic-bucket.min-refresh-interval`** (default: `24 hours`)
- Minimum time between refresh operations per partition
- Prevents excessive I/O overhead from frequent disk scans
- Longer intervals recommended for systems with many partitions (500+)
## Implementation Details
### Modified Files
- `PartitionIndex.java`: Core refresh logic with async executor
- `HashBucketAssigner.java`: Pass configuration parameters
- `CoreOptions.java`: Add two new configuration options
- `BucketProcessor.scala`: Spark integration
- `HashBucketAssignerOperator.java`: Flink integration
- `HashBucketAssignerTest.java`: 5 new comprehensive tests
### Key Features
1. **Backward Compatible**: Disabled by default (threshold = -1)
2. **Data Skew Friendly**: Works independently per partition
3. **Production Safe**: Unbounded queue ensures no task rejection
4. **Well Tested**: 15/15 tests pass including 5 new tests
## Testing
New tests cover:
- `testRefreshTriggeredWhenBucketNearFull`: Verify refresh triggers at threshold
- `testRefreshDisabledWhenThresholdIsNegative`: Backward compatibility
- `testRefreshRespectsMinimumInterval`: Interval enforcement
- `testRefreshWithMultiplePartitionsDataSkew`: Data skew handling
- `testRefreshDiscoversFreedBucketsAfterCompaction`: Main use case
All existing tests continue to pass.
## Benefits
- ✅ Reduces unnecessary bucket creation
- ✅ Better resource utilization
- ✅ Handles data skew (per-partition refresh)
- ✅ Low overhead (configurable intervals)
- ✅ Production safe (unbounded queue, no task rejection)
- ✅ Backward compatible (disabled by default)
Contributor guide
No contributing guide indexed for this repository
Research direction
Read PartitionIndex.java first to understand the existing nonFullBucketInformation flow, then trace configuration through CoreOptions.java and HashBucketAssigner.java. Review BucketProcessor.scala and HashBucketAssignerOperator.java for Spark and Flink integration, and run HashBucketAssignerTest.java. Done means the five named refresh tests and all existing tests pass, with refresh disabled by the default threshold.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, scala, spark
- Domain
- data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100