apache / apache/paimon

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

Open
#6,389 5 comments 0 reactions 0 assignees View on GitHub
enhancement
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.