[kv] Add row TTL support

fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java

@@ -25,6 +25,7 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.decode.FixedSchemaDecoder;
+import org.apache.fluss.row.encode.KvValueLayout;
 import org.apache.fluss.utils.CopyOnWriteMap;
 import org.apache.fluss.utils.concurrent.FutureUtils;
 
@@ -37,6 +38,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
+import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 
 /** Abstract lookuper implementation for common methods. */
@@ -51,6 +53,8 @@ abstract class AbstractLookuper implements Lookuper {
 
     private final SchemaGetter schemaGetter;
 
+    private final KvValueLayout kvValueLayout;
+
     /**
      * Cache for row decoders for different schema ids. Use CopyOnWriteMap for fast access, as it is
      * not frequently updated.
@@ -67,12 +71,20 @@ abstract class AbstractLookuper implements Lookuper {
         this.lookupClient = lookupClient;
         this.targetSchemaId = (short) tableInfo.getSchemaId();
         this.schemaGetter = schemaGetter;
+        this.kvValueLayout =
+                KvValueLayout.forKvFormatVersion(
+                        tableInfo
+                                .getTableConfig()
+                                .getKvFormatVersion()
+                                .orElse(KV_FORMAT_VERSION_2));
         this.decoders = new CopyOnWriteMap<>();
         // initialize the decoder for the same schema
         this.decoders.put(
                 targetSchemaId,
                 new FixedSchemaDecoder(
-                        tableInfo.getTableConfig().getKvFormat(), tableInfo.getSchema()));
+                        tableInfo.getTableConfig().getKvFormat(),
+                        tableInfo.getSchema(),
+                        kvValueLayout));
     }
 
     protected void handleLookupResponse(
@@ -85,7 +97,7 @@ protected void handleLookupResponse(
                 continue;
             }
             MemorySegment memorySegment = MemorySegment.wrap(valueBytes);
-            short schemaId = memorySegment.getShort(0);
+            short schemaId = kvValueLayout.readSchemaId(memorySegment);
             if (targetSchemaId != schemaId) {
                 allTargetSchema = false;
                 if (!decoders.containsKey(schemaId)) {
@@ -141,7 +153,7 @@ protected LookupResult processAllTargetSchemaRows(List<MemorySegment> valueList)
     protected LookupResult processSchemaMismatchedRows(List<MemorySegment> valueList) {
         List<InternalRow> rowList = new ArrayList<>(valueList.size());
         for (MemorySegment value : valueList) {
-            short schemaId = value.getShort(0);
+            short schemaId = kvValueLayout.readSchemaId(value);
             FixedSchemaDecoder decoder = decoders.get(schemaId);
             checkArgument(decoder != null, "Decoder for schema id %s not found", schemaId);
             InternalRow row = decoder.decode(value);
@@ -161,7 +173,7 @@ protected LookupResult processSchemaRequestedRows(
         // process the value list to convert to target schema
         List<InternalRow> rowList = new ArrayList<>(valueList.size());
         for (MemorySegment value : valueList) {
-            short schemaId = value.getShort(0);
+            short schemaId = kvValueLayout.readSchemaId(value);
             FixedSchemaDecoder decoder =
                     decoders.computeIfAbsent(
                             schemaId,
@@ -170,7 +182,8 @@ protected LookupResult processSchemaRequestedRows(
                                 return new FixedSchemaDecoder(
                                         tableInfo.getTableConfig().getKvFormat(),
                                         sourceSchema,
-                                        tableInfo.getSchema());
+                                        tableInfo.getSchema(),
+                                        kvValueLayout);
                             });
             InternalRow row = decoder.decode(value);
             rowList.add(row);

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java

@@ -228,6 +228,10 @@ public BatchScanner createBatchScanner(TableBucket tableBucket, long snapshotId)
                 snapshotMeta.getSnapshotFiles(),
                 projectedColumns,
                 scannerTmpDir,
+                tableInfo
+                        .getTableConfig()
+                        .getKvFormatVersion()
+                        .orElse(ConfigOptions.KV_FORMAT_VERSION_2),
                 tableInfo.getTableConfig().getKvFormat(),
                 conn.getOrCreateRemoteFileDownloader());
     }

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java

@@ -30,6 +30,7 @@
 import org.apache.fluss.record.ValueRecordReadContext;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.ProjectedRow;
+import org.apache.fluss.row.encode.KvValueLayout;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
 import org.apache.fluss.rpc.messages.PbScanReqForBucket;
 import org.apache.fluss.rpc.messages.ScanKvRequest;
@@ -56,6 +57,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
+
 /**
  * A {@link BatchScanner} that streams every live row of a single primary-key bucket from the tablet
  * server's RocksDB state via a sequence of {@code ScanKv} RPCs. The scan has snapshot isolation:
@@ -109,7 +112,13 @@ public KvBatchScanner(
         this.projectedColumns = projectedColumns;
         this.readContext =
                 ValueRecordReadContext.createReadContext(
-                        schemaGetter, tableInfo.getTableConfig().getKvFormat());
+                        schemaGetter,
+                        tableInfo.getTableConfig().getKvFormat(),
+                        KvValueLayout.forKvFormatVersion(
+                                tableInfo
+                                        .getTableConfig()
+                                        .getKvFormatVersion()
+                                        .orElse(KV_FORMAT_VERSION_2)));
     }
 
     @Nullable

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java

@@ -83,6 +83,7 @@ public class KvSnapshotBatchScanner implements BatchScanner {
 
     private final Path snapshotLocalDirectory;
     private final RemoteFileDownloader remoteFileDownloader;
+    private final int kvFormatVersion;
     private final KvFormat kvFormat;
 
     private final ReentrantLock lock = new ReentrantLock();
@@ -104,6 +105,7 @@ public KvSnapshotBatchScanner(
             List<FsPathAndFileName> fsPathAndFileNames,
             @Nullable int[] projectedFields,
             String scannerTmpDir,
+            int kvFormatVersion,
             KvFormat kvFormat,
             RemoteFileDownloader remoteFileDownloader) {
         this.targetSchema = targetSchema;
@@ -112,6 +114,7 @@ public KvSnapshotBatchScanner(
         this.tableBucket = tableBucket;
         this.fsPathAndFileNames = fsPathAndFileNames;
         this.projectedFields = projectedFields;
+        this.kvFormatVersion = kvFormatVersion;
         this.kvFormat = kvFormat;
         // create a directory to store the snapshot files
         this.snapshotLocalDirectory =
@@ -220,7 +223,8 @@ private void initReaderAsynchronously() {
                                                         projectedFields,
                                                         targetSchemaId,
                                                         targetSchema,
-                                                        schemaGetter);
+                                                        schemaGetter,
+                                                        kvFormatVersion);
                                         readerIsReady.signalAll();
                                     } catch (Throwable e) {
                                         IOUtils.closeQuietly(closeableRegistry);

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java

@@ -34,6 +34,7 @@
 import org.apache.fluss.row.GenericRow;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.ProjectedRow;
+import org.apache.fluss.row.encode.KvValueLayout;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
 import org.apache.fluss.rpc.messages.LimitScanRequest;
 import org.apache.fluss.rpc.messages.LimitScanResponse;
@@ -57,6 +58,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
+
 /** A {@link BatchScanner} implementation that scans a limited number of records from a table. */
 public class LimitBatchScanner implements BatchScanner {
 
@@ -67,6 +70,7 @@ public class LimitBatchScanner implements BatchScanner {
     private final CompletableFuture<LimitScanResponse> scanFuture;
     private final SchemaGetter schemaGetter;
     private final KvFormat kvFormat;
+    private final KvValueLayout kvValueLayout;
     private final int targetSchemaId;
     /** The chunked allocation manager factory to reuse memory for arrow log write batch. */
     private final ChunkedAllocationManager.ChunkedFactory chunkedFactory;
@@ -119,6 +123,12 @@ public LimitBatchScanner(
         this.scanFuture = gateway.limitScan(limitScanRequest);
 
         this.kvFormat = tableInfo.getTableConfig().getKvFormat();
+        this.kvValueLayout =
+                KvValueLayout.forKvFormatVersion(
+                        tableInfo
+                                .getTableConfig()
+                                .getKvFormatVersion()
+                                .orElse(KV_FORMAT_VERSION_2));
         this.endOfInput = false;
         this.chunkedFactory = new ChunkedAllocationManager.ChunkedFactory();
     }
@@ -163,7 +173,7 @@ private List<InternalRow> parseLimitScanResponse(LimitScanResponse limitScanResp
             DefaultValueRecordBatch valueRecords =
                     DefaultValueRecordBatch.pointToByteBuffer(recordsBuffer);
             ValueRecordReadContext readContext =
-                    ValueRecordReadContext.createReadContext(schemaGetter, kvFormat);
+                    ValueRecordReadContext.createReadContext(schemaGetter, kvFormat, kvValueLayout);
             for (ValueRecord record : valueRecords.records(readContext)) {
                 InternalRow row = record.getRow();
                 if (targetSchemaId != record.schemaId()) {

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java

@@ -27,6 +27,7 @@
 import org.apache.fluss.rocksdb.RocksIteratorWrapper;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.ProjectedRow;
+import org.apache.fluss.row.encode.KvValueLayout;
 import org.apache.fluss.row.encode.ValueDecoder;
 import org.apache.fluss.utils.CloseableIterator;
 import org.apache.fluss.utils.CloseableRegistry;
@@ -80,12 +81,15 @@ class SnapshotFilesReader implements CloseableIterator<InternalRow> {
             @Nullable int[] projectedFields,
             int targetSchemaId,
             Schema targetSchema,
-            SchemaGetter schemaGetter)
+            SchemaGetter schemaGetter,
+            int kvFormatVersion)
             throws IOException {
         this.targetSchemaId = targetSchemaId;
         this.targetSchema = targetSchema;
         this.schemaGetter = schemaGetter;
-        this.valueDecoder = new ValueDecoder(schemaGetter, kvFormat);
+        this.valueDecoder =
+                new ValueDecoder(
+                        schemaGetter, kvFormat, KvValueLayout.forKvFormatVersion(kvFormatVersion));
         this.projectedFields = projectedFields;
         closeableRegistry = new CloseableRegistry();
         try {

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -26,6 +26,7 @@
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.LogFormat;
 import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.metadata.RowTtlChangelogMode;
 import org.apache.fluss.utils.ArrayUtils;
 
 import java.lang.reflect.Field;
@@ -56,7 +57,9 @@ public class ConfigOptions {
     public static final String DEFAULT_LISTENER_NAME = "FLUSS";
 
     public static final int KV_FORMAT_VERSION_2 = 2;
+    public static final int KV_FORMAT_VERSION_3 = 3;
     public static final int CURRENT_KV_FORMAT_VERSION = KV_FORMAT_VERSION_2;
+    public static final int MAX_KV_FORMAT_VERSION = KV_FORMAT_VERSION_3;
 
     @Internal
     public static final String[] PARENT_FIRST_LOGGING_PATTERNS =
@@ -1560,7 +1563,9 @@ public class ConfigOptions {
                                     + "when bucket key differs from primary key, which ensures proper prefix lookup support. "
                                     + "When bucket key equals primary key (default bucket key), it still uses datalake's encoder "
                                     + "for optimization (encoded bytes can be reused for bucket calculation). "
-                                    + "Bucket key encoding always uses datalake's encoder to align with datalake bucket calculation.");
+                                    + "Bucket key encoding always uses datalake's encoder to align with datalake bucket calculation. "
+                                    + "(3) Version 3: TTL-enabled primary key tables encode a Fluss-owned TTL timestamp "
+                                    + "before the row payload so compaction filters can clean up expired rows.");
 
     public static final ConfigOption<Boolean> TABLE_KV_STANDBY_REPLICA_ENABLED =
             key("table.kv.standby-replica.enabled")
@@ -1573,6 +1578,33 @@ public class ConfigOptions {
                                     + "Tables created before this option was introduced are treated as disabled. "
                                     + "Can be dynamically enabled via ALTER TABLE.");
 
+    public static final ConfigOption<Duration> TABLE_ROW_TTL =
+            key("table.row.ttl")
+                    .durationType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The best-effort row-level TTL for primary key tables. "
+                                    + "If not set, row-level TTL is disabled. "
+                                    + "Expired rows may remain visible until RocksDB compaction removes them.");
+
+    public static final ConfigOption<RowTtlChangelogMode> TABLE_ROW_TTL_CHANGELOG_MODE =
+            key("table.row.ttl.changelog-mode")
+                    .enumType(RowTtlChangelogMode.class)
+                    .defaultValue(RowTtlChangelogMode.NONE)
+                    .withDescription(
+                            "The changelog mode for row-level TTL cleanup. "
+                                    + "Only 'none' is supported in this version, which means TTL cleanup does not emit delete records.");
+
+    public static final ConfigOption<String> TABLE_ROW_TTL_TIME_COLUMN =
+            key("table.row.ttl.time-column")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The event-time column for row-level TTL. "
+                                    + "If not set, row-level TTL uses processing time. "
+                                    + "If set, the column must be BIGINT epoch milliseconds or TIMESTAMP_LTZ. "
+                                    + "Rows with null event-time values do not expire through TTL.");
+
     public static final ConfigOption<Boolean> TABLE_AUTO_PARTITION_ENABLED =
             key("table.auto-partition.enabled")
                     .booleanType()

fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java

@@ -25,6 +25,7 @@
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.LogFormat;
 import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.metadata.RowTtlChangelogMode;
 import org.apache.fluss.utils.AutoPartitionStrategy;
 
 import java.time.Duration;
@@ -41,6 +42,9 @@
 @PublicEvolving
 public class TableConfig {
 
+    /** Internal table property that binds row TTL time-column to a stable schema column id. */
+    public static final String ROW_TTL_TIME_COLUMN_ID_KEY = "table.row.ttl.time-column-id";
+
     // the table properties configuration
     private final Configuration config;
 
@@ -90,6 +94,27 @@ public long getLogTTLMs() {
         return config.get(ConfigOptions.TABLE_LOG_TTL).toMillis();
     }
 
+    /** Gets the row-level TTL of the table. */
+    public Optional<Duration> getRowTTL() {
+        return config.getOptional(ConfigOptions.TABLE_ROW_TTL);
+    }
+
+    /** Gets the row-level TTL changelog mode of the table. */
+    public RowTtlChangelogMode getRowTTLChangelogMode() {
+        return config.get(ConfigOptions.TABLE_ROW_TTL_CHANGELOG_MODE);
+    }
+
+    /** Gets the optional row-level TTL time column of the table. */
+    public Optional<String> getRowTTLTimeColumn() {
+        return config.getOptional(ConfigOptions.TABLE_ROW_TTL_TIME_COLUMN);
+    }
+
+    /** Gets the internal row-level TTL time-column id, if event-time TTL is enabled. */
+    public Optional<Integer> getRowTTLTimeColumnId() {
+        String value = config.toMap().get(ROW_TTL_TIME_COLUMN_ID_KEY);
+        return value == null ? Optional.empty() : Optional.of(Integer.parseInt(value));
+    }
+
     /** Gets the local segments to retain for tiered log of the table. */
     public int getTieredLogLocalSegments() {
         return config.get(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS);

fluss-common/src/main/java/org/apache/fluss/metadata/RowTtlChangelogMode.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.metadata;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Changelog mode for row-level TTL cleanup.
+ *
+ * @since 0.10
+ */
+@PublicEvolving
+public enum RowTtlChangelogMode {
+    NONE
+}