[kv] Add row TTL support

[platinumhamburg]

Purpose

Linked issue: close #3620

Motivation

Primary key tables currently retain their latest row state indefinitely. Applications that only need recent state must perform external cleanup, which adds operational complexity and cannot take advantage of RocksDB's native compaction lifecycle.

This change introduces row-level TTL as an opt-in table capability while keeping its cleanup semantics explicit and preserving compatibility for clients and stored data.

Solution

Row TTL is configured with table.kv.row.ttl and is disabled when the option is absent. The table.kv.row.ttl namespace explicitly scopes row-level expiration to primary-key table KV state and distinguishes it from log or partition retention. Row TTL uses processing time by default. Applications can instead configure table.kv.row.ttl.time-column with a BIGINT epoch-millisecond or TIMESTAMP_LTZ column; a null event-time value does not expire.

Expired rows are removed asynchronously by RocksDB compaction through Flink's compaction filter. Cleanup is therefore best effort: a row becomes eligible after its TTL but may remain visible until compaction processes it. TTL cleanup does not emit delete records to changelog or binlog consumers.

Fluss owns the versioned KV value layout used by the compaction filter. All value encoding and decoding is centralized in the layout abstraction, allowing processing-time and event-time TTL to share one storage contract and leaving room for future compaction metadata. Recovery preserves the original TTL basis instead of restarting a row's lifetime.

Compatibility is enforced at the raw KV API boundary. Clients that do not understand the TTL-aware value layout are rejected when accessing TTL tables rather than being allowed to misread or overwrite their values.

Changing, disabling, or changing the time column of row TTL through ALTER TABLE is not supported in this version. Exact row-count pushdown is also disabled for TTL tables because compaction removes rows outside the normal write path.

Brief change log

• Add processing-time and event-time row TTL configuration and validation for primary key tables.
• Add a Fluss-owned, versioned KV value layout carrying compaction metadata.
• Integrate best-effort expiration with RocksDB compaction and preserve TTL semantics across recovery and partial updates.
• Add client compatibility guards and prevent exact row-count pushdown for TTL tables.

Tests

• Added coverage for table option validation and KV value layout encoding/decoding.
• Added processing-time and event-time expiration tests, including null event time and partial updates.
• Added recovery tests that verify the TTL basis is preserved.
• Added real KV tablet compaction tests and old-client API compatibility tests.
• Added Flink connector coverage for row-count pushdown behavior.

API and Format

• Adds the table.kv.row.ttl option family, including time-column and changelog-mode.
• Adds KV format version 3 for TTL-enabled primary key tables. The coordinator selects it automatically; users do not select it independently.
• TTL-sensitive raw KV requests require a row-TTL-aware API version when the target table has TTL enabled.

Documentation

• Documents the row TTL option family in the Flink connector option reference.
• Adds row TTL semantics, examples, compatibility constraints, and current limitations to the TTL guide.

[Copilot]

Pull request overview

Adds row-level TTL support for primary-key (KV) tables, introducing a new table option (table.row.ttl) and a KV format v3 value layout that can carry an internal timestamp used for best-effort RocksDB-compaction-based cleanup. The PR also updates client/server RPC compatibility (API versions) and propagates the KV format version through scanners/lookups so clients can decode values correctly.

Changes:

• Introduce table.row.ttl (plus related options) and enforce/derive KV format v3 for TTL-enabled PK tables.
• Implement TTL timestamp tagging + RocksDB compaction-filter based cleanup wiring in the KV tablet path (including recovery), plus client-version gating for TTL tables.
• Update Flink connector option conversion + pushdown logic and extend docs to describe row TTL behavior and limitations.

Reviewed changes

