apache / apache/fluss

[Feature] Support Partition Mark Done for Fluss Tiering (Paimon Lake)

Open
#3,314 0 comments 0 reactions 1 assignee Claimed by @beryllw View on GitHub
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.