[FEATURE] Commit-based Clustering Plan Strategy
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
## Summary
Introduce a new clustering plan strategy that processes data files based on their commit/ingestion order, enabling accurate latency tracking and preventing re-clustering of already stitched files.
---
## Motivation
Currently, clustering operations in Hudi are planned based on partitions, aggregating all files that meet predefined criteria (e.g., small file size). However, the execution order of collected files does not align with their ingestion order—meaning data files produced later may be clustered before those produced earlier.
**Limitations of the current approach:**
1. **Partition-based planning is optimized for backfill**, not streaming ingestion
2. **No ingestion order guarantee** — files are clustered without regard to when they were written
3. **Inaccurate latency calculations** — clustering latency metrics cannot be computed correctly when files are processed out of order
4. **Re-listing of already stitched files** — previously clustered files may be re-evaluated unnecessarily
**Streaming ingestion requirements:**
For streaming workloads, it is critical that:
- Newly generated small files are merged promptly to avoid performance degradation
- Files produced earlier are merged first to ensure accurate latency measurement
- Already-clustered files are excluded from future clustering plans
---
## Proposed Solution
Introduce a new `CommitBasedClusteringPlanStrategy` that:
1. **Clusters data files according to their ingestion order**
2. **Prevents previously stitched files from being re-listed**
3. **Guarantees correctness for clustering latency calculation**
---
## Architecture
### Checkpoint Management
The commit-based plan requires a **checkpoint** to track the last processed commit instant. This enables subsequent clustering runs to identify which new commits need processing.
**Checkpoint storage:**
- The checkpoint is persisted in the **Table Service Manager** (not in commit files)
- The **Table Service Runner** fetches the checkpoint before starting a clustering job
- Upon successful completion, the Table Service Runner updates the checkpoint
**Why not store checkpoints in commit files (like DeltaStreamer)?**
- Clustering produces **replace commits**, and the last replace commit is not guaranteed to remain in the timeline
- Multiple types of replace commits may coexist (e.g., encryption clustering, column pruning clustering), each potentially using different clustering plans
---
## Implementation Details
The `CommitBasedClusteringPlanStrategy` processes commits in order using the following steps:
1. **Fetch new commits** — Retrieve all commits since the last checkpoint
2. **Collect files by partition** — For each commit:
- Collect eligible files organized by partition
- For replace commits, also collect replaced files to prevent re-clustering of old file versions
3. **Group files for clustering** — Files in the same partition across commits are grouped together unless they exceed the maximum bytes per clustering group. Files that have been replaced by a previous clustering job are filtered out.
4. **Generate clustering plan** — The plan is finalized when:
- Maximum number of clustering groups is reached, OR
- All new commits have been processed
**Atomicity guarantee:** A single commit must be fully processed or not at all. All files meeting the small file criteria within a commit are planned for clustering, even if this exceeds the maximum clustering data size configured.
---
## Example Scenario
Consider a timeline with:
- Commit C1 (partition P1: file1, file2; partition P2: file3)
- Commit C2 (partition P1: file4)
- Replace Commit R1 (replaces file1, file2 with file5)
**Clustering plan generation:**
- Files from P1 across C1 and C2 are grouped: `[file4]` (file1, file2 excluded as they were replaced)
- Files from P2: `[file3]`
- Result: Two clustering groups formed, excluding replaced files
---
## Configuration
| Property | Description |
|----------|-------------|
| `hoodie.clustering.plan.strategy.class` | Set to `CommitBasedClusteringPlanStrategy` |
| `hoodie.clustering.plan.strategy.max.bytes.per.group` | Maximum bytes per clustering group |
| `hoodie.clustering.plan.strategy.max.num.groups` | Maximum number of clustering groups per plan |
---
## Benefits
1. **Streaming-friendly** — Ensures timely merging of small files in streaming pipelines
2. **Ordered processing** — Files are clustered in ingestion order
3. **Accurate metrics** — Enables correct clustering latency calculation
4. **Efficiency** — Avoids re-listing and re-processing of already clustered files
5. **Checkpoint-based resumption** — Enables incremental clustering across runs
---
## Compatibility
- Backward compatible with existing partition-based clustering strategies
- Works alongside other clustering plan strategies (encryption, column pruning)
- Integrates with existing Table Service infrastructure
---
## Tasks
- [ ] Implement `CommitBasedClusteringPlanStrategy` class
- [ ] Add checkpoint persistence in Table Service Manager
- [ ] Update Table Service Runner to fetch/update checkpoints
- [ ] Handle replace commit file exclusion logic
- [ ] Add configuration properties
- [ ] Unit tests for commit-based planning
- [ ] Integration tests with streaming ingestion
- [ ] Documentation updates
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reading the existing partition-based clustering strategies and the Table Service Manager and Table Service Runner entry points. Trace how plans, checkpoints, replace commits, and clustering groups are currently handled, then review the proposed unit and integration test scenarios. Done means the new strategy, checkpoint flow, replace-commit exclusion, configuration, tests, and documentation are covered.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, stream-processing
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100