Copilot reviewed 53 out of 53 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
website/docs/table-design/data-distribution/ttl.md	Documents row TTL behavior and configuration for PK tables.
website/docs/engine-flink/options.md	Adds Flink connector option entries for row TTL-related properties and KV format v3 behavior notes.
website/docs/engine-flink/ddl.md	Clarifies ALTER TABLE SET/RESET constraints, including row TTL being non-alterable.
fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java	Adds validation tests for row TTL properties and KV format constraints.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java	Wires a manual clock into KV manager creation for TTL-sensitive tests.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java	Updates limit-scan API usage and adds TTL client-version rejection coverage.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerRowTtlClientVersionTest.java	New focused tests for per-API required client versions on TTL tables.
fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlTimestampProviderTest.java	Tests TTL timestamp selection for write/recovery and event-time handling.
fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlCompactionFilterTest.java	Tests compaction filter behavior and TTL duration validation.
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java	Extends KV tablet tests for value tagging, compaction cleanup visibility, and row-count disablement under TTL.
fluss-server/src/test/java/org/apache/fluss/server/kv/KvRecoverHelperTest.java	Adds recovery tests ensuring v3 tag behavior is based on log timestamps or time-column.
fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java	IT coverage for coordinator defaults (kv format v3) and storing time-column id metadata.
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java	Adds row TTL property validation and KV format version gating logic.
fluss-server/src/main/java/org/apache/fluss/server/utils/RowTtlUtils.java	New utility for validating/ceiling TTL durations to milliseconds with overflow safety.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Adds client-version checks for scan APIs and passes API version into limit-scan.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java	Passes server clock into KV manager creation.
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java	Introduces shared client-version validation helpers requiring table metadata access.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Adds API-key-aware client version validation for row TTL tables and updates limit-scan signature.
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Disables exact row count when TTL enabled and passes table config into recovery helper.
fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlTimestampProvider.java	New provider that supplies the v3 value tag timestamp for processing-time/event-time TTL.
fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlCompactionFilterFactory.java	New factory creating RocksDB compaction filters configured for TTL cleanup via the value tag offset.
fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java	Preserves value metadata when producing merged rows (via BinaryValue.withRow).
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java	Adds a setter to install a compaction filter factory on column family options.
fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java	Preserves value metadata when applying partial updates (via BinaryValue.withRow).
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java	Wires TTL compaction filter, v3 encoding/tagging, and disables exact row count under TTL.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java	Updates recovery path to rebuild v3 values using recovery-time tag providers.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java	Threads a clock through KV tablet creation/loading and passes kv format version/table config.
fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/PerSchemaAutoIncrementUpdater.java	Preserves value metadata when updating auto-increment columns (via BinaryValue.withRow).
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Applies defaults for TTL tables (kv format v3) and stores internal time-column id.
fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java	Adds assertions for new highest-supported API versions required by TTL-aware clients.
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java	Bumps supported versions for snapshot/scan/raw-KV APIs to reflect TTL client compatibility requirements.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java	Adds option-conversion test coverage for table.row.ttl.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java	Adds/adjusts batch planning tests for COUNT pushdown constraints with WAL and row TTL.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java	Fixes Duration option conversion to handle no-default-value Duration options.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java	Prevents COUNT pushdown when exact row count isn’t available (WAL or row TTL).
fluss-common/src/test/java/org/apache/fluss/row/encode/KvValueLayoutTest.java	New tests for v3 layout encoding/decoding and value-tag semantics.
fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java	Introduces version-aware encoding (v2 vs v3) and value-tagged encoding support.
fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java	Decodes raw values using a versioned KvValueLayout, including optional value tags.
fluss-common/src/main/java/org/apache/fluss/row/encode/KvValueLayout.java	New layout abstraction defining schema id/value tag offsets for different kv format versions.
fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java	Uses KvValueLayout to locate row payload when decoding raw values.
fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java	Carries a KvValueLayout so record batches can decode versioned raw value layouts.
fluss-common/src/main/java/org/apache/fluss/record/ValueRecordBatch.java	Extends read-context interface to provide the KvValueLayout.
fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecord.java	Updates record format handling to decode via layout-provided offsets.
fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java	Adds optional value-tag support and preserves metadata across transformations.
fluss-common/src/main/java/org/apache/fluss/metadata/RowTtlChangelogMode.java	New enum defining TTL cleanup changelog behavior (currently NONE).
fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java	Adds row TTL getters and internal table.row.ttl.time-column-id property handling.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds row TTL-related config options and introduces kv format v3 constants/max version.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java	Propagates kv format version into snapshot scanning paths for correct decoding.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java	Decodes snapshot values using a layout derived from kv format version.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java	Uses kv format version layout for decoding limit-scan response batches.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java	Plumbs kv format version through snapshot-scanner initialization.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java	Uses kv format version layout when decoding ScanKv batches.
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java	Uses kv format version layout for schema-id parsing and fixed-schema decoding.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[platinumhamburg]

Row TTL touches encoding, snapshot reads, lookups, and Flink conversion, so the intended expiration boundary would be helpful to state explicitly in the PR description. Is a row considered expired when currentTime == expirationTime, and is the decision based on a server-supplied clock or the reading client’s clock? A focused test around the exact boundary, including a snapshot created before expiry but read after expiry, would make the behavior easier to preserve across scanners.

Hi @hiSandog , thanks for your suggestion, the documentation changes in the PR explicitly define TTL as a best-effort mechanism rather than providing precise expiration guarantees. There is no commitment to immediate expiration once the time threshold is crossed (i.e., we will not perform additional row-level visibility filtering strictly for TTL semantics, as this would generally exceed typical user requirements and incur significant performance overhead). Furthermore, the expiration decision is explicitly based on the server clock. For scenarios involving event-time TTL, the TTL configuration must clearly specify either precise timezone semantics or an exact timestamp.