[platinumhamburg]

Purpose

Linked issue: close #3620

Motivation

Primary key tables retain their latest row state indefinitely. Applications that only need recent state must currently clean it up externally.

This change introduces opt-in row-level TTL for primary key tables. Row TTL is intended for cases that require per-row expiration and can accept weaker changelog semantics. Time-partitioned expiration remains the recommended approach when applicable because dropping complete partitions preserves changelog completeness.

Solution

Row TTL is enabled with table.kv.ttl and is disabled when the option is absent.

CREATE TABLE pk_table (
    id BIGINT,
    event_time TIMESTAMP_LTZ(3),
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.kv.ttl' = '7 d',
    'table.kv.ttl.time-column' = 'event_time'
);

[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.

[hiSandog]

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.

[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.

[wuchong]

Hi @platinumhamburg , thanks for the great pull request.

I have a major compatibility concern with reusing table.kv.format-version=3 for the new RocksDB value layout. This option currently represents the key encoding version: it is consumed by KeyEncoder and KeyDecoder, and existing clients only understand versions 1 and 2. Assigning version 3 to a new table would therefore break existing clients even though the key encoding itself has not changed.

I think we need to distinguish the following three independent compatibility dimensions:

Dimension	Suggested name	Scope	Persisted table property?
Key encoding	Keep table.kv.format-version in this PR; consider table.kv.key-encoding-version in a separate change	Encoding of keys used by clients, RPC requests, and RocksDB	Yes; persisted and immutable
RPC value encoding	Conceptually table.kv.value-encoding-version	Wire format used by Lookup/PutKV and other KV RPCs	No; negotiated through the RPC API version
RocksDB value layout	table.kv.value-layout-version	Physical layout of values stored in RocksDB	Yes, but preferably as a system-managed internal property

These versions solve different compatibility problems:

• The key encoding version is an end-to-end table property because clients must encode keys in exactly the same way as the server and RocksDB. It must remain stable for the lifetime of the table.
• The RPC value encoding version is a wire-protocol concern. It should be negotiated through the RPC API version rather than persisted in table metadata. Clients should continue to receive [schema_id][value_bytes] without needing to understand the server-side timestamp layout.
• The RocksDB value layout version describes the server-side physical representation, such as version 1 being [schema_id][value_bytes] and version 2 being [timestamp][schema_id][value_bytes]. It needs to be persisted so the server can correctly interpret existing data, but application clients should normally not depend on it.

For this PR, I suggest keeping table.kv.format-version unchanged with versions 1 and 2, and introducing only table.kv.value-layout-version. Existing tables without the new property would default to value layout version 1, while new TTL-enabled tables could use:

table.kv.format-version = 2
table.kv.value-layout-version = 2

Before returning values through KV RPCs, the server should remove the storage-only timestamp and expose the existing [schema_id][value_bytes] representation. This keeps the RocksDB layout internal, avoids unnecessarily coupling RPC clients to the storage format, and preserves compatibility with existing lookup/write clients.

This also lays the foundation for dynamically enabling KV TTL, meaning that all tables will default to value-layout-version 2, regardless of whether KV TTL is enabled. However, this can be deferred to future work.

[platinumhamburg]

@wuchong Thanks for your detailed review feedback, The current implementation responds to the P0 compatibility concern with the following model.

Compatibility model

The implementation now separates the three compatibility dimensions discussed in the review:

1. KV key encoding

   table.kv.format-version continues to describe key encoding only. Row TTL does not introduce a new KV format version, so existing key encoders and decoders remain unchanged.

2. RocksDB value layout

   table.kv.value-layout-version is a system-managed internal table property describing the physical RocksDB value layout:

   • PLAIN (version 1): [little-endian schema_id][BinaryRow]
   • TAGGED (version 2): [big-endian 64-bit tag][little-endian schema_id][BinaryRow]

   Tables without this property default to PLAIN, preserving compatibility with existing tables. New non-TTL tables also use PLAIN; only tables that require the tag use TAGGED.

3. KV RPC value encoding

   The KV RPC wire format remains unchanged as [schema_id][BinaryRow]. The internal tag is neither exposed to RPC clients nor transmitted over the network.

   Lookup and prefix-lookup responses pass a slice of the RocksDB value directly to the protobuf serializer, excluding the storage-only prefix without allocating a temporary stripped byte array. Limit scan and scan responses perform the same projection while constructing the existing value-record representation.

Consequently, regular PUT, lookup, prefix lookup, limit scan, and scan APIs did not receive row-TTL-specific version bumps. Existing clients can continue to access ordinary tables and use the unchanged RPC paths against TTL tables.

The exception is direct KV snapshot reading, because that client opens RocksDB data and therefore must understand the physical value layout. GET_KV_SNAPSHOT_METADATA v1 is the only row-TTL-specific compatibility boundary. A v0 client is rejected only when it attempts to read a tagged snapshot; it can still read snapshots using the plain layout.

All physical value encoding and decoding offsets are centralized in KvValueLayout. The storage layout is not inferred from the key format or duplicated in individual read/write paths.

Intentional differences from the suggestions

There are a few places where the implementation does not follow the suggestions literally:

1. The field remains a generic value tag rather than being named timestamp.

   Row TTL currently interprets the tag as a timestamp, but the physical layout does not own that semantic. Other compaction strategies may use the same layout for different metadata; for example, partition TTL for a secondary index may encode a partition ID. Keeping the layout generic avoids coupling a reusable storage contract to row TTL.

2. Non-TTL tables continue to use PLAIN instead of preemptively using TAGGED.

   Dynamic TTL enablement is not supported in this change. Adding an unused eight-byte tag to every ordinary KV value would impose permanent storage and encoding overhead without providing a current capability. A future design for dynamically enabling TTL can define the required layout transition explicitly.

3. The Flink row-count pushdown guard is retained.

   TTL compaction removes entries outside the normal KV write path and therefore does not maintain an exact stored row count. Allowing count pushdown would return incorrect results. The connector declines the pushdown for TTL tables, allowing Flink to execute a normal aggregate over the table scan instead of failing or returning an inaccurate count.

4. No changelog boolean is introduced yet.

   TTL cleanup currently has only one supported behavior: it does not emit delete records. An option whose only meaningful value is false would not provide a real choice. A changelog option can be introduced together with delete-event support when that behavior exists.

The remaining review suggestions were applied, including using Duration.toMillis(), resolving the event-time column from the schema, supporting TIMESTAMP, documenting byte order, retaining partition-expiration documentation, and limiting the RPC compatibility check to tagged snapshot reads. PTAL when you have time.

[wuchong]

Thanks @platinumhamburg , I left some new comments.

[platinumhamburg]

Thanks @platinumhamburg , I left some new comments.

@wuchong Thanks, all comments addressed; please take another look when you have time.

[wuchong]

LookupResultForBucket currently carries both a shared KvValueLayout and a per-value List<KvValueLayout>. This makes the relationship between values, kvValueLayout, and kvValueLayouts difficult to understand and leaks RocksDB storage-layout details into upper layers.

I think raw values should be converted as soon as they leave RocksDB. For example, KvTablet#multiGet—and similarly multiGetFromBufferOrKv, prefixLookup, and limitScan—can use the KvValueLayout already owned by the tablet to return RPC-ready ByteArraySlice values.

From KvTablet upward, callers would only pass RPC-ready values. Then LookupResultForBucket and PrefixLookupResultForBucket would no longer need to carry layout information, and ReplicaManager would not need to reconstruct KvValueLayout from TableConfig. This establishes a clear boundary for raw storage values, prevents accidental misuse of plain byte[], preserves zero-copy RPC serialization, and removes a significant amount of conversion plumbing.

[wuchong]

I squashed and appended a commit to fix the above issue and improved some code. Please take a look. @platinumhamburg

[platinumhamburg]

I squashed and appended a commit to fix the above issue and improved some code. Please take a look. @platinumhamburg

LGTM. The storage boundary is much clearer after this refactoring.