[Feature] Support Partition Mark Done for Fluss Tiering (Paimon Lake)
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
For partitioned tables with Paimon lake tiering enabled, downstream batch jobs need a signal that a partition's data is "ready" (i.e., fully tiered). This is the **Mark Done** mechanism — when a partition has been idle (no new data tiered) for a configurable duration, Fluss should execute Paimon's mark-done actions (e.g., write `_SUCCESS` file, notify metastore) so downstream schedulers can safely begin processing.
Currently, Fluss tiering operates at the table level with no partition-level idle tracking or mark-done capability. Paimon already has a complete mark-done action framework (`PartitionMarkDoneAction`), but its trigger mechanism is tightly coupled to Flink checkpoint lifecycle, which doesn't fit Fluss's tiering model.
### Solution
**Summary.** Carry table-level tiering state as a single optional `tiering_state` field in the offset file, transported as one `bytes tiering_state_json` in RPC — following Fluss's existing `schema_json` / `table_json` pattern, so adding new state never changes proto/RPC. Partition mark-done is the first consumer, using a **delete-on-done** model that keeps the state tiny and self-GCing.
#### Storage & RPC
- The offset file (`version=1`) gains an optional top-level `tiering_state` (JSON object) next to `bucket_offsets`. Its content is owned by the lake side (`LakeTieringTableState` in `fluss-common`); the offset serde passes it through **without parsing**.
- RPC: `PbTableOffsets` (in `PrepareLakeTableSnapshotRequest`) and `GetLakeSnapshotResponse` each get one new `optional bytes tiering_state_json` (new field number → wire-compatible). The server moves it through as an opaque blob (persist on PREPARE, read back on GET); the client parses it via `LakeSnapshot.getLakeTieringTableState()`.
- Because it is a single `bytes` field, future state fields need no further proto/RPC change.
#### Partition mark-done (delete-on-done)
- **State**: a table-level `partition_done_initialized` flag + `partition_update_times` of only the not-yet-done partitions; no `doneTime` (read Paimon `_SUCCESS` if needed). Once marked done, a partition is removed.
```json
{
"version": 1,
"table_id": 8,
"partition_offsets": [ { "partition_id": 5, "bucket_offsets": [100, 230] } ],
"tiering_state": {
"version": 1,
"partition_done_initialized": true,
"partition_update_times": { "5": 1704153550000 }
}
}
```
- **Per round**: the job reads the previous full state, updates `partitionUpdateTime`, judges, marks done, drops done partitions, and uploads the new full state; the coordinator overwrites it (tiny state → no merge).
- **Judgement (reuses Paimon)**: ready when `now - max(partitionUpdateTime, partitionEndTime) > idle`, then call Paimon `PartitionMarkDoneAction` (writes `_SUCCESS` / metastore). `partitionEndTime` is derived from the partition value at judge time (not stored). Actively-written partitions never go idle; a done (dropped) partition that receives new data re-enters and is judged again.
- **Cold start**: the `partition_done_initialized` flag handles partitions that stopped being written before the feature was enabled — the first run lists all partitions, back-fills done by `partitionEndTime`, then sets `partition_done_initialized = true`. The flag is required because, under delete-on-done, an empty state (all done) is indistinguishable from "first run".
#### Compatibility & future work
- New proto fields are optional with new numbers → wire-compatible. The `tiering_state` payload carries its own internal `version` (`LakeTieringTableState`'s JSON serde) to support structured evolution; adding a state field later only touches that serde, and unknown fields are preserved (pass-through) for rolling upgrades.
- Future (not in this iteration): an Admin API to reset `partition_done_initialized` (rewrite/delete the offset file) triggers a full-table re-judgement via the cold-start path.
> Config: enabled via `paimon.*` table properties; idle threshold `partition.idle-time-to-done`.
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.