[Feature] Add asynchronous bucket refresh mechanism for dynamic bucket mode #6389

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)