You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/08/28 01:31:42 UTC
(fluss) 01/02: [kv] Add row TTL support (#3621)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 3ab6b9c044b2c2e5aa3ff3784ae0fa387e851cc0
Author: 白鵺 <oc...@alibaba-inc.com>
AuthorDate: Thu Jul 9 23:00:37 2026 +0800
[kv] Add row TTL support (#3621)
---
.../fluss/client/lookup/AbstractLookuper.java | 7 +-
.../fluss/client/table/scanner/TableScan.java | 3 +
.../scanner/batch/KvSnapshotBatchScanner.java | 7 +-
.../table/scanner/batch/SnapshotFilesReader.java | 6 +-
.../fluss/client/admin/FlussAdminITCase.java | 18 +-
.../fluss/client/table/FlussTableITCase.java | 4 +
.../table/scanner/batch/BatchScannerITCase.java | 18 +
.../org/apache/fluss/config/ConfigOptions.java | 32 ++
.../java/org/apache/fluss/config/TableConfig.java | 22 ++
.../fluss/lake/lakestorage/LakeTableLookuper.java | 3 +-
.../java/org/apache/fluss/record/BinaryValue.java | 4 +-
.../apache/fluss/record/DefaultValueRecord.java | 40 ++-
.../fluss/record/DefaultValueRecordBatch.java | 15 +-
.../fluss/record/ValueRecordReadContext.java | 1 +
.../fluss/row/decode/FixedSchemaDecoder.java | 9 +-
.../org/apache/fluss/row/encode/KvValueLayout.java | 208 +++++++++++
.../org/apache/fluss/row/encode/ValueDecoder.java | 20 +-
.../org/apache/fluss/row/encode/ValueEncoder.java | 70 +++-
.../org/apache/fluss/utils/ProtoCodecUtils.java | 1 +
.../org/apache/fluss/config/TableConfigTest.java | 10 +
.../fluss/record/DefaultValueRecordBatchTest.java | 34 +-
.../apache/fluss/row/encode/KvValueLayoutTest.java | 99 ++++++
.../fluss/flink/source/FlinkTableSource.java | 5 +-
.../apache/fluss/flink/utils/FlinkConversions.java | 1 -
.../fluss/flink/catalog/FlinkCatalogITCase.java | 4 +
.../flink/source/FlinkTableSourceBatchITCase.java | 30 +-
.../fluss/flink/utils/FlinkConversionsTest.java | 50 +++
.../generator/generator/ProtobufBytesField.java | 27 +-
.../generator/ProtobufRepeatedBytesField.java | 25 +-
.../org/apache/fluss/protogen/tests/BytesTest.java | 43 ++-
.../fluss/rpc/entity/LookupResultForBucket.java | 59 +++-
.../rpc/entity/PrefixLookupResultForBucket.java | 23 +-
.../org/apache/fluss/rpc/protocol/ApiKeys.java | 2 +-
.../org/apache/fluss/rpc/protocol/ApiKeysTest.java | 5 +
.../org/apache/fluss/server/RpcServiceBase.java | 19 +-
.../server/coordinator/CoordinatorService.java | 46 ++-
.../java/org/apache/fluss/server/kv/KvManager.java | 61 +++-
.../apache/fluss/server/kv/KvRecoverHelper.java | 12 +-
.../java/org/apache/fluss/server/kv/KvTablet.java | 197 +++++++++--
.../apache/fluss/server/kv/KvWriteProcessor.java | 54 ++-
.../server/kv/RowTtlCompactionFilterFactory.java | 68 ++++
.../fluss/server/kv/RowTtlTimestampProvider.java | 131 +++++++
.../fluss/server/kv/rocksdb/RocksDBKvBuilder.java | 21 ++
.../kv/rocksdb/RocksDBResourceContainer.java | 4 +
.../fluss/server/kv/scan/ScannerContext.java | 11 +-
.../org/apache/fluss/server/replica/Replica.java | 20 +-
.../fluss/server/replica/ReplicaManager.java | 49 +--
.../historical/HistoricalPartitionManager.java | 21 +-
.../apache/fluss/server/tablet/TabletServer.java | 7 +-
.../apache/fluss/server/tablet/TabletService.java | 27 +-
.../org/apache/fluss/server/utils/RowTtlUtils.java | 55 +++
.../fluss/server/utils/ServerRpcMessageUtils.java | 17 +-
.../server/utils/TableDescriptorValidation.java | 154 +++++++-
.../RpcServiceBaseKvSnapshotVersionTest.java | 66 ++++
.../server/coordinator/TableManagerITCase.java | 121 ++++++-
.../org/apache/fluss/server/kv/KvTabletTest.java | 391 ++++++++++++++++++++-
.../server/kv/RowTtlCompactionFilterTest.java | 98 ++++++
.../server/kv/RowTtlTimestampProviderTest.java | 149 ++++++++
.../server/kv/rocksdb/RocksDBKvBuilderTest.java | 50 ++-
.../replica/KvRecoverFromRemoteLogITCase.java | 58 ++-
.../fluss/server/replica/ReplicaManagerTest.java | 79 ++++-
.../fluss/server/replica/ReplicaTestBase.java | 3 +-
.../historical/HistoricalPartitionManagerTest.java | 35 +-
.../fluss/server/tablet/TabletServiceITCase.java | 35 +-
.../apache/fluss/server/utils/RowTtlUtilsTest.java | 53 +++
.../server/utils/ServerRpcMessageUtilsTest.java | 22 ++
.../utils/TableDescriptorValidationTest.java | 316 +++++++++++++++++
website/docs/engine-flink/options.md | 2 +
website/docs/table-design/data-distribution/ttl.md | 46 +++
69 files changed, 3141 insertions(+), 262 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java b/fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java
index 6cd93adf7..fd8ec59a4 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java
@@ -25,6 +25,7 @@ import org.apache.fluss.metadata.SchemaInfo;
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;
@@ -85,7 +86,7 @@ abstract class AbstractLookuper implements Lookuper {
continue;
}
MemorySegment memorySegment = MemorySegment.wrap(valueBytes);
- short schemaId = memorySegment.getShort(0);
+ short schemaId = KvValueLayout.PLAIN.readSchemaId(memorySegment);
if (targetSchemaId != schemaId) {
allTargetSchema = false;
if (!decoders.containsKey(schemaId)) {
@@ -141,7 +142,7 @@ abstract class AbstractLookuper implements Lookuper {
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.PLAIN.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 +162,7 @@ abstract class AbstractLookuper implements Lookuper {
// 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.PLAIN.readSchemaId(value);
FixedSchemaDecoder decoder =
decoders.computeIfAbsent(
schemaId,
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
index 51f12abc7..796f6da3f 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
@@ -37,6 +37,7 @@ import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.types.RowType;
import javax.annotation.Nullable;
@@ -208,6 +209,7 @@ public class TableScan implements Scan {
}
String scannerTmpDir =
conn.getConfiguration().getString(ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR);
+ KvValueLayout kvValueLayout = KvValueLayout.fromTableConfig(tableInfo.getTableConfig());
Admin admin = conn.getAdmin();
final KvSnapshotMetadata snapshotMeta;
try {
@@ -228,6 +230,7 @@ public class TableScan implements Scan {
snapshotMeta.getSnapshotFiles(),
projectedColumns,
scannerTmpDir,
+ kvValueLayout,
tableInfo.getTableConfig().getKvFormat(),
conn.getOrCreateRemoteFileDownloader());
}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java
index 52a2c36fb..d3f829c42 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java
@@ -26,6 +26,7 @@ import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.CloseableRegistry;
import org.apache.fluss.utils.FileUtils;
@@ -83,6 +84,7 @@ public class KvSnapshotBatchScanner implements BatchScanner {
private final Path snapshotLocalDirectory;
private final RemoteFileDownloader remoteFileDownloader;
+ private final KvValueLayout kvValueLayout;
private final KvFormat kvFormat;
private final ReentrantLock lock = new ReentrantLock();
@@ -104,6 +106,7 @@ public class KvSnapshotBatchScanner implements BatchScanner {
List<FsPathAndFileName> fsPathAndFileNames,
@Nullable int[] projectedFields,
String scannerTmpDir,
+ KvValueLayout kvValueLayout,
KvFormat kvFormat,
RemoteFileDownloader remoteFileDownloader) {
this.targetSchema = targetSchema;
@@ -112,6 +115,7 @@ public class KvSnapshotBatchScanner implements BatchScanner {
this.tableBucket = tableBucket;
this.fsPathAndFileNames = fsPathAndFileNames;
this.projectedFields = projectedFields;
+ this.kvValueLayout = kvValueLayout;
this.kvFormat = kvFormat;
// create a directory to store the snapshot files
this.snapshotLocalDirectory =
@@ -220,7 +224,8 @@ public class KvSnapshotBatchScanner implements BatchScanner {
projectedFields,
targetSchemaId,
targetSchema,
- schemaGetter);
+ schemaGetter,
+ kvValueLayout);
readerIsReady.signalAll();
} catch (Throwable e) {
IOUtils.closeQuietly(closeableRegistry);
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
index 8a52814f3..5a6cbac5a 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
@@ -27,6 +27,7 @@ import org.apache.fluss.rocksdb.RocksDBHandle;
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,13 @@ class SnapshotFilesReader implements CloseableIterator<InternalRow> {
@Nullable int[] projectedFields,
int targetSchemaId,
Schema targetSchema,
- SchemaGetter schemaGetter)
+ SchemaGetter schemaGetter,
+ KvValueLayout kvValueLayout)
throws IOException {
this.targetSchemaId = targetSchemaId;
this.targetSchema = targetSchema;
this.schemaGetter = schemaGetter;
- this.valueDecoder = new ValueDecoder(schemaGetter, kvFormat);
+ this.valueDecoder = new ValueDecoder(schemaGetter, kvFormat, kvValueLayout);
this.projectedFields = projectedFields;
closeableRegistry = new CloseableRegistry();
try {
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 7b8cd0058..9e491d360 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -118,9 +118,12 @@ import static org.apache.fluss.config.ConfigOptions.CURRENT_KV_FORMAT_VERSION;
import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.config.ConfigOptions.TABLE_KV_FORMAT_VERSION;
+import static org.apache.fluss.config.ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION;
import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
import static org.apache.fluss.record.TestData.DATA1_PARTITIONED_TABLE_DESCRIPTOR;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
+import static org.apache.fluss.row.encode.KvValueLayout.PLAIN;
import static org.apache.fluss.testutils.DataTestUtils.row;
import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow;
import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
@@ -296,9 +299,8 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
TableDescriptor tableDescriptor =
DEFAULT_TABLE_DESCRIPTOR.withReplicationFactor(3).withDataLakeFormat(PAIMON);
Map<String, String> options = new HashMap<>(tableDescriptor.getProperties());
- options.put(
- ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
- String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_FORMAT_VERSION.key(), String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_VALUE_LAYOUT_VERSION.key(), String.valueOf(PLAIN.version()));
options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
assertThat(tableInfo.toTableDescriptor())
.isEqualTo(tableDescriptor.withProperties(options));
@@ -326,9 +328,8 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
TableDescriptor expected =
DEFAULT_TABLE_DESCRIPTOR.withReplicationFactor(3).withDataLakeFormat(PAIMON);
options = new HashMap<>(expected.getProperties());
- options.put(
- ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
- String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_FORMAT_VERSION.key(), String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_VALUE_LAYOUT_VERSION.key(), String.valueOf(PLAIN.version()));
options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected.withProperties(options));
assertThat(schemaInfo2).isEqualTo(schemaInfo);
@@ -1062,9 +1063,8 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
TableDescriptor expected =
DEFAULT_TABLE_DESCRIPTOR.withReplicationFactor(3).withDataLakeFormat(PAIMON);
Map<String, String> options = new HashMap<>(expected.getProperties());
- options.put(
- ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
- String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_FORMAT_VERSION.key(), String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ options.put(TABLE_KV_VALUE_LAYOUT_VERSION.key(), String.valueOf(PLAIN.version()));
options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected.withProperties(options));
}
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
index 03bcb8d96..60837895e 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
@@ -51,6 +51,7 @@ import org.apache.fluss.row.BinaryString;
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.row.indexed.IndexedRow;
import org.apache.fluss.types.BigIntType;
import org.apache.fluss.types.DataTypes;
@@ -109,6 +110,9 @@ class FlussTableITCase extends ClientToServerITCaseBase {
.withDataLakeFormat(DataLakeFormat.PAIMON);
Map<String, String> options = new HashMap<>(expected.getProperties());
options.put(ConfigOptions.TABLE_KV_FORMAT_VERSION.key(), "2");
+ options.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version()));
options.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
expected = expected.withProperties(options);
assertThat(tableInfo.toTableDescriptor()).isEqualTo(expected);
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/BatchScannerITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/BatchScannerITCase.java
index 67e1616eb..b4c833fa5 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/BatchScannerITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/BatchScannerITCase.java
@@ -25,6 +25,7 @@ import org.apache.fluss.client.table.Table;
import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.client.write.HashBucketAssigner;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
@@ -126,6 +127,23 @@ class BatchScannerITCase extends ClientToServerITCaseBase {
testSnapshotRead(tablePath, expectedRowByBuckets);
}
+ @Test
+ void testScanTaggedSnapshot() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "test-tagged-table-snapshot");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(DEFAULT_SCHEMA)
+ .distributedBy(DEFAULT_BUCKET_NUM, "id")
+ .property(ConfigOptions.TABLE_KV_TTL.key(), "1 h")
+ .build();
+ long tableId = createTable(tablePath, tableDescriptor, true);
+
+ Map<TableBucket, List<InternalRow>> expectedRowByBuckets = putRows(tableId, tablePath, 10);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshots(expectedRowByBuckets.keySet());
+
+ testSnapshotRead(tablePath, expectedRowByBuckets);
+ }
+
@Test
void testScanSnapshotDuringSchemaChange() throws Exception {
TablePath tablePath = TablePath.of(DEFAULT_DB, "test-table-snapshot-schema-change");
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index a0104c7eb..16301f461 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1746,6 +1746,16 @@ public class ConfigOptions {
+ "for optimization (encoded bytes can be reused for bucket calculation). "
+ "Bucket key encoding always uses datalake's encoder to align with datalake bucket calculation.");
+ /** The version of the physical KV value layout. */
+ public static final ConfigOption<Integer> TABLE_KV_VALUE_LAYOUT_VERSION =
+ key("table.kv.value-layout-version")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The version of the physical KV value layout in RocksDB. "
+ + "The coordinator sets this option during table creation. "
+ + "Tables created before this option was introduced use the plain layout.");
+
public static final ConfigOption<Boolean> TABLE_KV_STANDBY_REPLICA_ENABLED =
key("table.kv.standby-replica.enabled")
.booleanType()
@@ -1757,6 +1767,28 @@ 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_KV_TTL =
+ key("table.kv.ttl")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "The best-effort row-level TTL for primary key data. "
+ + "The changelog retention of a primary key table is controlled separately by 'table.log.ttl'. "
+ + "If not set, row-level TTL is disabled. "
+ + "The duration must be at least 1 millisecond. "
+ + "Expired rows may remain visible until RocksDB compaction removes them.");
+
+ public static final ConfigOption<String> TABLE_KV_TTL_TIME_COLUMN =
+ key("table.kv.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, TIMESTAMP, or TIMESTAMP_LTZ. "
+ + "TIMESTAMP values are interpreted in the TabletServer's system time zone. "
+ + "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()
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
index 1cdc52956..ea046e3b1 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
@@ -77,6 +77,18 @@ public class TableConfig {
return config.getOptional(ConfigOptions.TABLE_KV_FORMAT_VERSION);
}
+ /**
+ * Gets the physical KV value layout version persisted with the table.
+ *
+ * <p>The KV format version controls the key encoding strategy, while the value layout version
+ * controls the fixed header surrounding the BinaryRow payload in RocksDB. An empty value
+ * identifies tables created before value layouts were versioned; those tables use the plain
+ * layout.
+ */
+ public Optional<Integer> getKvValueLayoutVersion() {
+ return config.getOptional(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION);
+ }
+
/**
* Whether standby replicas are enabled for this primary key table. Returns false for legacy
* tables that were created before this option was introduced.
@@ -90,6 +102,16 @@ public class TableConfig {
return config.get(ConfigOptions.TABLE_LOG_TTL).toMillis();
}
+ /** Gets the row-level TTL of the table. */
+ public Optional<Duration> getKvTTL() {
+ return config.getOptional(ConfigOptions.TABLE_KV_TTL);
+ }
+
+ /** Gets the optional row-level TTL time column of the table. */
+ public Optional<String> getKvTTLTimeColumn() {
+ return config.getOptional(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN);
+ }
+
/** Gets the local segments to retain for tiered log of the table. */
public int getTieredLogLocalSegments() {
return config.get(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS);
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java
index d5d228881..ab2d201b7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java
@@ -54,7 +54,8 @@ public interface LakeTableLookuper extends AutoCloseable {
*
* @param key lake-format encoded primary key bytes
* @param context lookup context
- * @return Fluss value bytes, or null if the key does not exist
+ * @return plain Fluss value bytes encoded as {@code [schema-id][BinaryRow]}, or null if the key
+ * does not exist
*/
@Nullable
byte[] lookup(byte[] key, LookupContext context) throws Exception;
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java b/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
index 13906347d..befa3eda4 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
@@ -34,8 +34,8 @@ public class BinaryValue {
}
/**
- * Encode the value (consisted of {@code row} with a {@code schemaId}) to a byte array value to
- * be expected persisted to kv store.
+ * Encodes this value using the PLAIN {@code [schema id][BinaryRow]} representation used by
+ * ordinary KV values and RPC value bodies.
*/
public byte[] encodeValue() {
return ValueEncoder.encodeValue(schemaId, row);
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecord.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecord.java
index 8b97825e4..94b7ec010 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecord.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecord.java
@@ -21,6 +21,7 @@ import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.memory.MemorySegmentOutputView;
import org.apache.fluss.row.BinaryRow;
import org.apache.fluss.row.decode.RowDecoder;
+import org.apache.fluss.row.encode.KvValueLayout;
import java.io.IOException;
@@ -31,8 +32,7 @@ import java.io.IOException;
*
* <ul>
* <li>Length => int32
- * <li>SchemaId => int16
- * <li>Value => {@link BinaryRow}
+ * <li>RawValue => schema id and {@link BinaryRow}
* </ul>
*
* @since 0.3
@@ -41,16 +41,22 @@ public class DefaultValueRecord implements ValueRecord {
static final int LENGTH_OFFSET = 0;
static final int LENGTH_LENGTH = 4;
- static final int SCHEMA_ID_OFFSET = LENGTH_LENGTH;
- static final int SCHEMA_ID_LENGTH = 2;
- static final int VALUE_OFFSET = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH;
private final short schemaId;
private final BinaryRow row;
+ private final int sizeInBytes;
public DefaultValueRecord(short schemaId, BinaryRow row) {
+ this(
+ schemaId,
+ row,
+ LENGTH_LENGTH + KvValueLayout.PLAIN.rowPayloadOffset() + row.getSizeInBytes());
+ }
+
+ private DefaultValueRecord(short schemaId, BinaryRow row, int sizeInBytes) {
this.schemaId = schemaId;
this.row = row;
+ this.sizeInBytes = sizeInBytes;
}
@Override
@@ -65,27 +71,33 @@ public class DefaultValueRecord implements ValueRecord {
@Override
public int getSizeInBytes() {
- return row.getSizeInBytes() + LENGTH_LENGTH + SCHEMA_ID_LENGTH;
+ return sizeInBytes;
}
public int writeTo(MemorySegmentOutputView outputView) throws IOException {
- int sizeInBytes = getSizeInBytes();
- outputView.writeInt(sizeInBytes - LENGTH_LENGTH);
- outputView.writeShort(schemaId);
+ KvValueLayout layout = KvValueLayout.PLAIN;
+ int valueLength = layout.schemaIdLength() + row.getSizeInBytes();
+ int sizeInBytes = LENGTH_LENGTH + valueLength;
+ outputView.writeInt(valueLength);
+ int valueOffset = outputView.getPosition();
+ outputView.setPosition(valueOffset + layout.schemaIdLength());
+ layout.writeSchemaId(outputView.getSharedBuffer(), valueOffset, schemaId);
outputView.write(row.getSegments()[0], row.getOffset(), row.getSizeInBytes());
return sizeInBytes;
}
public static DefaultValueRecord readFrom(
MemorySegment segment, int position, ValueRecordBatch.ReadContext readContext) {
- int sizeInBytesWithoutLength = segment.getInt(position + LENGTH_OFFSET);
- short schemaId = segment.getShort(position + SCHEMA_ID_OFFSET);
+ int valueLength = segment.getInt(position + LENGTH_OFFSET);
+ int valueOffset = position + LENGTH_LENGTH;
+ KvValueLayout layout = KvValueLayout.PLAIN;
+ short schemaId = layout.readSchemaId(segment, valueOffset);
RowDecoder decoder = readContext.getRowDecoder(schemaId);
BinaryRow value =
decoder.decode(
segment,
- position + VALUE_OFFSET,
- sizeInBytesWithoutLength - SCHEMA_ID_LENGTH);
- return new DefaultValueRecord(schemaId, value);
+ valueOffset + layout.rowPayloadOffset(),
+ layout.rowPayloadLength(valueLength));
+ return new DefaultValueRecord(schemaId, value, LENGTH_LENGTH + valueLength);
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java
index 558772154..3787f24f7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java
@@ -268,13 +268,22 @@ public class DefaultValueRecordBatch implements ValueRecordBatch {
* @param valueBytes consisted of schema id and the row encoded in the value bytes
*/
public void append(byte[] valueBytes) throws IOException {
+ append(valueBytes, 0, valueBytes.length);
+ }
+
+ /**
+ * @param valueBytes consisted of schema id and the row encoded in the value bytes
+ * @param offset the offset of the value bytes to append
+ * @param length the number of value bytes to append
+ */
+ public void append(byte[] valueBytes, int offset, int length) throws IOException {
if (isClosed) {
throw new IllegalStateException(
"Tried to add a record, but ValueRecordBatchBuilder is closed for record adds.");
}
- outputView.writeInt(valueBytes.length);
- outputView.write(valueBytes);
- sizeInBytes += valueBytes.length + LENGTH_LENGTH;
+ outputView.writeInt(length);
+ outputView.write(valueBytes, offset, length);
+ sizeInBytes += length + LENGTH_LENGTH;
currentRecordNumber++;
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java b/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java
index e00559e00..368989ab0 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java
@@ -38,6 +38,7 @@ public class ValueRecordReadContext implements ValueRecordBatch.ReadContext {
this.kvFormat = kvFormat;
}
+ /** Creates a read context for raw RPC values. */
public static ValueRecordReadContext createReadContext(
SchemaGetter schemaGetter, KvFormat kvFormat) {
return new ValueRecordReadContext(schemaGetter, kvFormat);
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java
index d809a095f..faa42df1f 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java
@@ -23,11 +23,10 @@ import org.apache.fluss.metadata.Schema;
import org.apache.fluss.record.ValueRecord;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.ProjectedRow;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.types.DataType;
import org.apache.fluss.utils.SchemaUtil;
-import static org.apache.fluss.row.encode.ValueEncoder.SCHEMA_ID_LENGTH;
-
/**
* A decoder that deserializes raw byte arrays of {@link ValueRecord} with dynamic or
* self-describing schemas into structured {@link InternalRow} objects conforming to a predefined,
@@ -90,6 +89,10 @@ public class FixedSchemaDecoder {
* adheres to the fixed {@code targetSchema}.
*/
public InternalRow decode(MemorySegment valueSegment) {
- return decode(valueSegment, SCHEMA_ID_LENGTH, valueSegment.size() - SCHEMA_ID_LENGTH);
+ KvValueLayout layout = KvValueLayout.PLAIN;
+ return decode(
+ valueSegment,
+ layout.rowPayloadOffset(),
+ layout.rowPayloadLength(valueSegment.size()));
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/KvValueLayout.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/KvValueLayout.java
new file mode 100644
index 000000000..4d43b8756
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/KvValueLayout.java
@@ -0,0 +1,208 @@
+/*
+ * 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.row.encode;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.utils.UnsafeUtils;
+
+import static org.apache.fluss.utils.Preconditions.checkState;
+
+/**
+ * Physical layout of raw KV value bytes.
+ *
+ * <p>PLAIN (version 1): {@code [little-endian schema id][BinaryRow payload]}.
+ *
+ * <p>TAGGED (version 2): {@code [big-endian long tag][little-endian schema id][BinaryRow payload]}.
+ * The tag is an opaque signed 64-bit value whose semantics are defined by the producer and consumer
+ * of the selected layout. Its byte order follows FRocksDB's native {@code DeserializeTimestamp}
+ * contract, which reads the eight-byte value most-significant byte first.
+ */
+@Internal
+public final class KvValueLayout {
+
+ private static final int SCHEMA_ID_LENGTH = 2;
+ private static final int VALUE_TAG_LENGTH = 8;
+ private static final int NO_OFFSET = -1;
+
+ public static final KvValueLayout PLAIN = new KvValueLayout(1, 0, 0, 2, NO_OFFSET);
+ public static final KvValueLayout TAGGED = new KvValueLayout(2, 8, 8, 10, 0);
+
+ private final int version;
+ private final int valueBodyOffset;
+ private final int schemaIdOffset;
+ private final int rowPayloadOffset;
+ private final int valueTagOffset;
+
+ private KvValueLayout(
+ int version,
+ int valueBodyOffset,
+ int schemaIdOffset,
+ int rowPayloadOffset,
+ int valueTagOffset) {
+ this.version = version;
+ this.valueBodyOffset = valueBodyOffset;
+ this.schemaIdOffset = schemaIdOffset;
+ this.rowPayloadOffset = rowPayloadOffset;
+ this.valueTagOffset = valueTagOffset;
+ }
+
+ /** Returns the layout for the persisted version. */
+ public static KvValueLayout fromVersion(int version) {
+ switch (version) {
+ case 1:
+ return PLAIN;
+ case 2:
+ return TAGGED;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported KV value layout version " + version + ".");
+ }
+ }
+
+ /** Returns the layout persisted in the table config, defaulting to the plain layout. */
+ public static KvValueLayout fromTableConfig(TableConfig tableConfig) {
+ return fromVersion(tableConfig.getKvValueLayoutVersion().orElse(PLAIN.version()));
+ }
+
+ /** Returns the persisted version of this layout. */
+ public int version() {
+ return version;
+ }
+
+ /** Returns the byte offset of the RPC value body in a raw KV value. */
+ public int valueBodyOffset() {
+ return valueBodyOffset;
+ }
+
+ /** Returns the RPC value body length for a raw KV value length. */
+ public int valueBodyLength(int valueLength) {
+ return valueLength - valueBodyOffset;
+ }
+
+ /** Returns the byte offset of schema id in a raw KV value. */
+ public int schemaIdOffset() {
+ return schemaIdOffset;
+ }
+
+ /** Returns the encoded schema id length in bytes. */
+ public int schemaIdLength() {
+ return SCHEMA_ID_LENGTH;
+ }
+
+ /** Returns whether the raw KV value has an internal value tag. */
+ public boolean hasValueTag() {
+ return valueTagOffset != NO_OFFSET;
+ }
+
+ /** Returns the byte offset of the internal value tag. */
+ public int valueTagOffset() {
+ checkState(hasValueTag(), "KV value layout does not have a value tag.");
+ return valueTagOffset;
+ }
+
+ /** Returns the internal value tag length in bytes. */
+ public int valueTagLength() {
+ return hasValueTag() ? VALUE_TAG_LENGTH : 0;
+ }
+
+ /** Returns the byte offset of row payload in a raw KV value. */
+ public int rowPayloadOffset() {
+ return rowPayloadOffset;
+ }
+
+ /** Returns the row payload length for a raw KV value length. */
+ public int rowPayloadLength(int valueLength) {
+ if (valueLength < rowPayloadOffset) {
+ throw new IllegalArgumentException(
+ "valueLength must be at least row payload offset "
+ + rowPayloadOffset
+ + ", but was "
+ + valueLength
+ + ".");
+ }
+ return valueLength - rowPayloadOffset;
+ }
+
+ /** Reads the schema id from a raw KV value. */
+ public short readSchemaId(MemorySegment value) {
+ return readSchemaId(value, 0);
+ }
+
+ /** Reads the schema id from a raw KV value embedded at the given offset. */
+ public short readSchemaId(MemorySegment value, int valueOffset) {
+ return value.getShort(valueOffset + schemaIdOffset);
+ }
+
+ /** Writes the schema id to a raw KV value. */
+ public void writeSchemaId(byte[] value, short schemaId) {
+ writeSchemaId(value, 0, schemaId);
+ }
+
+ /** Writes the schema id to a raw KV value embedded at the given offset. */
+ public void writeSchemaId(byte[] value, int valueOffset, short schemaId) {
+ short littleEndianSchemaId =
+ MemorySegment.LITTLE_ENDIAN ? schemaId : Short.reverseBytes(schemaId);
+ UnsafeUtils.putShort(value, valueOffset + schemaIdOffset, littleEndianSchemaId);
+ }
+
+ /**
+ * Reads the opaque signed 64-bit value tag from a raw KV value. The layout does not assign
+ * semantics to this field.
+ */
+ public long readValueTag(MemorySegment value) {
+ return readValueTag(value, 0);
+ }
+
+ /**
+ * Reads the opaque signed 64-bit value tag from a raw KV value embedded at the given offset.
+ * The layout does not assign semantics to this field.
+ */
+ public long readValueTag(MemorySegment value, int valueOffset) {
+ return value.getLongBigEndian(valueOffset + valueTagOffset());
+ }
+
+ /**
+ * Writes an opaque signed 64-bit value tag to a raw KV value. The producer defines the tag's
+ * semantics.
+ */
+ public void writeValueTag(byte[] value, long valueTag) {
+ writeValueTag(value, 0, valueTag);
+ }
+
+ /**
+ * Writes an opaque signed 64-bit value tag to a raw KV value embedded at the given offset. The
+ * producer defines the tag's semantics.
+ */
+ public void writeValueTag(byte[] value, int valueOffset, long valueTag) {
+ int absoluteValueTagOffset = valueOffset + valueTagOffset();
+ if (absoluteValueTagOffset < 0
+ || absoluteValueTagOffset > value.length - VALUE_TAG_LENGTH) {
+ throw new IndexOutOfBoundsException(
+ "Cannot write value tag at offset "
+ + absoluteValueTagOffset
+ + " to value of length "
+ + value.length
+ + ".");
+ }
+ long bigEndianValueTag =
+ MemorySegment.LITTLE_ENDIAN ? Long.reverseBytes(valueTag) : valueTag;
+ UnsafeUtils.putLong(value, absoluteValueTagOffset, bigEndianValueTag);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
index 5e243799c..2c0727a27 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
@@ -29,28 +29,30 @@ import org.apache.fluss.types.DataType;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import static org.apache.fluss.row.encode.ValueEncoder.SCHEMA_ID_LENGTH;
-
-/**
- * A decoder to decode a schema id and {@link BinaryRow} from a byte array value which is encoded by
- * {@link ValueEncoder#encodeValue(short, BinaryRow)}.
- */
+/** Decodes a schema id and {@link BinaryRow} from bytes described by a {@link KvValueLayout}. */
public class ValueDecoder {
private final Map<Short, RowDecoder> rowDecoders;
private final SchemaGetter schemaGetter;
private final KvFormat kvFormat;
+ private final KvValueLayout kvValueLayout;
+ /** Creates a decoder for the plain KV value layout. */
public ValueDecoder(SchemaGetter schemaGetter, KvFormat kvFormat) {
+ this(schemaGetter, kvFormat, KvValueLayout.PLAIN);
+ }
+
+ public ValueDecoder(SchemaGetter schemaGetter, KvFormat kvFormat, KvValueLayout kvValueLayout) {
this.rowDecoders = new ConcurrentHashMap<>();
this.schemaGetter = schemaGetter;
this.kvFormat = kvFormat;
+ this.kvValueLayout = kvValueLayout;
}
/** Decode the value bytes and return the schema id and the row encoded in the value bytes. */
public BinaryValue decodeValue(byte[] valueBytes) {
MemorySegment memorySegment = MemorySegment.wrap(valueBytes);
- short schemaId = memorySegment.getShort(0);
+ short schemaId = kvValueLayout.readSchemaId(memorySegment);
RowDecoder rowDecoder =
rowDecoders.computeIfAbsent(
@@ -64,7 +66,9 @@ public class ValueDecoder {
BinaryRow row =
rowDecoder.decode(
- memorySegment, SCHEMA_ID_LENGTH, valueBytes.length - SCHEMA_ID_LENGTH);
+ memorySegment,
+ kvValueLayout.rowPayloadOffset(),
+ kvValueLayout.rowPayloadLength(valueBytes.length));
return new BinaryValue(schemaId, row);
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
index 2d842ab17..13069a165 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
@@ -17,25 +17,67 @@
package org.apache.fluss.row.encode;
+import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.row.BinaryRow;
-import org.apache.fluss.utils.UnsafeUtils;
+
+import java.util.function.ToLongFunction;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
/** An encoder to encode {@link BinaryRow} with a schema id as value to be stored in kv store. */
-public class ValueEncoder {
+public final class ValueEncoder {
- public static final int SCHEMA_ID_LENGTH = 2;
+ private static final ValueEncoder PLAIN_ENCODER = new ValueEncoder(KvValueLayout.PLAIN, null);
- /**
- * Encode the {@code row} with a {@code schemaId} to a byte array value to be expected persisted
- * to kv store.
- *
- * @param schemaId the schema id of the row
- * @param row the row to encode
- */
- public static byte[] encodeValue(short schemaId, BinaryRow row) {
- byte[] values = new byte[SCHEMA_ID_LENGTH + row.getSizeInBytes()];
- UnsafeUtils.putShort(values, 0, schemaId);
- row.copyTo(values, SCHEMA_ID_LENGTH);
+ private final KvValueLayout kvValueLayout;
+ private final ToLongFunction<BinaryRow> valueTagProvider;
+
+ private ValueEncoder(KvValueLayout kvValueLayout, ToLongFunction<BinaryRow> valueTagProvider) {
+ this.kvValueLayout = kvValueLayout;
+ this.valueTagProvider = valueTagProvider;
+ }
+
+ /** Returns an encoder for a layout without an internal value tag. */
+ public static ValueEncoder forLayout(KvValueLayout kvValueLayout) {
+ checkNotNull(kvValueLayout, "kvValueLayout must not be null.");
+ if (kvValueLayout != KvValueLayout.PLAIN) {
+ throw new IllegalArgumentException(
+ "A value tag provider is required for this KV value layout.");
+ }
+ return PLAIN_ENCODER;
+ }
+
+ /** Returns an encoder for a layout with an internal value tag. */
+ public static ValueEncoder forLayout(
+ KvValueLayout kvValueLayout, ToLongFunction<BinaryRow> valueTagProvider) {
+ checkNotNull(kvValueLayout, "kvValueLayout must not be null.");
+ checkNotNull(valueTagProvider, "valueTagProvider must not be null.");
+ if (!kvValueLayout.hasValueTag()) {
+ throw new IllegalArgumentException(
+ "A value tag provider is not supported for this KV value layout.");
+ }
+ return new ValueEncoder(kvValueLayout, valueTagProvider);
+ }
+
+ /** Returns whether this encoder writes an internal value tag before the row bytes. */
+ public boolean hasValueTag() {
+ return kvValueLayout.hasValueTag();
+ }
+
+ /** Encodes a binary value using the layout bound to this encoder. */
+ public byte[] encodeValue(BinaryValue value) {
+ int rowPayloadOffset = kvValueLayout.rowPayloadOffset();
+ byte[] values = new byte[rowPayloadOffset + value.row.getSizeInBytes()];
+ kvValueLayout.writeSchemaId(values, value.schemaId);
+ if (valueTagProvider != null) {
+ kvValueLayout.writeValueTag(values, valueTagProvider.applyAsLong(value.row));
+ }
+ value.row.copyTo(values, rowPayloadOffset);
return values;
}
+
+ /** Encodes a plain binary value for callers that do not select a layout explicitly. */
+ public static byte[] encodeValue(short schemaId, BinaryRow row) {
+ return PLAIN_ENCODER.encodeValue(new BinaryValue(schemaId, row));
+ }
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/ProtoCodecUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/ProtoCodecUtils.java
index daa37f9b6..4ab4ccdec 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/ProtoCodecUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/ProtoCodecUtils.java
@@ -302,6 +302,7 @@ public final class ProtoCodecUtils {
/** Holder for a byte array and byte length. */
public static final class BytesHolder {
public byte[] b;
+ public int offset;
public int len;
}
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java
index 68b1e6c50..4a732beb7 100644
--- a/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java
@@ -61,4 +61,14 @@ class TableConfigTest {
conf.removeConfig(ConfigOptions.TABLE_LOG_LOCAL_TTL);
assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofDays(3).toMillis());
}
+
+ @Test
+ void testKvValueLayoutVersion() {
+ Configuration conf = new Configuration();
+ TableConfig tableConfig = new TableConfig(conf);
+
+ assertThat(tableConfig.getKvValueLayoutVersion()).isEmpty();
+ conf.set(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION, 2);
+ assertThat(tableConfig.getKvValueLayoutVersion()).hasValue(2);
+ }
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java b/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java
index 5558ec7c0..67e3a3a0b 100644
--- a/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java
@@ -17,14 +17,20 @@
package org.apache.fluss.record;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.row.encode.ValueEncoder;
+
import org.junit.jupiter.api.Test;
import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for {@link DefaultValueRecordBatch}. */
+/** Tests for {@link DefaultValueRecordBatch}. */
class DefaultValueRecordBatchTest {
@Test
@@ -66,4 +72,30 @@ class DefaultValueRecordBatchTest {
builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "a2"}));
return builder.build();
}
+
+ @Test
+ void testAppendStorageValueSliceDecodesOriginalRpcRow() throws Exception {
+ BinaryRow row = compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
+ KvValueLayout layout = KvValueLayout.TAGGED;
+ byte[] storageValue =
+ ValueEncoder.forLayout(layout, ignored -> 123L)
+ .encodeValue(new BinaryValue(DEFAULT_SCHEMA_ID, row));
+
+ DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder();
+ builder.append(
+ storageValue,
+ layout.valueBodyOffset(),
+ layout.valueBodyLength(storageValue.length));
+
+ ValueRecord record =
+ builder.build()
+ .records(
+ ValueRecordReadContext.createReadContext(
+ new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA1_SCHEMA),
+ KvFormat.COMPACTED))
+ .iterator()
+ .next();
+ assertThat(record.schemaId()).isEqualTo(DEFAULT_SCHEMA_ID);
+ assertThat(record.getRow()).isEqualTo(row);
+ }
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/row/encode/KvValueLayoutTest.java b/fluss-common/src/test/java/org/apache/fluss/row/encode/KvValueLayoutTest.java
new file mode 100644
index 000000000..674854cd4
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/row/encode/KvValueLayoutTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.row.encode;
+
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.record.BinaryValue;
+import org.apache.fluss.record.TestingSchemaGetter;
+import org.apache.fluss.row.BinaryRow;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
+import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for KV value layouts. */
+class KvValueLayoutTest {
+
+ @Test
+ void testLayoutVersionMapping() {
+ assertThat(KvValueLayout.fromVersion(1)).isSameAs(KvValueLayout.PLAIN);
+ assertThat(KvValueLayout.fromVersion(2)).isSameAs(KvValueLayout.TAGGED);
+ assertThatThrownBy(() -> KvValueLayout.fromVersion(3))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported KV value layout version 3");
+ }
+
+ @Test
+ void testLongTagKeepsRpcValueBodyAsSuffix() {
+ BinaryRow row = compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
+ byte[] value =
+ ValueEncoder.forLayout(KvValueLayout.TAGGED, ignored -> 123L)
+ .encodeValue(new BinaryValue(DEFAULT_SCHEMA_ID, row));
+
+ assertThat(KvValueLayout.TAGGED.valueBodyOffset()).isEqualTo(8);
+ assertThat(KvValueLayout.TAGGED.schemaIdOffset()).isEqualTo(8);
+ assertThat(KvValueLayout.TAGGED.rowPayloadOffset()).isEqualTo(10);
+ assertThat(KvValueLayout.TAGGED.readValueTag(MemorySegment.wrap(value))).isEqualTo(123L);
+ assertThat(KvValueLayout.TAGGED.readSchemaId(MemorySegment.wrap(value)))
+ .isEqualTo(DEFAULT_SCHEMA_ID);
+ }
+
+ @Test
+ void testTwoArgumentValueDecoderUsesPlainLayout() {
+ BinaryRow row = compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
+ byte[] value = ValueEncoder.encodeValue(DEFAULT_SCHEMA_ID, row);
+ TestingSchemaGetter schemaGetter = new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
+
+ BinaryValue implicitPlain =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED).decodeValue(value);
+ BinaryValue explicitPlain =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.PLAIN)
+ .decodeValue(value);
+
+ assertThat(implicitPlain).isEqualTo(explicitPlain);
+ }
+
+ @Test
+ void testPersistedHeaderByteOrder() {
+ byte[] plainValue = new byte[KvValueLayout.PLAIN.rowPayloadOffset()];
+ KvValueLayout.PLAIN.writeSchemaId(plainValue, (short) 0x1234);
+ assertThat(plainValue).containsExactly((byte) 0x34, (byte) 0x12);
+
+ byte[] taggedValue = new byte[KvValueLayout.TAGGED.rowPayloadOffset()];
+ KvValueLayout.TAGGED.writeValueTag(taggedValue, 0x0102030405060708L);
+ KvValueLayout.TAGGED.writeSchemaId(taggedValue, (short) 0x1234);
+ assertThat(taggedValue)
+ .containsExactly(
+ (byte) 0x01,
+ (byte) 0x02,
+ (byte) 0x03,
+ (byte) 0x04,
+ (byte) 0x05,
+ (byte) 0x06,
+ (byte) 0x07,
+ (byte) 0x08,
+ (byte) 0x34,
+ (byte) 0x12);
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
index 25b50a8f0..b64483565 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
@@ -844,7 +844,7 @@ public class FlinkTableSource
List<int[]> groupingSets,
List<AggregateExpression> aggregateExpressions,
DataType dataType) {
- // Only supports 'select count(*)/count(1) from source' for log table now.
+ // Only supports global count when an exact row count is available.
if (streaming
|| aggregateExpressions.size() != 1
|| groupingSets.size() > 1
@@ -852,7 +852,8 @@ public class FlinkTableSource
// The count pushdown feature is not supported when the data lake is enabled.
// Otherwise, it'll cause miss count data in lake. But In the future, we can push
// down count into lake.
- || isDataLakeEnabled) {
+ || isDataLakeEnabled
+ || (hasPrimaryKey() && tableConfig.getKvTTL().isPresent())) {
return false;
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
index c200db857..9a6fb038c 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
@@ -128,7 +128,6 @@ public class FlinkConversions {
// put fluss table properties into flink options, to make the properties visible to users
convertFlussTablePropertiesToFlinkOptions(tableInfo.getProperties().toMap(), newOptions);
-
org.apache.flink.table.api.Schema.Builder schemaBuilder =
org.apache.flink.table.api.Schema.newBuilder();
if (tableInfo.hasPrimaryKey()) {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
index 57b2b0fab..5ce250d62 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
@@ -32,6 +32,7 @@ import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.testutils.common.MultiVersionTest;
@@ -758,6 +759,9 @@ abstract class FlinkCatalogITCase {
expectedTableProperties.put("table.replication.factor", "1");
expectedTableProperties.put(
"table.kv.format-version", String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ expectedTableProperties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version()));
expectedTableProperties.put("table.kv.standby-replica.enabled", "true");
assertThat(tableInfo.getProperties().toMap()).isEqualTo(expectedTableProperties);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
index 0e48a01bc..24f69a5d0 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
@@ -20,7 +20,6 @@ package org.apache.fluss.flink.source;
import org.apache.fluss.client.table.Table;
import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.client.table.writer.UpsertWriter;
-import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.flink.utils.FlinkTestBase;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.BinaryString;
@@ -658,16 +657,39 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " with ('bucket.num' = '4', 'table.changelog.image' = 'wal')",
tableName))
.await();
- // normal scan
+
String query = String.format("SELECT COUNT(*) FROM %s", tableName);
- assertThatThrownBy(() -> tEnv.executeSql(query))
- .hasRootCauseInstanceOf(InvalidTableException.class)
+ assertThatThrownBy(() -> tEnv.explainSql(query))
.hasMessageContaining(
String.format(
"Row count is disabled for this table '%s.test_count_table_with_wal'.",
databaseName));
}
+ @Test
+ void testCountIsNotPushedDownForRowTtlTable() throws Exception {
+ String tableName = "test_count_table_with_row_ttl";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " id int not null,"
+ + " address varchar,"
+ + " name varchar,"
+ + " primary key (id) NOT ENFORCED)"
+ + " with ('bucket.num' = '4',"
+ + " 'table.kv.ttl' = '1 h')",
+ tableName))
+ .await();
+
+ String query = String.format("SELECT COUNT(*) FROM %s", tableName);
+ assertThat(tEnv.explainSql(query))
+ .contains("HashAggregate", "TableSourceScan")
+ .doesNotContain("aggregates=[grouping=[]");
+ CloseableIterator<Row> iterRows = tEnv.executeSql(query).collect();
+ assertThat(collectRowsWithTimeout(iterRows, 1))
+ .isEqualTo(Collections.singletonList("+I[0]"));
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testCountPushDownForLogTable(boolean partitionTable) throws Exception {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
index 71a81d3a0..bbb4aa3b1 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
@@ -23,6 +23,7 @@ import org.apache.fluss.metadata.KvFormat;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.DataTypeChecks;
import org.apache.fluss.types.DataTypes;
@@ -359,6 +360,46 @@ public class FlinkConversionsTest {
assertThat(flussTable.getCustomProperties()).containsExactlyEntriesOf(customProperties);
}
+ @Test
+ void testKvValueLayoutVersionIsExposed() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(ConfigOptions.TABLE_KV_FORMAT.key(), "compacted");
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(
+ org.apache.fluss.metadata.Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .primaryKey("id")
+ .build())
+ .distributedBy(1, "id")
+ .properties(properties)
+ .customProperty("client.custom-option", "custom-value")
+ .build();
+ long currentMillis = System.currentTimeMillis();
+ TableInfo tableInfo =
+ TableInfo.of(
+ TablePath.of("db", "table"),
+ 1L,
+ 1,
+ descriptor,
+ DEFAULT_REMOTE_DATA_DIR,
+ currentMillis,
+ currentMillis);
+
+ CatalogTable flinkTable = (CatalogTable) FlinkConversions.toFlinkTable(tableInfo);
+
+ assertThat(flinkTable.getOptions())
+ .containsEntry(ConfigOptions.TABLE_KV_FORMAT.key(), "compacted")
+ .containsEntry("client.custom-option", "custom-value")
+ .containsEntry(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+ }
+
@Test
void testOptionConversions() {
ConfigOption<?> flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_KV_FORMAT);
@@ -400,6 +441,15 @@ public class FlinkConversionsTest {
.withDescription(
ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE
.description()));
+
+ flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_KV_TTL);
+ assertThat(flinkOption)
+ .isEqualTo(
+ org.apache.flink.configuration.ConfigOptions.key(
+ ConfigOptions.TABLE_KV_TTL.key())
+ .stringType()
+ .noDefaultValue()
+ .withDescription(ConfigOptions.TABLE_KV_TTL.description()));
}
@Test
diff --git a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufBytesField.java b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufBytesField.java
index 0cb0c97cb..d6e7e13f2 100644
--- a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufBytesField.java
+++ b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufBytesField.java
@@ -30,6 +30,7 @@ public class ProtobufBytesField extends ProtobufField<Field.Bytes> {
@Override
public void declaration(PrintWriter w) {
w.format("private byte[] %s = null;\n", ccName);
+ w.format("private int _%sOffset = 0;\n", ccName);
w.format("private int _%sLen = -1;\n", ccName);
}
@@ -38,6 +39,7 @@ public class ProtobufBytesField extends ProtobufField<Field.Bytes> {
w.format("_%sLen = ProtoCodecUtils.readBytesLen(_buffer);\n", ccName);
w.format("%s = new byte[_%sLen];\n", ccName, ccName);
w.format("_buffer.readBytes(%s);\n", ccName);
+ w.format("_%sOffset = 0;\n", ccName);
}
@Override
@@ -53,9 +55,20 @@ public class ProtobufBytesField extends ProtobufField<Field.Bytes> {
w.format(
"public %s %s(byte[] %s) {\n",
enclosingType, ProtoGenUtil.camelCase("set", ccName), ccName);
+ w.format(
+ " return %s(%s, 0, %s.length);\n",
+ ProtoGenUtil.camelCase("set", ccName), ccName, ccName);
+ w.format("}\n");
+ w.println();
+
+ // set byte[] slice
+ w.format(
+ "public %s %s(byte[] %s, int offset, int length) {\n",
+ enclosingType, ProtoGenUtil.camelCase("set", ccName), ccName);
w.format(" this.%s = %s;\n", ccName, ccName);
+ w.format(" _%sOffset = offset;\n", ccName);
w.format(" _bitField%d |= %s;\n", bitFieldIndex(), fieldMask());
- w.format(" _%sLen = %s.length;\n", ccName, ccName);
+ w.format(" _%sLen = length;\n", ccName);
w.format(" _cachedSize = -1;\n");
w.format(" return this;\n");
w.format("}\n");
@@ -75,13 +88,21 @@ public class ProtobufBytesField extends ProtobufField<Field.Bytes> {
// get byte[]
w.format("public byte[] %s() {\n", ProtoGenUtil.camelCase("get", ccName));
- w.format(" return %s;\n", ccName);
+ w.format(
+ " if (%s == null || (_%sOffset == 0 && _%sLen == %s.length)) {\n",
+ ccName, ccName, ccName, ccName);
+ w.format(" return %s;\n", ccName);
+ w.format(" }\n");
+ w.format(
+ " return java.util.Arrays.copyOfRange(%s, _%sOffset, _%sOffset + _%sLen);\n",
+ ccName, ccName, ccName, ccName);
w.format("}\n");
}
@Override
public void clear(PrintWriter w) {
w.format("%s = null;\n", ccName);
+ w.format("_%sOffset = 0;\n", ccName);
w.format("_%sLen = -1;\n", ccName);
}
@@ -95,7 +116,7 @@ public class ProtobufBytesField extends ProtobufField<Field.Bytes> {
public void serialize(PrintWriter w) {
w.format("_w.writeVarInt(%s);\n", tagName());
w.format("_w.writeVarInt(_%sLen);\n", ccName);
- w.format("_w.writeByteArray(%s, 0, _%sLen);\n", ccName, ccName);
+ w.format("_w.writeByteArray(%s, _%sOffset, _%sLen);\n", ccName, ccName, ccName);
}
@Override
diff --git a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufRepeatedBytesField.java b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufRepeatedBytesField.java
index fbe2bd0f3..100123240 100644
--- a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufRepeatedBytesField.java
+++ b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufRepeatedBytesField.java
@@ -77,7 +77,12 @@ public class ProtobufRepeatedBytesField extends ProtobufAbstractRepeated<Field.B
" throw new IndexOutOfBoundsException(\"Index \" + idx + \" is out of the list size (\" + _%sCount + \") for field '%s'\");\n",
pluralName, field.getName());
w.format(" }\n");
- w.format(" return %s.get(idx).b;\n", pluralName);
+ w.format(" ProtoCodecUtils.BytesHolder _bh = %s.get(idx);\n", pluralName);
+ w.format(" if (_bh.offset == 0 && _bh.len == _bh.b.length) {\n");
+ w.format(" return _bh.b;\n");
+ w.format(" }\n");
+ w.format(
+ " return java.util.Arrays.copyOfRange(_bh.b, _bh.offset, _bh.offset + _bh.len);\n");
w.format("}\n");
}
@@ -87,7 +92,7 @@ public class ProtobufRepeatedBytesField extends ProtobufAbstractRepeated<Field.B
w.format(" ProtoCodecUtils.BytesHolder _bh = %s.get(i);\n", pluralName);
w.format(" _w.writeVarInt(%s);\n", tagName());
w.format(" _w.writeVarInt(_bh.len);\n");
- w.format(" _w.writeByteArray(_bh.b, 0, _bh.len);\n");
+ w.format(" _w.writeByteArray(_bh.b, _bh.offset, _bh.len);\n");
w.format("}\n");
}
@@ -97,17 +102,23 @@ public class ProtobufRepeatedBytesField extends ProtobufAbstractRepeated<Field.B
w.format(
"public void %s(byte[] %s) {\n",
ProtoGenUtil.camelCase("add", singularName), singularName);
- w.format(" if (%s == null) {\n", pluralName);
w.format(
- " %s = new java.util.ArrayList<ProtoCodecUtils.BytesHolder>();\n",
- pluralName);
- w.format(" }\n");
+ " %s(%s, 0, %s.length);\n",
+ ProtoGenUtil.camelCase("add", singularName), singularName, singularName);
+ w.format("}\n");
+ w.println();
+
+ // add byte[] slice
+ w.format(
+ "public void %s(byte[] %s, int offset, int length) {\n",
+ ProtoGenUtil.camelCase("add", singularName), singularName);
w.format(
" ProtoCodecUtils.BytesHolder _bh = _%sBytesHolder();\n",
ProtoGenUtil.camelCase("new", singularName));
w.format(" _cachedSize = -1;\n");
w.format(" _bh.b = %s;\n", singularName);
- w.format(" _bh.len = %s.length;\n", singularName);
+ w.format(" _bh.offset = offset;\n");
+ w.format(" _bh.len = length;\n");
w.format("}\n");
w.println();
diff --git a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java
index b2ccc1187..2e84f27bc 100644
--- a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java
+++ b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/BytesTest.java
@@ -55,10 +55,12 @@ public class BytesTest {
@Test
public void testBytes() throws Exception {
- B lpb = new B().setPayload(new byte[] {1, 2, 3, 4, 5});
+ byte[] payload = new byte[] {1, 2, 3, 4, 5};
+ B lpb = new B().setPayload(payload);
assertThat(lpb.getPayloadSize()).isEqualTo(5);
assertThat(lpb.getPayload()).isEqualTo(new byte[] {1, 2, 3, 4, 5});
+ assertThat(lpb.getPayload()).isSameAs(payload);
// test binary equals to protobuf
Bytes.B pbb =
@@ -101,6 +103,32 @@ public class BytesTest {
assertThat(b3).isEqualTo(expected);
}
+ @Test
+ public void testBytesSlice() throws Exception {
+ byte[] backing = new byte[] {9, 9, 1, 2, 3, 9};
+ B message = new B().setPayload(backing, 2, 3);
+
+ assertThat(message.getPayloadSize()).isEqualTo(3);
+ assertThat(message.getPayload()).containsExactly(1, 2, 3);
+
+ message.writeTo(bb1);
+ B parsed = new B();
+ parsed.parseFrom(bb1, bb1.readableBytes());
+ assertThat(parsed.getPayload()).containsExactly(1, 2, 3);
+ }
+
+ @Test
+ public void testUnsetAndClearedPayloadReturnsNull() {
+ B message = new B();
+
+ assertThat(message.getPayload()).isNull();
+
+ message.setPayload(new byte[] {1});
+ message.clearPayload();
+
+ assertThat(message.getPayload()).isNull();
+ }
+
@Test
public void testRecordsLazilyParsed() {
RD rd = new RD();
@@ -243,14 +271,16 @@ public class BytesTest {
@Test
public void testRepeatedBytes() throws Exception {
+ byte[] firstExtraItem = new byte[] {1, 2, 3};
B lpb = new B();
- lpb.addExtraItem(new byte[] {1, 2, 3});
+ lpb.addExtraItem(firstExtraItem);
lpb.addExtraItem(new byte[] {4, 5, 6, 7});
assertThat(lpb.getExtraItemsCount()).isEqualTo(2);
assertThat(lpb.getExtraItemSizeAt(0)).isEqualTo(3);
assertThat(lpb.getExtraItemSizeAt(1)).isEqualTo(4);
assertThat(lpb.getExtraItemAt(0)).isEqualTo(new byte[] {1, 2, 3});
+ assertThat(lpb.getExtraItemAt(0)).isSameAs(firstExtraItem);
assertThat(lpb.getExtraItemAt(1)).isEqualTo(new byte[] {4, 5, 6, 7});
Bytes.B pbb =
@@ -276,4 +306,13 @@ public class BytesTest {
assertThat(parsed.getExtraItemAt(0)).isEqualTo(new byte[] {1, 2, 3});
assertThat(parsed.getExtraItemAt(1)).isEqualTo(new byte[] {4, 5, 6, 7});
}
+
+ @Test
+ public void testRepeatedBytesSlice() throws Exception {
+ byte[] backing = new byte[] {9, 1, 2, 3, 9};
+ B message = new B();
+ message.addExtraItem(backing, 1, 3);
+
+ assertThat(message.getExtraItemAt(0)).containsExactly(1, 2, 3);
+ }
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/LookupResultForBucket.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/LookupResultForBucket.java
index 08b153d66..b81c66083 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/LookupResultForBucket.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/LookupResultForBucket.java
@@ -18,6 +18,7 @@
package org.apache.fluss.rpc.entity;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.messages.LookupRequest;
import org.apache.fluss.rpc.protocol.ApiError;
@@ -25,33 +26,66 @@ import javax.annotation.Nullable;
import java.util.List;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
/** Result of {@link LookupRequest} for each table bucket. */
public class LookupResultForBucket extends ResultForBucket {
private final List<byte[]> values;
+ @Nullable private final KvValueLayout kvValueLayout;
+ @Nullable private final List<KvValueLayout> kvValueLayouts;
/** Identifies the original partition for historical lookup; null for normal lookup. */
private final @Nullable String originalPartitionName;
- public LookupResultForBucket(TableBucket tableBucket, List<byte[]> values) {
- this(tableBucket, values, null, ApiError.NONE);
+ public LookupResultForBucket(
+ TableBucket tableBucket, List<byte[]> values, KvValueLayout kvValueLayout) {
+ this(tableBucket, values, kvValueLayout, null, null, ApiError.NONE);
}
public LookupResultForBucket(TableBucket tableBucket, ApiError error) {
- this(tableBucket, null, null, error);
+ this(tableBucket, null, null, null, null, error);
+ }
+
+ /** Creates a successful historical lookup result. */
+ public LookupResultForBucket(
+ TableBucket tableBucket,
+ List<byte[]> values,
+ KvValueLayout kvValueLayout,
+ String originalPartitionName) {
+ this(tableBucket, values, kvValueLayout, null, originalPartitionName, ApiError.NONE);
+ }
+
+ /** Creates a successful historical lookup result whose values use different layouts. */
+ public LookupResultForBucket(
+ TableBucket tableBucket,
+ List<byte[]> values,
+ List<KvValueLayout> kvValueLayouts,
+ String originalPartitionName) {
+ this(tableBucket, values, null, kvValueLayouts, originalPartitionName, ApiError.NONE);
}
- /**
- * Creates a lookup result with the original partition name used to identify a historical lookup
- * response.
- */
+ /** Creates a failed historical lookup result. */
public LookupResultForBucket(
+ TableBucket tableBucket, String originalPartitionName, ApiError error) {
+ this(tableBucket, null, null, null, originalPartitionName, error);
+ }
+
+ private LookupResultForBucket(
TableBucket tableBucket,
List<byte[]> values,
+ @Nullable KvValueLayout kvValueLayout,
+ @Nullable List<KvValueLayout> kvValueLayouts,
@Nullable String originalPartitionName,
ApiError error) {
super(tableBucket, error);
+ checkArgument(
+ kvValueLayouts == null
+ || (values != null && kvValueLayouts.size() == values.size()),
+ "A layout must be provided for every lookup value.");
this.values = values;
+ this.kvValueLayout = kvValueLayout;
+ this.kvValueLayouts = kvValueLayouts;
this.originalPartitionName = originalPartitionName;
}
@@ -63,4 +97,15 @@ public class LookupResultForBucket extends ResultForBucket {
public @Nullable String originalPartitionName() {
return originalPartitionName;
}
+
+ /** Returns the shared value layout, or null for a failed lookup or mixed-layout values. */
+ @Nullable
+ public KvValueLayout getKvValueLayout() {
+ return kvValueLayout;
+ }
+
+ /** Returns the physical layout of the value at the given index. */
+ public KvValueLayout getKvValueLayout(int valueIndex) {
+ return kvValueLayouts == null ? kvValueLayout : kvValueLayouts.get(valueIndex);
+ }
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PrefixLookupResultForBucket.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PrefixLookupResultForBucket.java
index 517563b63..d423f3fbd 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PrefixLookupResultForBucket.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PrefixLookupResultForBucket.java
@@ -18,31 +18,46 @@
package org.apache.fluss.rpc.entity;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.messages.PrefixLookupRequest;
import org.apache.fluss.rpc.protocol.ApiError;
+import javax.annotation.Nullable;
+
import java.util.List;
/** The Result of {@link PrefixLookupRequest} for each table bucket. */
public class PrefixLookupResultForBucket extends ResultForBucket {
private final List<List<byte[]>> values;
+ @Nullable private final KvValueLayout kvValueLayout;
- public PrefixLookupResultForBucket(TableBucket tableBucket, List<List<byte[]>> values) {
- this(tableBucket, values, ApiError.NONE);
+ public PrefixLookupResultForBucket(
+ TableBucket tableBucket, List<List<byte[]>> values, KvValueLayout kvValueLayout) {
+ this(tableBucket, values, kvValueLayout, ApiError.NONE);
}
public PrefixLookupResultForBucket(TableBucket tableBucket, ApiError error) {
- this(tableBucket, null, error);
+ this(tableBucket, null, null, error);
}
private PrefixLookupResultForBucket(
- TableBucket tableBucket, List<List<byte[]>> values, ApiError error) {
+ TableBucket tableBucket,
+ List<List<byte[]>> values,
+ @Nullable KvValueLayout kvValueLayout,
+ ApiError error) {
super(tableBucket, error);
this.values = values;
+ this.kvValueLayout = kvValueLayout;
}
public List<List<byte[]>> prefixLookupValues() {
return values;
}
+
+ /** Returns the storage layout for a successful prefix lookup result. */
+ @Nullable
+ public KvValueLayout getKvValueLayout() {
+ return kvValueLayout;
+ }
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index 979674c25..baf425665 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -65,7 +65,7 @@ public enum ApiKeys {
LIST_OFFSETS(1021, 0, 0, PUBLIC),
COMMIT_KV_SNAPSHOT(1022, 0, 0, PRIVATE),
GET_LATEST_KV_SNAPSHOTS(1023, 0, 0, PUBLIC),
- GET_KV_SNAPSHOT_METADATA(1024, 0, 0, PUBLIC),
+ GET_KV_SNAPSHOT_METADATA(1024, 0, 1, PUBLIC),
GET_FILESYSTEM_SECURITY_TOKEN(1025, 0, 0, PUBLIC),
INIT_WRITER(1026, 0, 0, PUBLIC),
COMMIT_REMOTE_LOG_MANIFEST(1027, 0, 0, PRIVATE),
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java
index 0ef7ad22e..bcaf1ad0b 100644
--- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java
@@ -38,4 +38,9 @@ class ApiKeysTest {
assertThat(api.highestSupportedVersion).isGreaterThanOrEqualTo((short) 0);
}
}
+
+ @Test
+ void testSnapshotMetadataSupportsLayoutAwareClient() {
+ assertThat(ApiKeys.GET_KV_SNAPSHOT_METADATA.highestSupportedVersion).isEqualTo((short) 1);
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
index a534a3551..e2ef1094e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -29,6 +29,7 @@ import org.apache.fluss.exception.PartitionNotExistException;
import org.apache.fluss.exception.SecurityDisabledException;
import org.apache.fluss.exception.SecurityTokenException;
import org.apache.fluss.exception.TableNotPartitionedException;
+import org.apache.fluss.exception.UnsupportedVersionException;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.fs.token.ObtainedSecurityToken;
import org.apache.fluss.metadata.DatabaseInfo;
@@ -38,6 +39,7 @@ import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.RpcGatewayService;
import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway;
import org.apache.fluss.rpc.messages.ApiVersionsRequest;
@@ -184,6 +186,9 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
public abstract void authorizeTable(OperationType operationType, long tableId);
+ /** Returns table information for a table id known by the concrete server role. */
+ protected abstract TableInfo getTableInfo(long tableId);
+
public void authorizeDatabase(OperationType operationType, String databaseName) {
if (authorizer != null) {
authorizer.authorize(currentSession(), operationType, Resource.database(databaseName));
@@ -382,7 +387,6 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
+ tablePath
+ "' is a partitioned table, but partition name is not provided.");
}
-
try {
// get table id
long tableId = tableInfo.getTableId();
@@ -425,6 +429,8 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
GetKvSnapshotMetadataRequest request) {
long tableId = request.getTableId();
authorizeTable(OperationType.DESCRIBE, tableId);
+ TableInfo tableInfo = getTableInfo(tableId);
+ validateKvSnapshotMetadataVersion(currentSession().getApiVersion(), tableInfo);
TableBucket tableBucket =
new TableBucket(
@@ -450,6 +456,17 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
}
}
+ static void validateKvSnapshotMetadataVersion(short apiVersion, TableInfo tableInfo) {
+ if (apiVersion < 1
+ && KvValueLayout.fromTableConfig(tableInfo.getTableConfig()).hasValueTag()) {
+ throw new UnsupportedVersionException(
+ String.format(
+ "Client API version %d cannot read tagged KV snapshots for table '%s'. "
+ + "Please upgrade your Fluss client to a newer version.",
+ apiVersion, tableInfo.getTablePath()));
+ }
+ }
+
@Override
public CompletableFuture<GetFileSystemSecurityTokenResponse> getFileSystemSecurityToken(
GetFileSystemSecurityTokenRequest request) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index 04119bb0f..928f0af99 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -29,7 +29,6 @@ import org.apache.fluss.config.cluster.AlterConfigOpType;
import org.apache.fluss.exception.ApiException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.InvalidAlterTableException;
-import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.InvalidDatabaseException;
import org.apache.fluss.exception.InvalidPartitionException;
@@ -59,6 +58,7 @@ import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.messages.AcquireKvSnapshotLeaseRequest;
import org.apache.fluss.rpc.messages.AcquireKvSnapshotLeaseResponse;
@@ -372,13 +372,20 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
return tablePath;
}
- private void validateKvTable(long tableId) {
+ @Override
+ protected TableInfo getTableInfo(long tableId) {
+ TablePath tablePath = getTablePathById(tableId);
+ return metadataManager.getTable(tablePath);
+ }
+
+ private TableInfo validateKvTable(long tableId) {
TablePath tablePath = getTablePathById(tableId);
TableInfo tableInfo = metadataManager.getTable(tablePath);
if (!tableInfo.hasPrimaryKey()) {
throw new NonPrimaryKeyTableException(
"Table '" + tablePath + "' is not a primary key table");
}
+ return tableInfo;
}
@Override
@@ -462,9 +469,9 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
tablePath.validate();
authorizeDatabase(OperationType.CREATE, tablePath.getDatabaseName());
- TableDescriptor tableDescriptor;
+ TableDescriptor requestedDescriptor;
try {
- tableDescriptor = TableDescriptor.fromJsonBytes(request.getTableJson());
+ requestedDescriptor = TableDescriptor.fromJsonBytes(request.getTableJson());
} catch (Exception e) {
if (e instanceof UncheckedIOException) {
throw new InvalidTableException(
@@ -479,14 +486,17 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
lakeCatalogDynamicLoader.getLakeCatalogContainer();
// Check table creation permissions based on table type
- validateTableCreationPermission(tableDescriptor, tablePath);
+ validateTableCreationPermission(requestedDescriptor, tablePath);
// apply system defaults if the config is not set
- tableDescriptor = applySystemDefaults(tableDescriptor, lakeCatalogContainer);
+ TableDescriptor resolvedDescriptor =
+ applySystemDefaults(requestedDescriptor, lakeCatalogContainer);
// validate table descriptor before creating table in lake or fluss metadata,
// to avoid orphaned lake tables when validation fails
- metadataManager.validateTableDescriptor(tableDescriptor);
+ metadataManager.validateTableDescriptor(resolvedDescriptor);
+
+ TableDescriptor tableDescriptor = resolvedDescriptor;
// the distribution and bucket count must be set now
//noinspection OptionalGetWithoutIsPresent
@@ -812,23 +822,23 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
if (newDescriptor.hasPrimaryKey()) {
Map<String, String> newProperties = new HashMap<>(newDescriptor.getProperties());
- Integer formatVersion =
- Configuration.fromMap(newProperties).get(ConfigOptions.TABLE_KV_FORMAT_VERSION);
+ Configuration newTableConf = Configuration.fromMap(newProperties);
+ Integer formatVersion = newTableConf.get(ConfigOptions.TABLE_KV_FORMAT_VERSION);
+ // The option has no default so that its absence continues to identify legacy tables.
if (formatVersion == null) {
- // set current kv format version for default
newProperties.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
- } else {
- if (formatVersion > CURRENT_KV_FORMAT_VERSION) {
- throw new InvalidConfigException(
- String.format(
- "Unsupported kv format version %d. "
- + "The maximum supported version is %d.",
- formatVersion, CURRENT_KV_FORMAT_VERSION));
- }
}
+ int layoutVersion =
+ newTableConf.getOptional(ConfigOptions.TABLE_KV_TTL).isPresent()
+ ? KvValueLayout.TAGGED.version()
+ : KvValueLayout.PLAIN.version();
+ newProperties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(layoutVersion));
+
// Enable standby replica for new PK tables if not explicitly configured
if (!newProperties.containsKey(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key())) {
newProperties.put(ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(), "true");
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
index 84273ab38..9f00156e2 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
@@ -49,6 +49,8 @@ import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocator;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocatorUtil;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.FlussPaths;
+import org.apache.fluss.utils.clock.Clock;
+import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.types.Tuple2;
import org.rocksdb.RateLimiter;
@@ -120,6 +122,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
private final ZooKeeperClient zkClient;
+ private final Clock clock;
+
private final Map<TableBucket, KvTablet> currentKvs = new ConcurrentHashMap<>();
/**
@@ -154,7 +158,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
int recoveryThreadsPerDataDir,
LogManager logManager,
TabletServerMetricGroup tabletServerMetricGroup,
- @Nullable KvFlushScheduler kvFlushScheduler)
+ @Nullable KvFlushScheduler kvFlushScheduler,
+ Clock clock)
throws IOException {
super(TabletType.KV, localDiskManager.dataDirs(), conf, recoveryThreadsPerDataDir);
this.localDiskManager = localDiskManager;
@@ -162,6 +167,7 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
this.arrowBufferAllocator = BufferAllocatorUtil.createBufferAllocator(null);
this.memorySegmentPool = LazyMemorySegmentPool.createServerBufferPool(conf);
this.zkClient = zkClient;
+ this.clock = clock;
this.remoteKvDir = FlussPaths.remoteKvDir(conf);
this.remoteFileSystem = remoteKvDir.getFileSystem();
this.serverMetricGroup = tabletServerMetricGroup;
@@ -197,7 +203,14 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
TabletServerMetricGroup tabletServerMetricGroup,
LocalDiskManager localDiskManager)
throws IOException {
- return create(conf, zkClient, logManager, tabletServerMetricGroup, localDiskManager, null);
+ return create(
+ conf,
+ zkClient,
+ logManager,
+ tabletServerMetricGroup,
+ localDiskManager,
+ null,
+ SystemClock.getInstance());
}
/**
@@ -214,6 +227,38 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
LocalDiskManager localDiskManager,
@Nullable KvFlushScheduler kvFlushScheduler)
throws IOException {
+ return create(
+ conf,
+ zkClient,
+ logManager,
+ tabletServerMetricGroup,
+ localDiskManager,
+ kvFlushScheduler,
+ SystemClock.getInstance());
+ }
+
+ public static KvManager create(
+ Configuration conf,
+ ZooKeeperClient zkClient,
+ LogManager logManager,
+ TabletServerMetricGroup tabletServerMetricGroup,
+ LocalDiskManager localDiskManager,
+ Clock clock)
+ throws IOException {
+ return create(
+ conf, zkClient, logManager, tabletServerMetricGroup, localDiskManager, null, clock);
+ }
+
+ @VisibleForTesting
+ public static KvManager create(
+ Configuration conf,
+ ZooKeeperClient zkClient,
+ LogManager logManager,
+ TabletServerMetricGroup tabletServerMetricGroup,
+ LocalDiskManager localDiskManager,
+ @Nullable KvFlushScheduler kvFlushScheduler,
+ Clock clock)
+ throws IOException {
return new KvManager(
localDiskManager,
conf,
@@ -221,7 +266,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
conf.getInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS),
logManager,
tabletServerMetricGroup,
- kvFlushScheduler);
+ kvFlushScheduler,
+ clock);
}
public void startup() {
@@ -320,7 +366,9 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
sharedRocksDBRateLimiter,
kvFlushScheduler,
flushCompleteListener,
- autoIncrementManager);
+ autoIncrementManager,
+ clock,
+ tableConfig);
currentKvs.put(tableBucket, tablet);
LOG.info(
@@ -412,7 +460,6 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
// TODO: we should support recover schema from disk to decouple put and schema.
TablePath tablePath = physicalTablePath.getTablePath();
TableInfo tableInfo = getTableInfo(zkClient, tablePath);
-
TableConfig tableConfig = tableInfo.getTableConfig();
RowMerger rowMerger =
RowMerger.create(tableConfig, tableConfig.getKvFormat(), schemaGetter);
@@ -440,7 +487,9 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
sharedRocksDBRateLimiter,
kvFlushScheduler,
flushCompleteListener,
- autoIncrementManager);
+ autoIncrementManager,
+ clock,
+ tableConfig);
if (this.currentKvs.containsKey(tableBucket)) {
throw new IllegalStateException(
String.format(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
index 780533c68..a3afeec4b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
@@ -23,6 +23,7 @@ import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.LogRecord;
import org.apache.fluss.record.LogRecordBatch;
@@ -80,6 +81,8 @@ public class KvRecoverHelper {
private KeyEncoder keyEncoder;
private RowEncoder rowEncoder;
+ private final ValueEncoder valueEncoder;
+ @Nullable private final RowTtlTimestampProvider rowTtlTimestampProvider;
private final SchemaGetter schemaGetter;
private InternalRow.FieldGetter[] currentFieldGetters;
@@ -106,6 +109,8 @@ public class KvRecoverHelper {
this.kvFormat = kvFormat;
this.logFormat = logFormat;
this.schemaGetter = schemaGetter;
+ this.valueEncoder = kvTablet.getValueEncoder();
+ this.rowTtlTimestampProvider = kvTablet.getRowTtlTimestampProvider();
this.remoteLogFetcher = remoteLogFetcher;
this.historicalPartition = historicalPartition;
}
@@ -267,6 +272,9 @@ public class KvRecoverHelper {
AutoIncIDRangeUpdater autoIncIdRangeUpdater,
ThrowingConsumer<KeyValueAndLogOffset, Exception> resumeRecordConsumer)
throws Exception {
+ if (rowTtlTimestampProvider != null) {
+ rowTtlTimestampProvider.prepareForBatch(logRecordBatch.commitTimestamp());
+ }
try (CloseableIterator<LogRecord> logRecordIter = logRecordBatch.records(readContext)) {
while (logRecordIter.hasNext()) {
LogRecord logRecord = logRecordIter.next();
@@ -288,7 +296,9 @@ public class KvRecoverHelper {
// the log row format may not compatible with kv row format,
// e.g, arrow vs. compacted, thus needs a conversion here.
BinaryRow row = toKvRow(logRow);
- value = ValueEncoder.encodeValue(currentSchemaId.shortValue(), row);
+ value =
+ valueEncoder.encodeValue(
+ new BinaryValue(currentSchemaId.shortValue(), row));
}
resumeRecordConsumer.accept(
new KeyValueAndLogOffset(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
index 8fae46559..44027f3f3 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
@@ -21,6 +21,7 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.compression.ArrowCompressionInfo;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.KvStorageException;
import org.apache.fluss.exception.StorageBackpressureException;
@@ -34,6 +35,9 @@ import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.KvRecordBatch;
import org.apache.fluss.row.arrow.ArrowWriterPool;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.row.encode.ValueDecoder;
+import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.rpc.protocol.MergeMode;
import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
import org.apache.fluss.server.kv.autoinc.AutoIncrementManager;
@@ -59,7 +63,11 @@ import org.apache.fluss.server.utils.ResourceGuard;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocator;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.IOUtils;
+import org.apache.fluss.utils.clock.Clock;
+import org.apache.fluss.utils.clock.SystemClock;
+import org.rocksdb.AbstractCompactionFilter;
+import org.rocksdb.AbstractCompactionFilterFactory;
import org.rocksdb.RateLimiter;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksIterator;
@@ -73,10 +81,13 @@ import javax.annotation.concurrent.ThreadSafe;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -133,6 +144,10 @@ public final class KvTablet {
// A lock that guards all modifications to the kv.
private final ReadWriteLock kvLock = new ReentrantReadWriteLock();
+ private final KvValueLayout kvValueLayout;
+ private final ValueEncoder valueEncoder;
+ @Nullable private final RowTtlTimestampProvider rowTtlTimestampProvider;
+ private final boolean rowTtlEnabled;
private final AutoIncrementManager autoIncrementManager;
// RocksDB statistics accessor for this tablet
@@ -178,11 +193,17 @@ public final class KvTablet {
ArrowCompressionInfo arrowCompressionInfo,
SchemaGetter schemaGetter,
ChangelogImage changelogImage,
+ KvValueLayout kvValueLayout,
+ ValueEncoder valueEncoder,
+ ValueDecoder valueDecoder,
@Nullable RocksDBStatistics rocksDBStatistics,
KvFlushScheduler kvFlushScheduler,
boolean closeFlushScheduler,
@Nullable Runnable flushCompleteListener,
- AutoIncrementManager autoIncrementManager) {
+ AutoIncrementManager autoIncrementManager,
+ @Nullable RowTtlTimestampProvider rowTtlTimestampProvider,
+ Clock clock,
+ boolean rowTtlEnabled) {
this.physicalPath = physicalPath;
this.tableBucket = tableBucket;
this.historicalPartition =
@@ -197,6 +218,10 @@ public final class KvTablet {
this.kvPreWriteBuffer = new KvPreWriteBuffer(serverMetricGroup);
this.kvStateAccessor =
new KvStateAccessor(kvPreWriteBuffer, rocksDBKv, historicalPartition);
+ this.kvValueLayout = kvValueLayout;
+ this.valueEncoder = valueEncoder;
+ this.rowTtlTimestampProvider = rowTtlTimestampProvider;
+ this.rowTtlEnabled = rowTtlEnabled;
this.kvWriteProcessor =
new KvWriteProcessor(
tableBucket,
@@ -208,7 +233,11 @@ public final class KvTablet {
arrowCompressionInfo,
schemaGetter,
changelogImage,
- autoIncrementManager);
+ autoIncrementManager,
+ valueEncoder,
+ valueDecoder,
+ rowTtlTimestampProvider,
+ clock);
this.rocksDBStatistics = rocksDBStatistics;
this.autoIncrementManager = autoIncrementManager;
this.flushCompleteListener = flushCompleteListener;
@@ -216,7 +245,7 @@ public final class KvTablet {
// Historical state only contains the WAL tail that has not been tiered to lake, so it
// cannot maintain a table-level row count.
this.rowCount =
- historicalPartition || changelogImage == ChangelogImage.WAL
+ historicalPartition || changelogImage == ChangelogImage.WAL || rowTtlEnabled
? ROW_COUNT_DISABLED
: 0L;
}
@@ -265,7 +294,9 @@ public final class KvTablet {
new KvFlushScheduler(serverConf),
true,
null,
- autoIncrementManager);
+ autoIncrementManager,
+ SystemClock.getInstance(),
+ new TableConfig(new Configuration()));
}
public static KvTablet create(
@@ -285,7 +316,9 @@ public final class KvTablet {
RateLimiter sharedRateLimiter,
KvFlushScheduler kvFlushScheduler,
@Nullable Runnable flushCompleteListener,
- AutoIncrementManager autoIncrementManager)
+ AutoIncrementManager autoIncrementManager,
+ Clock clock,
+ TableConfig tableConfig)
throws IOException {
return create(
tablePath,
@@ -305,7 +338,9 @@ public final class KvTablet {
kvFlushScheduler,
false,
flushCompleteListener,
- autoIncrementManager);
+ autoIncrementManager,
+ clock,
+ tableConfig);
}
private static KvTablet create(
@@ -326,9 +361,33 @@ public final class KvTablet {
KvFlushScheduler kvFlushScheduler,
boolean closeFlushScheduler,
@Nullable Runnable flushCompleteListener,
- AutoIncrementManager autoIncrementManager)
+ AutoIncrementManager autoIncrementManager,
+ Clock clock,
+ TableConfig tableConfig)
throws IOException {
- RocksDBKv kv = buildRocksDBKv(serverConf, kvTabletDir, sharedRateLimiter);
+ checkNotNull(tableConfig, "tableConfig must not be null.");
+ Optional<Duration> rowTtl = tableConfig.getKvTTL();
+ KvValueLayout kvValueLayout = KvValueLayout.fromTableConfig(tableConfig);
+ @Nullable
+ RowTtlTimestampProvider rowTtlTimestampProvider =
+ kvValueLayout.hasValueTag()
+ ? RowTtlTimestampProvider.create(
+ tableConfig, schemaGetter, ZoneId.systemDefault())
+ : null;
+ ValueEncoder valueEncoder =
+ rowTtlTimestampProvider == null
+ ? ValueEncoder.forLayout(kvValueLayout)
+ : ValueEncoder.forLayout(kvValueLayout, rowTtlTimestampProvider);
+ ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, kvFormat, kvValueLayout);
+ @Nullable
+ AbstractCompactionFilterFactory<? extends AbstractCompactionFilter<?>>
+ compactionFilterFactory =
+ rowTtl.isPresent()
+ ? RowTtlCompactionFilterFactory.create(
+ kvValueLayout, rowTtl.get(), clock)
+ : null;
+ RocksDBKv kv =
+ buildRocksDBKv(serverConf, kvTabletDir, sharedRateLimiter, compactionFilterFactory);
// Create RocksDB statistics accessor (will be registered to TableMetricGroup by Replica)
// Pass ResourceGuard to ensure thread-safe access during concurrent close operations
@@ -357,28 +416,107 @@ public final class KvTablet {
arrowCompressionInfo,
schemaGetter,
changelogImage,
+ kvValueLayout,
+ valueEncoder,
+ valueDecoder,
rocksDBStatistics,
kvFlushScheduler,
closeFlushScheduler,
flushCompleteListener,
- autoIncrementManager);
+ autoIncrementManager,
+ rowTtlTimestampProvider,
+ clock,
+ rowTtl.isPresent());
+ }
+
+ public static KvTablet create(
+ PhysicalTablePath tablePath,
+ TableBucket tableBucket,
+ LogTablet logTablet,
+ File kvTabletDir,
+ Configuration serverConf,
+ TabletServerMetricGroup serverMetricGroup,
+ BufferAllocator arrowBufferAllocator,
+ MemorySegmentPool memorySegmentPool,
+ KvFormat kvFormat,
+ RowMerger rowMerger,
+ ArrowCompressionInfo arrowCompressionInfo,
+ SchemaGetter schemaGetter,
+ ChangelogImage changelogImage,
+ RateLimiter sharedRateLimiter,
+ AutoIncrementManager autoIncrementManager,
+ Clock clock,
+ TableConfig tableConfig)
+ throws IOException {
+ return create(
+ tablePath,
+ tableBucket,
+ logTablet,
+ kvTabletDir,
+ serverConf,
+ serverMetricGroup,
+ arrowBufferAllocator,
+ memorySegmentPool,
+ kvFormat,
+ rowMerger,
+ arrowCompressionInfo,
+ schemaGetter,
+ changelogImage,
+ sharedRateLimiter,
+ new KvFlushScheduler(serverConf),
+ true,
+ null,
+ autoIncrementManager,
+ clock,
+ tableConfig);
}
private static RocksDBKv buildRocksDBKv(
- Configuration configuration, File kvDir, RateLimiter sharedRateLimiter)
+ Configuration configuration,
+ File kvDir,
+ RateLimiter sharedRateLimiter,
+ @Nullable
+ AbstractCompactionFilterFactory<? extends AbstractCompactionFilter<?>>
+ compactionFilterFactory)
throws IOException {
- // Enable statistics to support RocksDB statistics collection
- RocksDBResourceContainer rocksDBResourceContainer =
- new RocksDBResourceContainer(configuration, kvDir, true, sharedRateLimiter);
- RocksDBKvBuilder rocksDBKvBuilder =
- new RocksDBKvBuilder(
- kvDir,
- rocksDBResourceContainer,
- rocksDBResourceContainer.getColumnOptions())
- .setFlussL0SlowdownTrigger(
- configuration.get(
- ConfigOptions.KV_BACKPRESSURE_L0_SLOWDOWN_TRIGGER));
- return rocksDBKvBuilder.build();
+ @Nullable RocksDBResourceContainer rocksDBResourceContainer = null;
+ boolean resourcesOwnedByBuilder = false;
+ try {
+ rocksDBResourceContainer =
+ new RocksDBResourceContainer(configuration, kvDir, true, sharedRateLimiter);
+ RocksDBKvBuilder rocksDBKvBuilder =
+ new RocksDBKvBuilder(
+ kvDir,
+ rocksDBResourceContainer,
+ rocksDBResourceContainer.getColumnOptions())
+ .setFlussL0SlowdownTrigger(
+ configuration.get(
+ ConfigOptions.KV_BACKPRESSURE_L0_SLOWDOWN_TRIGGER));
+ if (compactionFilterFactory != null) {
+ rocksDBKvBuilder.setCompactionFilterFactory(compactionFilterFactory);
+ }
+ resourcesOwnedByBuilder = true;
+ return rocksDBKvBuilder.build();
+ } finally {
+ if (!resourcesOwnedByBuilder) {
+ IOUtils.closeQuietly(rocksDBResourceContainer);
+ IOUtils.closeQuietly(compactionFilterFactory);
+ }
+ }
+ }
+
+ ValueEncoder getValueEncoder() {
+ return valueEncoder;
+ }
+
+ @Nullable
+ RowTtlTimestampProvider getRowTtlTimestampProvider() {
+ return rowTtlTimestampProvider;
+ }
+
+ /** Returns the physical value layout used by this tablet. */
+ public KvValueLayout getKvValueLayout() {
+ return kvValueLayout;
}
public TableBucket getTableBucket() {
@@ -426,16 +564,24 @@ public final class KvTablet {
}
void setRowCount(long rowCount) {
- this.rowCount = rowCount;
+ if (this.rowCount != ROW_COUNT_DISABLED) {
+ this.rowCount = rowCount;
+ }
}
// row_count is volatile, so it's safe to read without lock
public long getRowCount() {
if (rowCount == ROW_COUNT_DISABLED) {
+ if (rowTtlEnabled) {
+ throw new InvalidTableException(
+ String.format(
+ "Row count is disabled for this table '%s' because row TTL cleanup does not maintain exact row count.",
+ getTablePath()));
+ }
throw new InvalidTableException(
String.format(
"Row count is disabled for this table '%s'. This usually happens when the table is"
- + "created before v0.9 or the changelog image is set to WAL, "
+ + " created before v0.9 or the changelog image is set to WAL, "
+ "as maintaining row count in WAL mode is costly and not necessary for most use cases. "
+ "If you want to enable row count, please set changelog image to FULL.",
getTablePath()));
@@ -1044,7 +1190,8 @@ public final class KvTablet {
lease,
limit,
capturedLogOffset,
- initialAccessTimeMs);
+ initialAccessTimeMs,
+ kvValueLayout);
success = true;
return new OpenScanResult(context, capturedLogOffset);
} finally {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java
index 76fc94e5c..bc0fb53ff 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java
@@ -38,7 +38,9 @@ import org.apache.fluss.record.KvRecordReadContext;
import org.apache.fluss.row.BinaryRow;
import org.apache.fluss.row.PaddingRow;
import org.apache.fluss.row.arrow.ArrowWriterProvider;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.row.encode.ValueDecoder;
+import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.rpc.protocol.MergeMode;
import org.apache.fluss.server.kv.autoinc.AutoIncrementManager;
import org.apache.fluss.server.kv.autoinc.AutoIncrementUpdater;
@@ -56,6 +58,7 @@ import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.ByteArrayWrapper;
import org.apache.fluss.utils.BytesUtils;
+import org.apache.fluss.utils.clock.Clock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -105,6 +108,11 @@ public final class KvWriteProcessor {
// the changelog image mode for this tablet
private final ChangelogImage changelogImage;
private final AutoIncrementManager autoIncrementManager;
+ private final ValueEncoder valueEncoder;
+ private final ValueDecoder valueDecoder;
+ private final ValueDecoder lakeValueDecoder;
+ @Nullable private final RowTtlTimestampProvider rowTtlTimestampProvider;
+ private final Clock clock;
/** Creates a KV write processor. */
public KvWriteProcessor(
@@ -117,7 +125,11 @@ public final class KvWriteProcessor {
ArrowCompressionInfo arrowCompressionInfo,
SchemaGetter schemaGetter,
ChangelogImage changelogImage,
- AutoIncrementManager autoIncrementManager) {
+ AutoIncrementManager autoIncrementManager,
+ ValueEncoder valueEncoder,
+ ValueDecoder valueDecoder,
+ @Nullable RowTtlTimestampProvider rowTtlTimestampProvider,
+ Clock clock) {
this.tableBucket = tableBucket;
this.logTablet = logTablet;
this.arrowWriterProvider = arrowWriterProvider;
@@ -132,6 +144,11 @@ public final class KvWriteProcessor {
this.schemaGetter = schemaGetter;
this.changelogImage = changelogImage;
this.autoIncrementManager = autoIncrementManager;
+ this.valueEncoder = valueEncoder;
+ this.valueDecoder = valueDecoder;
+ this.lakeValueDecoder = new ValueDecoder(schemaGetter, kvFormat, KvValueLayout.PLAIN);
+ this.rowTtlTimestampProvider = rowTtlTimestampProvider;
+ this.clock = clock;
}
/** Processes a KV batch against the supplied state and appends its WAL. */
@@ -155,6 +172,9 @@ public final class KvWriteProcessor {
long logEndOffsetOfPrevBatch = logTablet.localLogEndOffset();
try {
+ if (rowTtlTimestampProvider != null) {
+ rowTtlTimestampProvider.prepareForBatch(clock.milliseconds());
+ }
processKvRecords(
kvRecords,
kvRecords.schemaId(),
@@ -297,10 +317,9 @@ public final class KvWriteProcessor {
throws Exception {
long logOffset = startLogOffset;
- // TODO: reuse the read context and decoder
+ // TODO: reuse the read context
KvRecordBatch.ReadContext readContext =
KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
- ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, kvFormat);
for (KvRecord kvRecord : kvRecords.records(readContext)) {
byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
@@ -313,7 +332,6 @@ public final class KvWriteProcessor {
processDeletion(
key,
currentMerger,
- valueDecoder,
walBuilder,
latestSchemaRow,
logOffset,
@@ -327,7 +345,6 @@ public final class KvWriteProcessor {
currentValue,
currentMerger,
autoIncrementUpdater,
- valueDecoder,
walBuilder,
latestSchemaRow,
logOffset,
@@ -341,7 +358,6 @@ public final class KvWriteProcessor {
private long processDeletion(
KvPreWriteBuffer.Key key,
RowMerger currentMerger,
- ValueDecoder valueDecoder,
WalBuilder walBuilder,
PaddingRow latestSchemaRow,
long logOffset,
@@ -353,15 +369,14 @@ public final class KvWriteProcessor {
return logOffset;
}
- byte[] oldValueBytes = getFromState(key, primaryKey, stateAccessor, memoizedLakeLookup);
- if (oldValueBytes == null) {
+ BinaryValue oldValue = getPreviousValue(key, primaryKey, stateAccessor, memoizedLakeLookup);
+ if (oldValue == null) {
LOG.debug(
"The specific key can't be found in kv tablet although the kv record is for deletion, "
+ "ignore it directly as it doesn't exist in the kv tablet yet.");
return logOffset;
}
- BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
BinaryValue newValue = currentMerger.delete(oldValue);
// if newValue is null, it means the row should be deleted
@@ -379,7 +394,6 @@ public final class KvWriteProcessor {
BinaryValue currentValue,
RowMerger currentMerger,
AutoIncrementUpdater autoIncrementUpdater,
- ValueDecoder valueDecoder,
WalBuilder walBuilder,
PaddingRow latestSchemaRow,
long logOffset,
@@ -392,8 +406,8 @@ public final class KvWriteProcessor {
key, null, currentValue, walBuilder, latestSchemaRow, logOffset, stateAccessor);
}
- byte[] oldValueBytes = getFromState(key, primaryKey, stateAccessor, memoizedLakeLookup);
- if (oldValueBytes == null) {
+ BinaryValue oldValue = getPreviousValue(key, primaryKey, stateAccessor, memoizedLakeLookup);
+ if (oldValue == null) {
BinaryValue valueToInsert = currentMerger.merge(null, currentValue);
return applyInsert(
key,
@@ -405,7 +419,6 @@ public final class KvWriteProcessor {
stateAccessor);
}
- BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
BinaryValue newValue = currentMerger.merge(oldValue, currentValue);
if (newValue == oldValue) {
@@ -441,7 +454,7 @@ public final class KvWriteProcessor {
throws Exception {
BinaryValue newValue = autoIncrementUpdater.updateAutoIncrementColumns(currentValue);
walBuilder.append(ChangeType.INSERT, latestSchemaRow.replaceRow(newValue.row));
- stateAccessor.insert(key, newValue.encodeValue(), logOffset);
+ stateAccessor.insert(key, valueEncoder.encodeValue(newValue), logOffset);
return logOffset + 1;
}
@@ -456,12 +469,12 @@ public final class KvWriteProcessor {
throws Exception {
if (changelogImage == ChangelogImage.WAL) {
walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
- stateAccessor.update(key, newValue.encodeValue(), logOffset);
+ stateAccessor.update(key, valueEncoder.encodeValue(newValue), logOffset);
return logOffset + 1;
} else {
walBuilder.append(ChangeType.UPDATE_BEFORE, latestSchemaRow.replaceRow(oldValue.row));
walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
- stateAccessor.update(key, newValue.encodeValue(), logOffset + 1);
+ stateAccessor.update(key, valueEncoder.encodeValue(newValue), logOffset + 1);
return logOffset + 2;
}
}
@@ -477,7 +490,7 @@ public final class KvWriteProcessor {
* partition namespace; used by the lake lookup
* @return the previous value, or null if the key is absent or locally marked as deleted
*/
- private byte[] getFromState(
+ private BinaryValue getPreviousValue(
KvPreWriteBuffer.Key localStateKey,
byte[] primaryKey,
KvStateAccessor stateAccessor,
@@ -486,9 +499,12 @@ public final class KvWriteProcessor {
KvStateLookupResult localResult = stateAccessor.lookup(localStateKey);
if (localResult.status() != KvStateLookupResult.Status.NOT_FOUND
|| memoizedLakeLookup == null) {
- return localResult.value();
+ return localResult.value() == null
+ ? null
+ : valueDecoder.decodeValue(localResult.value());
}
- return memoizedLakeLookup.lookup(primaryKey);
+ byte[] lakeValue = memoizedLakeLookup.lookup(primaryKey);
+ return lakeValue == null ? null : lakeValueDecoder.decodeValue(lakeValue);
}
private boolean canSkipOldValueLookup(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlCompactionFilterFactory.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlCompactionFilterFactory.java
new file mode 100644
index 000000000..8cb01d0c1
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlCompactionFilterFactory.java
@@ -0,0 +1,68 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.server.utils.RowTtlUtils;
+import org.apache.fluss.utils.clock.Clock;
+
+import org.rocksdb.FlinkCompactionFilter;
+import org.rocksdb.RocksDB;
+
+import java.time.Duration;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Factory utilities for row TTL compaction filters. */
+public final class RowTtlCompactionFilterFactory {
+
+ private static final long QUERY_TIME_AFTER_NUM_ENTRIES = 1000L;
+
+ private RowTtlCompactionFilterFactory() {}
+
+ /** Creates a configured native compaction filter factory for row TTL cleanup. */
+ public static FlinkCompactionFilter.FlinkCompactionFilterFactory create(
+ KvValueLayout kvValueLayout, Duration ttl, Clock clock) {
+ return create(kvValueLayout, ttl, QUERY_TIME_AFTER_NUM_ENTRIES, clock);
+ }
+
+ @VisibleForTesting
+ static FlinkCompactionFilter.FlinkCompactionFilterFactory create(
+ KvValueLayout kvValueLayout, Duration ttl, long queryTimeAfterNumEntries, Clock clock) {
+ long ttlMillis = RowTtlUtils.validateAndConvertTtlDurationToMillis(ttl);
+ checkNotNull(kvValueLayout, "kvValueLayout must not be null.");
+ checkNotNull(clock, "clock must not be null.");
+ checkArgument(kvValueLayout.hasValueTag(), "Row TTL requires a tagged KV value layout.");
+ checkArgument(
+ queryTimeAfterNumEntries > 0,
+ "queryTimeAfterNumEntries must be greater than zero.");
+
+ RocksDB.loadLibrary();
+ FlinkCompactionFilter.FlinkCompactionFilterFactory factory =
+ new FlinkCompactionFilter.FlinkCompactionFilterFactory(clock::milliseconds);
+ factory.configure(
+ FlinkCompactionFilter.Config.createNotList(
+ FlinkCompactionFilter.StateType.Value,
+ kvValueLayout.valueTagOffset(),
+ ttlMillis,
+ queryTimeAfterNumEntries));
+ return factory;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlTimestampProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlTimestampProvider.java
new file mode 100644
index 000000000..76644ddc0
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/RowTtlTimestampProvider.java
@@ -0,0 +1,131 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypeRoot;
+
+import java.time.ZoneId;
+import java.util.List;
+import java.util.function.ToLongFunction;
+
+import static org.apache.fluss.types.DataTypeChecks.getPrecision;
+
+/** Provides the timestamp used as the internal KV value tag for row TTL. */
+final class RowTtlTimestampProvider implements ToLongFunction<BinaryRow> {
+
+ static final long NEVER_EXPIRE_TIMESTAMP_MS = Long.MAX_VALUE / 2;
+
+ private final TimestampExtractor timestampExtractor;
+ private long timestampMs;
+
+ private RowTtlTimestampProvider(TimestampExtractor timestampExtractor) {
+ this.timestampExtractor = timestampExtractor;
+ }
+
+ static RowTtlTimestampProvider create(
+ TableConfig tableConfig, SchemaGetter schemaGetter, ZoneId serverTimeZone) {
+ if (tableConfig.getKvTTLTimeColumn().isPresent()) {
+ return forEventTime(tableConfig, schemaGetter, serverTimeZone);
+ }
+ return new RowTtlTimestampProvider(new BatchTimestampExtractor());
+ }
+
+ private static RowTtlTimestampProvider forEventTime(
+ TableConfig tableConfig, SchemaGetter schemaGetter, ZoneId serverTimeZone) {
+ String timeColumn = tableConfig.getKvTTLTimeColumn().get();
+ return new RowTtlTimestampProvider(
+ EventTimeTimestampExtractor.create(schemaGetter, timeColumn, serverTimeZone));
+ }
+
+ void prepareForBatch(long commitTimestamp) {
+ timestampMs = commitTimestamp;
+ }
+
+ @Override
+ public long applyAsLong(BinaryRow row) {
+ return timestampExtractor.extract(row, timestampMs);
+ }
+
+ private interface TimestampExtractor {
+ long extract(BinaryRow row, long timestampMs);
+ }
+
+ private static final class BatchTimestampExtractor implements TimestampExtractor {
+ @Override
+ public long extract(BinaryRow row, long timestampMs) {
+ return timestampMs;
+ }
+ }
+
+ private static final class EventTimeTimestampExtractor implements TimestampExtractor {
+ private final int fieldIndex;
+ private final DataType timeColumnType;
+ private final ZoneId serverTimeZone;
+
+ private EventTimeTimestampExtractor(
+ int fieldIndex, DataType timeColumnType, ZoneId serverTimeZone) {
+ this.fieldIndex = fieldIndex;
+ this.timeColumnType = timeColumnType;
+ this.serverTimeZone = serverTimeZone;
+ }
+
+ private static EventTimeTimestampExtractor create(
+ SchemaGetter schemaGetter, String timeColumn, ZoneId serverTimeZone) {
+ Schema schema = schemaGetter.getLatestSchemaInfo().getSchema();
+ List<Schema.Column> columns = schema.getColumns();
+ for (int i = 0; i < columns.size(); i++) {
+ Schema.Column column = columns.get(i);
+ if (column.getName().equals(timeColumn)) {
+ return new EventTimeTimestampExtractor(i, column.getDataType(), serverTimeZone);
+ }
+ }
+ throw new IllegalStateException(
+ String.format(
+ "Cannot find row TTL time column '%s' in latest schema.", timeColumn));
+ }
+
+ @Override
+ public long extract(BinaryRow row, long timestampMs) {
+ if (row.isNullAt(fieldIndex)) {
+ return NEVER_EXPIRE_TIMESTAMP_MS;
+ }
+ DataTypeRoot typeRoot = timeColumnType.getTypeRoot();
+ if (typeRoot == DataTypeRoot.BIGINT) {
+ return row.getLong(fieldIndex);
+ }
+ if (typeRoot == DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {
+ return row.getTimestampNtz(fieldIndex, getPrecision(timeColumnType))
+ .toLocalDateTime()
+ .atZone(serverTimeZone)
+ .toInstant()
+ .toEpochMilli();
+ }
+ if (typeRoot == DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
+ return row.getTimestampLtz(fieldIndex, getPrecision(timeColumnType))
+ .getEpochMillisecond();
+ }
+ throw new IllegalStateException(
+ String.format("Unsupported row TTL time column type: %s.", timeColumnType));
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java
index d28ef50e2..b44e89282 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java
@@ -24,6 +24,8 @@ import org.apache.fluss.server.utils.ResourceGuard;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.IOUtils;
+import org.rocksdb.AbstractCompactionFilter;
+import org.rocksdb.AbstractCompactionFilterFactory;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.NativeLibraryLoader;
@@ -96,6 +98,25 @@ public class RocksDBKvBuilder {
return this;
}
+ /**
+ * Sets the compaction filter factory and transfers its ownership to this builder.
+ *
+ * <p>The factory is closed with the built {@link RocksDBKv}, or when setup or building fails.
+ */
+ @SuppressWarnings("rawtypes")
+ public RocksDBKvBuilder setCompactionFilterFactory(
+ AbstractCompactionFilterFactory<? extends AbstractCompactionFilter<?>> factory) {
+ try {
+ columnFamilyOptions.setCompactionFilterFactory(
+ (AbstractCompactionFilterFactory) factory);
+ optionsContainer.registerCloseableResource(factory);
+ } catch (RuntimeException | Error e) {
+ IOUtils.closeQuietly(factory);
+ throw e;
+ }
+ return this;
+ }
+
public RocksDBKv build() throws KvBuildingException {
ColumnFamilyHandle defaultColumnFamilyHandle = null;
RocksDB db = null;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
index 6fb885b82..50f348f8d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
@@ -195,6 +195,10 @@ public class RocksDBResourceContainer implements AutoCloseable {
return opt;
}
+ void registerCloseableResource(AutoCloseable resource) {
+ handlesToClose.add(checkNotNull(resource, "resource must not be null"));
+ }
+
@Override
public void close() throws Exception {
handlesToClose.forEach(IOUtils::closeQuietly);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
index 6347a3d43..aeb49886c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server.kv.scan;
import org.apache.fluss.exception.KvStorageException;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.server.kv.rocksdb.RocksDBKv;
import org.apache.fluss.server.utils.ResourceGuard;
import org.apache.fluss.utils.IOUtils;
@@ -48,6 +49,7 @@ public class ScannerContext implements Closeable {
private final String scannerId;
private final byte[] scannerIdBytes;
private final TableBucket tableBucket;
+ private final KvValueLayout kvValueLayout;
private final RocksDBKv rocksDBKv;
private final RocksIterator iterator;
private final ReadOptions readOptions;
@@ -80,10 +82,12 @@ public class ScannerContext implements Closeable {
ResourceGuard.Lease resourceLease,
long limit,
long logOffset,
- long initialAccessTimeMs) {
+ long initialAccessTimeMs,
+ KvValueLayout kvValueLayout) {
this.scannerId = scannerId;
this.scannerIdBytes = scannerId.getBytes(StandardCharsets.UTF_8);
this.tableBucket = tableBucket;
+ this.kvValueLayout = kvValueLayout;
this.rocksDBKv = rocksDBKv;
this.iterator = iterator;
this.readOptions = readOptions;
@@ -112,6 +116,11 @@ public class ScannerContext implements Closeable {
return tableBucket;
}
+ /** Returns the physical layout of values read by this scanner. */
+ public KvValueLayout getKvValueLayout() {
+ return kvValueLayout;
+ }
+
public boolean isValid() {
return iterator.isValid() && remainingLimit != 0;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index 5e6036849..461672bcd 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -858,7 +858,8 @@ public final class Replica {
checkNotNull(kvTablet, "kv tablet should not be null.");
restoreStartOffset = completedSnapshot.getLogOffset();
- rowCount = completedSnapshot.getRowCount();
+ rowCount =
+ supportsExactRowCount(tableConfig) ? completedSnapshot.getRowCount() : null;
// currently, we only support one auto-increment column.
autoIncIDRange = completedSnapshot.getFirstAutoIncIDRange();
} else {
@@ -880,10 +881,10 @@ public final class Replica {
arrowCompressionInfo,
this::onKvFlushComplete);
- // we don't support rowCount
rowCount =
isHistoricalPartition()
|| tableConfig.getChangelogImage() == ChangelogImage.WAL
+ || tableConfig.getKvTTL().isPresent()
? null
: 0L;
// TODO: it is possible that this is a recovered kv tablet without kv snapshot but
@@ -1673,10 +1674,14 @@ public final class Replica {
checkNotNull(
kvTablet,
"KvTablet for the replica to limit scan shouldn't be null.");
- List<byte[]> bytes = kvTablet.limitScan(limit);
+ List<byte[]> values = kvTablet.limitScan(limit);
DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder();
- for (byte[] key : bytes) {
- builder.append(key);
+ int valueBodyOffset = kvTablet.getKvValueLayout().valueBodyOffset();
+ for (byte[] value : values) {
+ builder.append(
+ value,
+ valueBodyOffset,
+ kvTablet.getKvValueLayout().valueBodyLength(value.length));
}
return builder.build();
} catch (IOException e) {
@@ -1821,6 +1826,11 @@ public final class Replica {
});
}
+ private static boolean supportsExactRowCount(TableConfig tableConfig) {
+ return tableConfig.getChangelogImage() != ChangelogImage.WAL
+ && !tableConfig.getKvTTL().isPresent();
+ }
+
public long getOffset(RemoteLogManager remoteLogManager, ListOffsetsParam listOffsetsParam)
throws IOException {
return inReadLock(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index ef3b16b2c..4b9288f96 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -55,6 +55,7 @@ import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogSegment;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.entity.LimitScanResultForBucket;
@@ -965,7 +966,6 @@ public class ReplicaManager implements ServerReconfigurable {
result.add(
new LookupResultForBucket(
data.tableBucket(),
- null,
data.originalPartitionName(),
ApiError.fromThrowable(e)));
if (remainingLookups.decrementAndGet() == 0) {
@@ -980,7 +980,6 @@ public class ReplicaManager implements ServerReconfigurable {
? bucketResult
: new LookupResultForBucket(
data.tableBucket(),
- null,
data.originalPartitionName(),
ApiError.fromThrowable(error));
if (completedResult.failed()
@@ -1019,7 +1018,12 @@ public class ReplicaManager implements ServerReconfigurable {
tableMetrics = replica.tableMetrics();
tableMetrics.totalLookupRequests().inc();
lookupResultForBucketMap.put(
- tb, new LookupResultForBucket(tb, replica.lookups(entry.getValue())));
+ tb,
+ new LookupResultForBucket(
+ tb,
+ replica.lookups(entry.getValue()),
+ KvValueLayout.fromTableConfig(
+ replica.getTableInfo().getTableConfig())));
} catch (Exception e) {
if (isUnexpectedException(e)) {
LOG.error("Error lookup from local kv on replica {}", tb, e);
@@ -1151,7 +1155,13 @@ public class ReplicaManager implements ServerReconfigurable {
List<byte[]> resultForPerKey = replica.prefixLookup(prefixKey);
resultForBucket.add(resultForPerKey);
}
- result.put(tb, new PrefixLookupResultForBucket(tb, resultForBucket));
+ result.put(
+ tb,
+ new PrefixLookupResultForBucket(
+ tb,
+ resultForBucket,
+ KvValueLayout.fromTableConfig(
+ replica.getTableInfo().getTableConfig())));
} catch (Exception e) {
if (isUnexpectedException(e)) {
LOG.error("Error processing prefix lookup operation on replica {}", tb, e);
@@ -2559,26 +2569,25 @@ public class ReplicaManager implements ServerReconfigurable {
checkpointHighWatermarks();
}
- private void validateClientVersionForPkTable(int apiVersion, TableInfo tableInfo) {
- if (apiVersion > 0) {
+ /** Validates whether a client API version can access this primary-key table. */
+ public static void validateClientVersionForPkTable(int apiVersion, TableInfo tableInfo) {
+ if (apiVersion >= 1) {
return;
}
- // in the old version
TableConfig tableConfig = tableInfo.getTableConfig();
- // is with datalake format
- if (tableConfig.getDataLakeFormat().isPresent()) {
- Optional<Integer> kvFormatVersion = tableConfig.getKvFormatVersion();
- if (kvFormatVersion.isPresent()
- && kvFormatVersion.get() == KV_FORMAT_VERSION_2
- && !tableInfo.isDefaultBucketKey()) {
- throw new UnsupportedVersionException(
- String.format(
- "Client API version %d is not supported for table '%s'. "
- + "This table uses new key encoding strategy (kv format version %d). "
- + "Please upgrade your Fluss client to a newer version.",
- apiVersion, tableInfo.getTablePath(), kvFormatVersion.get()));
- }
+ Optional<Integer> kvFormatVersion = tableConfig.getKvFormatVersion();
+ if (apiVersion < 1
+ && tableConfig.getDataLakeFormat().isPresent()
+ && kvFormatVersion.isPresent()
+ && kvFormatVersion.get() == KV_FORMAT_VERSION_2
+ && !tableInfo.isDefaultBucketKey()) {
+ throw new UnsupportedVersionException(
+ String.format(
+ "Client API version %d is not supported for table '%s'. "
+ + "This table uses new key encoding strategy (kv format version %d). "
+ + "Please upgrade your Fluss client to a newer version.",
+ apiVersion, tableInfo.getTablePath(), kvFormatVersion.get()));
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java
index f54aef0c0..10a677079 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java
@@ -28,6 +28,7 @@ import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metrics.Counter;
import org.apache.fluss.plugin.PluginManager;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.entity.LookupResultForBucket;
import org.apache.fluss.rpc.entity.PutKvResultForBucket;
import org.apache.fluss.rpc.protocol.ApiError;
@@ -110,7 +111,6 @@ public final class HistoricalPartitionManager implements AutoCloseable {
() ->
new LookupResultForBucket(
tableBucket,
- null,
lookupData.originalPartitionName(),
ApiError.fromThrowable(
new HistoricalPartitionThrottledException(
@@ -123,7 +123,6 @@ public final class HistoricalPartitionManager implements AutoCloseable {
return CompletableFuture.completedFuture(
new LookupResultForBucket(
tableBucket,
- null,
lookupData.originalPartitionName(),
ApiError.fromThrowable(e)));
}
@@ -335,19 +334,25 @@ public final class HistoricalPartitionManager implements AutoCloseable {
Iterator<byte[]> lakeValueIterator = lakeValues.iterator();
List<byte[]> values = new ArrayList<>(localResults.size());
+ List<KvValueLayout> valueLayouts = new ArrayList<>(localResults.size());
+ KvValueLayout localValueLayout =
+ KvValueLayout.fromTableConfig(tableInfo.getTableConfig());
for (KvStateLookupResult localResult : localResults) {
// Consume one lake value for each NOT_FOUND result; local values and tombstones
// keep their original positions without advancing the lake iterator.
- values.add(
- localResult.status() == Status.NOT_FOUND
- ? lakeValueIterator.next()
- : localResult.value());
+ if (localResult.status() == Status.NOT_FOUND) {
+ values.add(lakeValueIterator.next());
+ valueLayouts.add(KvValueLayout.PLAIN);
+ } else {
+ values.add(localResult.value());
+ valueLayouts.add(localValueLayout);
+ }
}
return new LookupResultForBucket(
- tableBucket, values, originalPartitionName, ApiError.NONE);
+ tableBucket, values, valueLayouts, originalPartitionName);
} catch (Exception e) {
return new LookupResultForBucket(
- tableBucket, null, originalPartitionName, ApiError.fromThrowable(e));
+ tableBucket, originalPartitionName, ApiError.fromThrowable(e));
}
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 0b62270b9..45dfacd09 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -254,7 +254,12 @@ public class TabletServer extends ServerBase {
this.kvManager =
KvManager.create(
- conf, zkClient, logManager, tabletServerMetricGroup, localDiskManager);
+ conf,
+ zkClient,
+ logManager,
+ tabletServerMetricGroup,
+ localDiskManager,
+ clock);
kvManager.startup();
this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
index ba31a143d..bd3ef49b3 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
@@ -28,6 +28,7 @@ import org.apache.fluss.exception.UnknownScannerIdException;
import org.apache.fluss.exception.UnknownTableOrBucketException;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.DefaultValueRecordBatch;
import org.apache.fluss.record.KvRecordBatch;
@@ -615,9 +616,8 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
bucketReq.getBucketId());
Long limit = bucketReq.hasLimit() ? bucketReq.getLimit() : null;
- OpenScanResult openResult =
- scannerManager.createScanner(
- replicaManager.getReplicaOrException(tableBucket), limit);
+ Replica replica = replicaManager.getReplicaOrException(tableBucket);
+ OpenScanResult openResult = scannerManager.createScanner(replica, limit);
isNewScan = true;
initialLogOffset = openResult.getLogOffset();
@@ -689,8 +689,8 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
// Catch a leadership flip ahead of the eventual closeScannersForBucket callback so
// the client can redirect rather than consume a stale snapshot.
+ Replica replica = replicaManager.getReplicaOrException(context.getTableBucket());
if (!request.hasBucketScanReq()) {
- Replica replica = replicaManager.getReplicaOrException(context.getTableBucket());
if (!replica.isLeader()) {
throw new NotLeaderOrFollowerException(
String.format(
@@ -710,7 +710,11 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
boolean appendedAny = false;
while (context.isValid()
&& (!appendedAny || builder.sizeInBytes() < effectiveBatchSize)) {
- builder.append(context.currentValue());
+ byte[] value = context.currentValue();
+ builder.append(
+ value,
+ context.getKvValueLayout().valueBodyOffset(),
+ context.getKvValueLayout().valueBodyLength(value.length));
context.advance();
appendedAny = true;
}
@@ -795,6 +799,19 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
}
}
+ @Override
+ protected TableInfo getTableInfo(long tableId) {
+ TablePath tablePath = metadataCache.getTablePath(tableId).orElse(null);
+ if (tablePath == null) {
+ throw new UnknownTableOrBucketException(
+ String.format(
+ "This server %s does not know this table ID %s. This may happen when the table "
+ + "metadata cache in the server is not updated yet.",
+ serviceName, tableId));
+ }
+ return metadataManager.getTable(tablePath);
+ }
+
private void authorizeAnyTable(OperationType operationType, List<TablePath> tablePaths) {
if (authorizer != null) {
if (tablePaths.isEmpty()) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/RowTtlUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/RowTtlUtils.java
new file mode 100644
index 000000000..c418302fe
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/RowTtlUtils.java
@@ -0,0 +1,55 @@
+/*
+ * 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.server.utils;
+
+import org.apache.fluss.config.ConfigOptions;
+
+import java.time.Duration;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Utility methods for row-level TTL. */
+public final class RowTtlUtils {
+
+ private RowTtlUtils() {}
+
+ /** Validates a row TTL duration and returns its millisecond value. */
+ public static long validateAndConvertTtlDurationToMillis(Duration ttl) {
+ checkNotNull(ttl, "ttl must not be null.");
+ checkArgument(
+ ttl.compareTo(Duration.ZERO) > 0,
+ "'%s' must be positive.",
+ ConfigOptions.TABLE_KV_TTL.key());
+
+ try {
+ long ttlMillis = ttl.toMillis();
+ checkArgument(
+ ttlMillis >= 1,
+ "'%s' must be at least 1 ms.",
+ ConfigOptions.TABLE_KV_TTL.key());
+ return ttlMillis;
+ } catch (ArithmeticException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' exceeds the maximum supported row TTL.",
+ ConfigOptions.TABLE_KV_TTL.key()),
+ e);
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 81131f1bc..35a024f37 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -54,6 +54,7 @@ import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogSegment;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.entity.LimitScanResultForBucket;
import org.apache.fluss.rpc.entity.ListOffsetsResultForBucket;
@@ -1355,10 +1356,16 @@ public class ServerRpcMessageUtils {
lookupRespForBucket.setError(
bucketResult.getErrorCode(), bucketResult.getErrorMessage());
} else {
- for (byte[] value : bucketResult.lookupValues()) {
+ List<byte[]> values = bucketResult.lookupValues();
+ for (int i = 0; i < values.size(); i++) {
+ byte[] value = values.get(i);
PbValue pbValue = lookupRespForBucket.addValue();
if (value != null) {
- pbValue.setValues(value);
+ KvValueLayout kvValueLayout = bucketResult.getKvValueLayout(i);
+ pbValue.setValues(
+ value,
+ kvValueLayout.valueBodyOffset(),
+ kvValueLayout.valueBodyLength(value.length));
}
}
}
@@ -1390,10 +1397,14 @@ public class ServerRpcMessageUtils {
respForBucket.setError(bucketResult.getErrorCode(), bucketResult.getErrorMessage());
} else {
List<PbValueList> keyResultList = new ArrayList<>();
+ int valueBodyOffset = bucketResult.getKvValueLayout().valueBodyOffset();
for (List<byte[]> res : bucketResult.prefixLookupValues()) {
PbValueList pbValueList = new PbValueList();
for (byte[] bytes : res) {
- pbValueList.addValue(bytes);
+ pbValueList.addValue(
+ bytes,
+ valueBodyOffset,
+ bucketResult.getKvValueLayout().valueBodyLength(bytes.length));
}
keyResultList.add(pbValueList);
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index 3255fd87c..8ba8204a5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -38,6 +38,7 @@ import org.apache.fluss.metadata.MergeEngineType;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.DataTypeRoot;
import org.apache.fluss.types.RowType;
@@ -57,6 +58,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.fluss.config.ConfigOptions.CURRENT_KV_FORMAT_VERSION;
import static org.apache.fluss.config.FlussConfigUtils.TABLE_OPTIONS;
import static org.apache.fluss.config.FlussConfigUtils.isAlterableTableOption;
import static org.apache.fluss.config.FlussConfigUtils.isTableStorageConfig;
@@ -99,23 +101,24 @@ public class TableDescriptorValidation {
// check properties should only contain table.* options,
// and this cluster know it, and value is valid
for (String key : tableConf.keySet()) {
+ ConfigOption<?> option = TABLE_OPTIONS.get(key);
+ if (option != null) {
+ validateOptionValue(tableConf, option);
+ continue;
+ }
- if (!TABLE_OPTIONS.containsKey(key)) {
- if (isTableStorageConfig(key)) {
- throw new InvalidConfigException(
- String.format(
- "'%s' is not a recognized Fluss table property in the current cluster version. "
- + "You may be using an older Fluss cluster that does not support this property.",
- key));
- } else {
- throw new InvalidConfigException(
- String.format(
- "'%s' is not a Fluss table property. Please use '.customProperty(..)' to set custom properties.",
- key));
- }
+ if (isTableStorageConfig(key)) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' is not a recognized Fluss table property in the current cluster version. "
+ + "You may be using an older Fluss cluster that does not support this property.",
+ key));
+ } else {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' is not a Fluss table property. Please use '.customProperty(..)' to set custom properties.",
+ key));
}
- ConfigOption<?> option = TABLE_OPTIONS.get(key);
- validateOptionValue(tableConf, option);
}
// check distribution
@@ -129,6 +132,9 @@ public class TableDescriptorValidation {
checkDeleteBehavior(tableConf, hasPrimaryKey);
checkTieredLog(tableConf);
checkHistoricalPartition(tableDescriptor, tableConf);
+ checkKvTTL(tableConf, schema, hasPrimaryKey);
+ checkKvFormatVersion(tableConf);
+ checkKvValueLayout(tableConf, hasPrimaryKey);
checkPartition(tableConf, tableDescriptor.getPartitionKeys(), schema.getRowType());
checkSystemColumns(schema.getRowType());
validateStatisticsConfig(tableDescriptor);
@@ -304,6 +310,124 @@ public class TableDescriptorValidation {
}
}
+ private static void checkKvTTL(Configuration tableConf, Schema schema, boolean hasPrimaryKey) {
+ Optional<Duration> rowTTL = tableConf.getOptional(ConfigOptions.TABLE_KV_TTL);
+ Optional<String> timeColumn = tableConf.getOptional(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN);
+ if (timeColumn.isPresent() && !rowTTL.isPresent()) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' requires '%s' to be set.",
+ ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(),
+ ConfigOptions.TABLE_KV_TTL.key()));
+ }
+
+ if (!rowTTL.isPresent()) {
+ return;
+ }
+
+ if (!hasPrimaryKey) {
+ throw new InvalidTableException(
+ String.format(
+ "'%s' is only supported for primary key tables.",
+ ConfigOptions.TABLE_KV_TTL.key()));
+ }
+
+ validateKvTTLDuration(rowTTL.get());
+
+ if (timeColumn.isPresent()) {
+ Schema.Column column = getKvTTLTimeColumn(schema, timeColumn.get());
+ validateKvTTLTimeColumnType(column.getDataType());
+ }
+ }
+
+ private static Schema.Column getKvTTLTimeColumn(Schema schema, String timeColumn) {
+ for (Schema.Column column : schema.getColumns()) {
+ if (column.getName().equals(timeColumn)) {
+ return column;
+ }
+ }
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' refers to unknown column '%s'.",
+ ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), timeColumn));
+ }
+
+ private static void validateKvTTLTimeColumnType(DataType dataType) {
+ if (dataType.is(DataTypeRoot.BIGINT)
+ || dataType.is(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)
+ || dataType.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+ return;
+ }
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' only supports BIGINT, TIMESTAMP, or TIMESTAMP_LTZ columns, but was %s.",
+ ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), dataType));
+ }
+
+ private static void validateKvTTLDuration(Duration ttl) {
+ try {
+ RowTtlUtils.validateAndConvertTtlDurationToMillis(ttl);
+ } catch (IllegalArgumentException e) {
+ throw new InvalidConfigException(
+ String.format(
+ "Invalid value for '%s': %s",
+ ConfigOptions.TABLE_KV_TTL.key(), e.getMessage()));
+ }
+ }
+
+ private static void checkKvFormatVersion(Configuration tableConf) {
+ Optional<Integer> kvFormatVersion =
+ tableConf.getOptional(ConfigOptions.TABLE_KV_FORMAT_VERSION);
+ if (kvFormatVersion.isPresent() && kvFormatVersion.get() > CURRENT_KV_FORMAT_VERSION) {
+ throw new InvalidConfigException(
+ String.format(
+ "Unsupported kv format version %d. The maximum supported version is %d.",
+ kvFormatVersion.get(), CURRENT_KV_FORMAT_VERSION));
+ }
+ }
+
+ private static void checkKvValueLayout(Configuration tableConf, boolean hasPrimaryKey) {
+ boolean rowTtlEnabled = tableConf.getOptional(ConfigOptions.TABLE_KV_TTL).isPresent();
+ Optional<Integer> layoutVersion =
+ tableConf.getOptional(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION);
+ if (!layoutVersion.isPresent()) {
+ if (rowTtlEnabled) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' must be set when '%s' is set.",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ ConfigOptions.TABLE_KV_TTL.key()));
+ }
+ return;
+ }
+ if (!hasPrimaryKey) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' is only supported for primary key tables.",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key()));
+ }
+
+ KvValueLayout layout;
+ try {
+ layout = KvValueLayout.fromVersion(layoutVersion.get());
+ } catch (IllegalArgumentException e) {
+ throw new InvalidConfigException(
+ String.format(
+ "Invalid value for '%s': %s.",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ layoutVersion.get()));
+ }
+
+ if (layout.hasValueTag() != rowTtlEnabled) {
+ throw new InvalidConfigException(
+ String.format(
+ "'%s' version %d is incompatible with '%s'.",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ layout.version(),
+ ConfigOptions.TABLE_KV_TTL.key()));
+ }
+ }
+
private static void checkDistribution(TableDescriptor tableDescriptor, int maxBucketNum) {
if (!tableDescriptor.getTableDistribution().isPresent()) {
throw new InvalidTableException("Table distribution is required.");
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/RpcServiceBaseKvSnapshotVersionTest.java b/fluss-server/src/test/java/org/apache/fluss/server/RpcServiceBaseKvSnapshotVersionTest.java
new file mode 100644
index 000000000..8158e7a0e
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/RpcServiceBaseKvSnapshotVersionTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.server;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.UnsupportedVersionException;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.TestData;
+import org.apache.fluss.row.encode.KvValueLayout;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.server.RpcServiceBase.validateKvSnapshotMetadataVersion;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests KV snapshot metadata API compatibility. */
+class RpcServiceBaseKvSnapshotVersionTest {
+
+ @Test
+ void testTaggedSnapshotRequiresVersionOne() {
+ TableInfo taggedTableInfo = tableInfo(KvValueLayout.TAGGED);
+ TableInfo plainTableInfo = tableInfo(KvValueLayout.PLAIN);
+
+ assertThatThrownBy(() -> validateKvSnapshotMetadataVersion((short) 0, taggedTableInfo))
+ .isInstanceOf(UnsupportedVersionException.class);
+ validateKvSnapshotMetadataVersion((short) 1, taggedTableInfo);
+ validateKvSnapshotMetadataVersion((short) 0, plainTableInfo);
+ }
+
+ private static TableInfo tableInfo(KvValueLayout layout) {
+ Map<String, String> properties =
+ new HashMap<>(TestData.DATA1_TABLE_DESCRIPTOR_PK.getProperties());
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(layout.version()));
+ TableDescriptor descriptor = TestData.DATA1_TABLE_DESCRIPTOR_PK.withProperties(properties);
+ return TableInfo.of(
+ TablePath.of("db", "table_" + layout.version()),
+ layout.version(),
+ 1,
+ descriptor,
+ TestData.DEFAULT_REMOTE_DATA_DIR,
+ 1L,
+ 1L);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
index 944341a57..812564b08 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
@@ -42,6 +42,7 @@ import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.metrics.registry.MetricRegistry;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.GatewayClientProxy;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.gateway.AdminGateway;
@@ -305,6 +306,9 @@ class TableManagerITCase {
properties.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version()));
assertThat(gottenTable)
.isEqualTo(tableDescriptor.withProperties(properties).withReplicationFactor(1));
@@ -503,6 +507,9 @@ class TableManagerITCase {
properties.put(
ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
String.valueOf(CURRENT_KV_FORMAT_VERSION));
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version()));
tableDescriptor = tableDescriptor.withProperties(properties);
assertThat(TableDescriptor.fromJsonBytes(tableMetadata.getTableJson()))
@@ -587,7 +594,7 @@ class TableManagerITCase {
void testMetadataWithPartition(boolean isCoordinatorServer) throws Exception {
AdminReadOnlyGateway gateway = getAdminOnlyGateway(isCoordinatorServer);
AdminGateway adminGateway = getAdminGateway();
- String db1 = "db1";
+ String db1 = "db1_" + (isCoordinatorServer ? "coordinator" : "tablet");
String tb1 = "tb1";
// create a partitioned table, and request a not exist partition, should throw partition not
// exist exception
@@ -645,7 +652,9 @@ class TableManagerITCase {
.cause()
.isInstanceOf(PartitionNotExistException.class)
.hasMessage(
- "Table partition 'db1.partitioned_tb(p=not_exist_partition)' does not exist.");
+ String.format(
+ "Table partition '%s.partitioned_tb(p=not_exist_partition)' does not exist.",
+ db1));
}
@ParameterizedTest
@@ -688,6 +697,114 @@ class TableManagerITCase {
FLUSS_CLUSTER_EXTENSION.getTabletServerNodes());
}
+ @Test
+ void testCreatePersistsServerOwnedKvValueLayouts() throws Exception {
+ AdminReadOnlyGateway gateway = getAdminOnlyGateway(true);
+ AdminGateway adminGateway = getAdminGateway();
+
+ String databaseName = "db_value_layout";
+ TablePath plainTablePath = TablePath.of(databaseName, "plain_pk");
+ TablePath ttlTablePath = TablePath.of(databaseName, "ttl_pk");
+ TablePath logTablePath = TablePath.of(databaseName, "log_table");
+ adminGateway.createDatabase(newCreateDatabaseRequest(databaseName, false)).get();
+
+ adminGateway.createTable(newCreateTableRequest(plainTablePath, newPkTable(), false)).get();
+ TableDescriptor ttlDescriptor =
+ newPkTable()
+ .withProperties(
+ Collections.singletonMap(ConfigOptions.TABLE_KV_TTL.key(), "1 h"));
+ adminGateway.createTable(newCreateTableRequest(ttlTablePath, ttlDescriptor, false)).get();
+ adminGateway.createTable(newCreateTableRequest(logTablePath, newLogTable(), false)).get();
+
+ TableDescriptor plainTable =
+ TableDescriptor.fromJsonBytes(
+ gateway.getTableInfo(newGetTableInfoRequest(plainTablePath))
+ .get()
+ .getTableJson());
+ TableDescriptor ttlTable =
+ TableDescriptor.fromJsonBytes(
+ gateway.getTableInfo(newGetTableInfoRequest(ttlTablePath))
+ .get()
+ .getTableJson());
+ TableDescriptor logTable =
+ TableDescriptor.fromJsonBytes(
+ gateway.getTableInfo(newGetTableInfoRequest(logTablePath))
+ .get()
+ .getTableJson());
+
+ assertThat(plainTable.getProperties())
+ .containsEntry(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version()));
+ assertThat(ttlTable.getProperties())
+ .containsEntry(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+ assertThat(logTable.getProperties())
+ .doesNotContainKey(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key());
+ }
+
+ @Test
+ void testCreateOverridesClientSuppliedValueLayoutVersion() throws Exception {
+ AdminReadOnlyGateway gateway = getAdminOnlyGateway(true);
+ AdminGateway adminGateway = getAdminGateway();
+
+ String db1 = "db_event_time_row_ttl";
+ String tb1 = "tb_event_time_row_ttl";
+ TablePath tablePath = TablePath.of(db1, tb1);
+ adminGateway.createDatabase(newCreateDatabaseRequest(db1, false)).get();
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("event_time", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ Map<String, String> properties = new HashMap<>();
+ properties.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ properties.put(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), "event_time");
+ properties.put(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(), "1");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(3, "id")
+ .properties(properties)
+ .build();
+ adminGateway.createTable(newCreateTableRequest(tablePath, tableDescriptor, false)).get();
+
+ GetTableInfoResponse response =
+ gateway.getTableInfo(newGetTableInfoRequest(tablePath)).get();
+ TableDescriptor gottenTable = TableDescriptor.fromJsonBytes(response.getTableJson());
+
+ assertThat(gottenTable.getProperties())
+ .containsEntry(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+
+ String layoutVersion =
+ gottenTable.getProperties().get(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key());
+ adminGateway
+ .alterTable(
+ newAlterTableRequest(
+ tablePath,
+ Collections.singletonMap(
+ ConfigOptions.TABLE_KV_STANDBY_REPLICA_ENABLED.key(),
+ "false"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ false))
+ .get();
+
+ TableDescriptor alteredTable =
+ TableDescriptor.fromJsonBytes(
+ gateway.getTableInfo(newGetTableInfoRequest(tablePath))
+ .get()
+ .getTableJson());
+ assertThat(alteredTable.getProperties())
+ .containsEntry(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(), layoutVersion);
+ }
+
@Test
void testSchemaEvolution() throws Exception {
AdminReadOnlyGateway gateway = getAdminOnlyGateway(true);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 7b39bd8da..e0ac5205f 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -25,6 +25,7 @@ import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.InvalidTargetColumnException;
import org.apache.fluss.exception.OutOfOrderSequenceException;
import org.apache.fluss.exception.StorageBackpressureException;
+import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.memory.TestingMemorySegmentPool;
import org.apache.fluss.metadata.AggFunctions;
import org.apache.fluss.metadata.KvFormat;
@@ -35,6 +36,7 @@ import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.FileLogProjection;
import org.apache.fluss.record.KvRecord;
@@ -51,6 +53,8 @@ import org.apache.fluss.record.TestData;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.record.bytesview.MultiBytesView;
import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.server.kv.autoinc.AutoIncrementManager;
import org.apache.fluss.server.kv.autoinc.TestingSequenceGeneratorFactory;
@@ -76,6 +80,8 @@ import org.apache.fluss.types.DataTypes;
import org.apache.fluss.types.RowType;
import org.apache.fluss.types.StringType;
import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.clock.Clock;
+import org.apache.fluss.utils.clock.ManualClock;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.concurrent.FlussScheduler;
@@ -92,6 +98,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.lang.reflect.Field;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -151,27 +158,45 @@ class KvTabletTest {
@AfterEach
void afterEach() throws Exception {
- if (kvTablet != null) {
- kvTablet.close();
- }
- if (logTablet != null) {
- logTablet.close();
- }
- if (executor != null) {
- executor.shutdown();
+ try {
+ if (kvTablet != null) {
+ kvTablet.close();
+ }
+ } finally {
+ try {
+ if (logTablet != null) {
+ logTablet.close();
+ }
+ } finally {
+ if (executor != null) {
+ executor.shutdown();
+ }
+ }
}
}
private void initLogTabletAndKvTablet(Schema schema, Map<String, String> tableConfig)
throws Exception {
- initLogTabletAndKvTablet(TablePath.of("testDb", "t1"), schema, tableConfig);
+ initLogTabletAndKvTablet(
+ TablePath.of("testDb", "t1"), schema, tableConfig, SystemClock.getInstance());
+ }
+
+ private void initLogTabletAndKvTablet(
+ Schema schema, Map<String, String> tableConfig, Clock clock) throws Exception {
+ initLogTabletAndKvTablet(TablePath.of("testDb", "t1"), schema, tableConfig, clock);
}
private void initLogTabletAndKvTablet(
TablePath tablePath, Schema schema, Map<String, String> tableConfig) throws Exception {
+ initLogTabletAndKvTablet(tablePath, schema, tableConfig, SystemClock.getInstance());
+ }
+
+ private void initLogTabletAndKvTablet(
+ TablePath tablePath, Schema schema, Map<String, String> tableConfig, Clock clock)
+ throws Exception {
PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath);
schemaGetter = new TestingSchemaGetter(new SchemaInfo(schema, schemaId));
- logTablet = createLogTablet(tempLogDir, 0L, physicalTablePath);
+ logTablet = createLogTablet(tempLogDir, 0L, physicalTablePath, clock);
TableBucket tableBucket = logTablet.getTableBucket();
kvTablet =
createKvTablet(
@@ -180,11 +205,18 @@ class KvTabletTest {
logTablet,
tmpKvDir,
schemaGetter,
- tableConfig);
+ tableConfig,
+ clock);
}
private LogTablet createLogTablet(File tempLogDir, long tableId, PhysicalTablePath tablePath)
throws Exception {
+ return createLogTablet(tempLogDir, tableId, tablePath, SystemClock.getInstance());
+ }
+
+ private LogTablet createLogTablet(
+ File tempLogDir, long tableId, PhysicalTablePath tablePath, Clock clock)
+ throws Exception {
File logTabletDir =
LogTestUtils.makeRandomLogTabletDir(
tempLogDir, tablePath.getDatabaseName(), tableId, tablePath.getTableName());
@@ -201,7 +233,7 @@ class KvTabletTest {
LogFormat.ARROW,
1,
true,
- SystemClock.getInstance(),
+ clock,
true);
}
@@ -214,7 +246,34 @@ class KvTabletTest {
Map<String, String> tableConfig)
throws Exception {
return createKvTablet(
- tablePath, tableBucket, logTablet, tmpKvDir, schemaGetter, tableConfig, null);
+ tablePath,
+ tableBucket,
+ logTablet,
+ tmpKvDir,
+ schemaGetter,
+ tableConfig,
+ null,
+ SystemClock.getInstance());
+ }
+
+ private KvTablet createKvTablet(
+ PhysicalTablePath tablePath,
+ TableBucket tableBucket,
+ LogTablet logTablet,
+ File tmpKvDir,
+ SchemaGetter schemaGetter,
+ Map<String, String> tableConfig,
+ Clock clock)
+ throws Exception {
+ return createKvTablet(
+ tablePath,
+ tableBucket,
+ logTablet,
+ tmpKvDir,
+ schemaGetter,
+ tableConfig,
+ null,
+ clock);
}
private KvTablet createKvTablet(
@@ -224,7 +283,28 @@ class KvTabletTest {
File tmpKvDir,
SchemaGetter schemaGetter,
Map<String, String> tableConfig,
- @Nullable KvFlushScheduler kvFlushScheduler)
+ KvFlushScheduler kvFlushScheduler)
+ throws Exception {
+ return createKvTablet(
+ tablePath,
+ tableBucket,
+ logTablet,
+ tmpKvDir,
+ schemaGetter,
+ tableConfig,
+ kvFlushScheduler,
+ SystemClock.getInstance());
+ }
+
+ private KvTablet createKvTablet(
+ PhysicalTablePath tablePath,
+ TableBucket tableBucket,
+ LogTablet logTablet,
+ File tmpKvDir,
+ SchemaGetter schemaGetter,
+ Map<String, String> tableConfig,
+ @Nullable KvFlushScheduler kvFlushScheduler,
+ Clock clock)
throws Exception {
TableConfig tableConf = new TableConfig(Configuration.fromMap(tableConfig));
RowMerger rowMerger = RowMerger.create(tableConf, KvFormat.COMPACTED, schemaGetter);
@@ -251,7 +331,9 @@ class KvTabletTest {
schemaGetter,
tableConf.getChangelogImage(),
KvManager.getDefaultRateLimiter(),
- autoIncrementManager);
+ autoIncrementManager,
+ clock,
+ tableConf);
}
return KvTablet.create(
tablePath,
@@ -270,7 +352,225 @@ class KvTabletTest {
KvManager.getDefaultRateLimiter(),
kvFlushScheduler,
null,
- autoIncrementManager);
+ autoIncrementManager,
+ clock,
+ tableConf);
+ }
+
+ @Test
+ void testRowTtlProcessTimeValueTimestampUsesClock() throws Exception {
+ long writeTimestampMs = 123456789L;
+ long walTimestampFloorMs = writeTimestampMs + 1000L;
+ ManualClock clock = new ManualClock(walTimestampFloorMs);
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, kvTtlProcessTimeConfig(), clock);
+ logTablet.appendAsLeader(
+ logRecords(
+ 0L,
+ Collections.singletonList(ChangeType.INSERT),
+ Collections.singletonList(new Object[] {0, "seed"})));
+ clock.advanceTime(Duration.ofMillis(writeTimestampMs - walTimestampFloorMs));
+
+ KvRecord record = kvRecordFactory.ofRecord("k1".getBytes(), new Object[] {1, "a"});
+ LogAppendInfo appendInfo =
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(Collections.singletonList(record)), null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ byte[] value = kvTablet.multiGet(Collections.singletonList("k1".getBytes())).get(0);
+ BinaryValue decoded =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.TAGGED)
+ .decodeValue(value);
+
+ assertThat(readTaggedValueTag(value)).isEqualTo(writeTimestampMs);
+ assertThat(appendInfo.maxTimestamp()).isEqualTo(walTimestampFloorMs);
+ assertThat(decoded.row.getInt(0)).isEqualTo(1);
+ assertThat(decoded.row.getString(1).toString()).isEqualTo("a");
+ }
+
+ @Test
+ void testRowTtlEventTimeValueTimestampUsesTimeColumn() throws Exception {
+ Schema eventTimeSchema = eventTimeSchema();
+ Map<String, String> tableConfig = kvTtlEventTimeConfig();
+ initLogTabletAndKvTablet(eventTimeSchema, tableConfig, new ManualClock(123456789L));
+
+ KvRecordTestUtils.KvRecordFactory eventTimeRecordFactory =
+ KvRecordTestUtils.KvRecordFactory.of(eventTimeSchema.getRowType());
+ KvRecord record =
+ eventTimeRecordFactory.ofRecord(
+ "k1".getBytes(), new Object[] {1, 1234L, "event-time-row"});
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(Collections.singletonList(record)), null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ byte[] value = readValue("k1");
+ BinaryValue decoded = decodeTaggedValue(value);
+
+ assertThat(readTaggedValueTag(value)).isEqualTo(1234L);
+ assertThat(decoded.row.getInt(0)).isEqualTo(1);
+ assertThat(decoded.row.getLong(1)).isEqualTo(1234L);
+ assertThat(decoded.row.getString(2).toString()).isEqualTo("event-time-row");
+ }
+
+ @Test
+ void testRowTtlNullEventTimeValueNeverExpires() throws Exception {
+ Schema eventTimeSchema = eventTimeSchema();
+ Map<String, String> tableConfig = kvTtlEventTimeConfig();
+ ManualClock clock = new ManualClock(123456789L);
+ initLogTabletAndKvTablet(eventTimeSchema, tableConfig, clock);
+
+ KvRecordTestUtils.KvRecordFactory eventTimeRecordFactory =
+ KvRecordTestUtils.KvRecordFactory.of(eventTimeSchema.getRowType());
+ KvRecord record =
+ eventTimeRecordFactory.ofRecord(
+ "k1".getBytes(), new Object[] {1, null, "null-event-time-row"});
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(Collections.singletonList(record)), null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ byte[] value = readValue("k1");
+ BinaryValue decoded = decodeTaggedValue(value);
+
+ assertThat(readTaggedValueTag(value))
+ .isEqualTo(RowTtlTimestampProvider.NEVER_EXPIRE_TIMESTAMP_MS);
+ assertThat(decoded.row.getInt(0)).isEqualTo(1);
+ assertThat(decoded.row.isNullAt(1)).isTrue();
+ assertThat(decoded.row.getString(2).toString()).isEqualTo("null-event-time-row");
+
+ clock.advanceTime(Duration.ofHours(2L));
+ kvTablet.getRocksDBKv().getDb().compactRange();
+
+ assertThat(kvTablet.multiGet(Collections.singletonList("k1".getBytes())).get(0))
+ .isNotNull();
+ }
+
+ @Test
+ void testPartialUpdateFromNullEventTimeExpiresWholeRowAfterCompaction() throws Exception {
+ long eventTimestampMs = 1000L;
+ ManualClock clock = new ManualClock(eventTimestampMs);
+ Schema eventTimeSchema = eventTimeSchema();
+ initLogTabletAndKvTablet(eventTimeSchema, kvTtlEventTimeConfig(), clock);
+
+ String key = "partial-event-time";
+ KvRecordTestUtils.KvRecordFactory eventTimeRecordFactory =
+ KvRecordTestUtils.KvRecordFactory.of(eventTimeSchema.getRowType());
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ eventTimeRecordFactory.ofRecord(
+ key.getBytes(), new Object[] {1, null, "retained-name"})),
+ null);
+ // Persist the never-expiring version before overwriting it with a partial update.
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ eventTimeRecordFactory.ofRecord(
+ key.getBytes(),
+ new Object[] {1, eventTimestampMs, "ignored-name"})),
+ new int[] {0, 1});
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ byte[] value = readValue(key);
+ BinaryValue updatedValue = decodeTaggedValue(value);
+ assertThat(readTaggedValueTag(value)).isEqualTo(eventTimestampMs);
+ assertThat(updatedValue.row.getLong(1)).isEqualTo(eventTimestampMs);
+ assertThat(updatedValue.row.getString(2).toString()).isEqualTo("retained-name");
+
+ clock.advanceTime(Duration.ofHours(2L));
+ kvTablet.getRocksDBKv().getDb().compactRange();
+
+ assertThat(kvTablet.multiGet(Collections.singletonList(key.getBytes())).get(0)).isNull();
+ }
+
+ @Test
+ void testPartialUpdateWithoutEventTimeRetainsTimestampForCompaction() throws Exception {
+ long eventTimestampMs = 1000L;
+ ManualClock clock = new ManualClock(eventTimestampMs);
+ Schema eventTimeSchema = eventTimeSchema();
+ initLogTabletAndKvTablet(eventTimeSchema, kvTtlEventTimeConfig(), clock);
+
+ String key = "retained-event-time";
+ KvRecordTestUtils.KvRecordFactory eventTimeRecordFactory =
+ KvRecordTestUtils.KvRecordFactory.of(eventTimeSchema.getRowType());
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ eventTimeRecordFactory.ofRecord(
+ key.getBytes(), new Object[] {1, eventTimestampMs, "old-name"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ eventTimeRecordFactory.ofRecord(
+ key.getBytes(), new Object[] {1, null, "new-name"})),
+ new int[] {0, 2});
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ byte[] value = readValue(key);
+ BinaryValue updatedValue = decodeTaggedValue(value);
+ assertThat(readTaggedValueTag(value)).isEqualTo(eventTimestampMs);
+ assertThat(updatedValue.row.getLong(1)).isEqualTo(eventTimestampMs);
+ assertThat(updatedValue.row.getString(2).toString()).isEqualTo("new-name");
+
+ clock.advanceTime(Duration.ofHours(2L));
+ kvTablet.getRocksDBKv().getDb().compactRange();
+
+ assertThat(kvTablet.multiGet(Collections.singletonList(key.getBytes())).get(0)).isNull();
+ }
+
+ @Test
+ void testKvTTLCompactionRemovesExpiredRowsFromLookupAndScan() throws Exception {
+ long writeTimestampMs = 1000L;
+ ManualClock clock = new ManualClock(writeTimestampMs);
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, kvTtlProcessTimeConfig(), clock);
+
+ String expiredKey = "mm-expired-target";
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord(
+ expiredKey.getBytes(), new Object[] {0, "expired-target"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ clock.advanceTime(Duration.ofHours(2L));
+ List<KvRecord> freshRows = new ArrayList<>();
+ // These keys precede the expired key and cross the filter's 1000-entry clock refresh.
+ int freshPrefixRows = 1001;
+ for (int i = 0; i < freshPrefixRows; i++) {
+ String key = String.format("aa-fresh-%04d", i);
+ freshRows.add(
+ kvRecordFactory.ofRecord(
+ key.getBytes(), new Object[] {i + 1, "fresh-prefix-" + i}));
+ }
+ String freshKey = "zz-fresh";
+ freshRows.add(kvRecordFactory.ofRecord(freshKey.getBytes(), new Object[] {2000, "fresh"}));
+ kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(freshRows), null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ kvTablet.getRocksDBKv().getDb().compactRange();
+
+ List<byte[]> lookupValues =
+ kvTablet.multiGet(Arrays.asList(expiredKey.getBytes(), freshKey.getBytes()));
+ assertThat(lookupValues.get(0)).isNull();
+ assertThat(lookupValues.get(1)).isNotNull();
+
+ ValueDecoder valueDecoder =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.TAGGED);
+ OpenScanResult result = kvTablet.openScan("scanner-row-ttl", -1L, 0L);
+ ScannerContext context = result.getContext();
+ assertThat(context).isNotNull();
+ List<String> scannedNames = new ArrayList<>();
+ while (context.isValid()) {
+ BinaryValue value = valueDecoder.decodeValue(context.currentValue());
+ scannedNames.add(value.row.getString(1).toString());
+ context.advance();
+ }
+ context.checkIteratorStatus();
+ context.close();
+
+ assertThat(scannedNames)
+ .hasSize(freshPrefixRows + 1)
+ .doesNotContain("expired-target")
+ .contains("fresh");
}
@Test
@@ -2206,6 +2506,47 @@ class KvTabletTest {
kvTablet.close();
}
+ private byte[] readValue(String key) throws Exception {
+ return kvTablet.multiGet(Collections.singletonList(key.getBytes())).get(0);
+ }
+
+ private BinaryValue decodeTaggedValue(byte[] value) {
+ return new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.TAGGED)
+ .decodeValue(value);
+ }
+
+ private static long readTaggedValueTag(byte[] value) {
+ return KvValueLayout.TAGGED.readValueTag(MemorySegment.wrap(value));
+ }
+
+ private static Schema eventTimeSchema() {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("event_time", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ }
+
+ private static Map<String, String> kvTtlEventTimeConfig() {
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ tableConfig.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+ tableConfig.put(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), "event_time");
+ return tableConfig;
+ }
+
+ private static Map<String, String> kvTtlProcessTimeConfig() {
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ tableConfig.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+ return tableConfig;
+ }
+
@Test
void testRowCountWithUpsert() throws Exception {
initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
@@ -2262,6 +2603,24 @@ class KvTabletTest {
kvTablet.close();
}
+ @Test
+ void testRowCountDisabledForKvTTL() throws Exception {
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, kvTtlProcessTimeConfig());
+
+ KvRecordBatch batch =
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("key1", new Object[] {1, "val1"}));
+ kvTablet.putAsLeader(batch, null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ assertThatThrownBy(() -> kvTablet.getRowCount())
+ .isInstanceOf(InvalidTableException.class)
+ .hasMessageContaining("row TTL");
+ assertThat(kvTablet.getTabletState().getRowCount()).isNull();
+
+ kvTablet.close();
+ }
+
@Test
void testRowCountWithMixedOperations() throws Exception {
initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlCompactionFilterTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlCompactionFilterTest.java
new file mode 100644
index 000000000..762b3756c
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlCompactionFilterTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.record.BinaryValue;
+import org.apache.fluss.rocksdb.RocksDBHandle;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.utils.clock.ManualClock;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.FlinkCompactionFilter;
+import org.rocksdb.FlushOptions;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.Duration;
+
+import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
+import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests row TTL cleanup with native Flink compaction filter. */
+class RowTtlCompactionFilterTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void testFlinkCompactionFilterReadsTimestampFromTaggedValue() throws Exception {
+ byte[] expiredKey = "expired-key".getBytes(StandardCharsets.UTF_8);
+ byte[] freshKey = "fresh-key".getBytes(StandardCharsets.UTF_8);
+ BinaryRow row = compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
+ long now = 123456789L;
+
+ try (FlinkCompactionFilter.FlinkCompactionFilterFactory filterFactory =
+ RowTtlCompactionFilterFactory.create(
+ KvValueLayout.TAGGED,
+ Duration.ofHours(1L),
+ 1L,
+ new ManualClock(now));
+ DBOptions dbOptions = new DBOptions().setCreateIfMissing(true);
+ ColumnFamilyOptions cfOptions =
+ new ColumnFamilyOptions().setCompactionFilterFactory(filterFactory);
+ RocksDBHandle handle = new RocksDBHandle(tempDir.toFile(), dbOptions, cfOptions);
+ FlushOptions flushOptions = new FlushOptions().setWaitForFlush(true)) {
+ handle.openDB();
+ handle.getDb()
+ .put(expiredKey, encodeTaggedValue(row, now - Duration.ofHours(2L).toMillis()));
+ handle.getDb().put(freshKey, encodeTaggedValue(row, now));
+ handle.getDb().flush(flushOptions);
+
+ handle.getDb().compactRange();
+
+ assertThat(handle.getDb().get(expiredKey)).isNull();
+ assertThat(handle.getDb().get(freshKey)).isNotNull();
+ }
+ }
+
+ @Test
+ void testCreateRejectsInvalidTtlDuration() {
+ assertThatThrownBy(
+ () ->
+ RowTtlCompactionFilterFactory.create(
+ KvValueLayout.TAGGED,
+ Duration.ZERO,
+ 1L,
+ new ManualClock(0L)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL.key());
+ }
+
+ private static byte[] encodeTaggedValue(BinaryRow row, long valueTag) {
+ return ValueEncoder.forLayout(KvValueLayout.TAGGED, ignored -> valueTag)
+ .encodeValue(new BinaryValue(DEFAULT_SCHEMA_ID, row));
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlTimestampProviderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlTimestampProviderTest.java
new file mode 100644
index 000000000..90a93c330
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/RowTtlTimestampProviderTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.record.TestingSchemaGetter;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RowTtlTimestampProvider}. */
+class RowTtlTimestampProviderTest {
+
+ private static final short SCHEMA_ID = 0;
+ private static final ZoneId UTC = ZoneId.of("UTC");
+
+ @Test
+ void testProcessTimeProviderUsesPreparedBatchTimestamp() {
+ RowTtlTimestampProvider provider =
+ RowTtlTimestampProvider.create(
+ processTimeTableConfig(), schemaGetter(schema()), UTC);
+
+ provider.prepareForBatch(1234L);
+
+ assertThat(provider.applyAsLong(row(1, 100L))).isEqualTo(1234L);
+
+ provider.prepareForBatch(5678L);
+ assertThat(provider.applyAsLong(row(2, 200L))).isEqualTo(5678L);
+ }
+
+ @Test
+ void testBigintEventTimeProviderReadsEpochMillis() {
+ RowTtlTimestampProvider provider =
+ RowTtlTimestampProvider.create(eventTimeTableConfig(), schemaGetter(schema()), UTC);
+
+ provider.prepareForBatch(999L);
+
+ assertThat(provider.applyAsLong(row(1, 1234L))).isEqualTo(1234L);
+ }
+
+ @Test
+ void testTimestampLtzEventTimeProviderReadsEpochMillis() {
+ Schema timestampLtzSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("event_time", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("id")
+ .build();
+ RowTtlTimestampProvider provider =
+ RowTtlTimestampProvider.create(
+ eventTimeTableConfig(), schemaGetter(timestampLtzSchema), UTC);
+
+ BinaryRow row =
+ compactedRow(
+ timestampLtzSchema.getRowType(),
+ new Object[] {1, TimestampLtz.fromEpochMillis(5678L)});
+
+ assertThat(provider.applyAsLong(row)).isEqualTo(5678L);
+ }
+
+ @Test
+ void testTimestampEventTimeProviderUsesServerTimeZone() {
+ ZoneId serverTimeZone = ZoneId.of("Asia/Shanghai");
+ LocalDateTime eventTime = LocalDateTime.of(2026, 8, 25, 10, 0);
+ Schema timestampSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("event_time", DataTypes.TIMESTAMP(3))
+ .primaryKey("id")
+ .build();
+ RowTtlTimestampProvider provider =
+ RowTtlTimestampProvider.create(
+ eventTimeTableConfig(), schemaGetter(timestampSchema), serverTimeZone);
+
+ BinaryRow row =
+ compactedRow(
+ timestampSchema.getRowType(),
+ new Object[] {1, TimestampNtz.fromLocalDateTime(eventTime)});
+
+ assertThat(provider.applyAsLong(row))
+ .isEqualTo(eventTime.atZone(serverTimeZone).toInstant().toEpochMilli());
+ }
+
+ @Test
+ void testNullEventTimeNeverExpires() {
+ RowTtlTimestampProvider provider =
+ RowTtlTimestampProvider.create(eventTimeTableConfig(), schemaGetter(schema()), UTC);
+
+ assertThat(provider.applyAsLong(row(1, null)))
+ .isEqualTo(RowTtlTimestampProvider.NEVER_EXPIRE_TIMESTAMP_MS);
+ }
+
+ private static TableConfig processTimeTableConfig() {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.TABLE_KV_TTL, java.time.Duration.ofHours(1));
+ return new TableConfig(configuration);
+ }
+
+ private static TableConfig eventTimeTableConfig() {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.TABLE_KV_TTL, java.time.Duration.ofHours(1));
+ configuration.setString(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN, "event_time");
+ return new TableConfig(configuration);
+ }
+
+ private static TestingSchemaGetter schemaGetter(Schema schema) {
+ return new TestingSchemaGetter(new SchemaInfo(schema, SCHEMA_ID));
+ }
+
+ private static Schema schema() {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("event_time", DataTypes.BIGINT())
+ .primaryKey("id")
+ .build();
+ }
+
+ private static BinaryRow row(int id, Long eventTime) {
+ return compactedRow(schema().getRowType(), new Object[] {id, eventTime});
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilderTest.java
index 8106af9bf..7261fe436 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilderTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilderTest.java
@@ -17,15 +17,21 @@
package org.apache.fluss.server.kv.rocksdb;
+import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.server.exception.KvBuildingException;
+import org.apache.fluss.server.kv.RowTtlCompactionFilterFactory;
+import org.apache.fluss.utils.clock.ManualClock;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.rocksdb.FlinkCompactionFilter;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+import java.time.Duration;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -59,13 +65,47 @@ class RocksDBKvBuilderTest {
}
@Test
- void testBuildFail() {
+ void testCompactionFilterFactoryClosedWithKv(@TempDir Path tempDir) throws Exception {
+ FlinkCompactionFilter.FlinkCompactionFilterFactory filterFactory =
+ RowTtlCompactionFilterFactory.create(
+ KvValueLayout.TAGGED, Duration.ofHours(1L), new ManualClock(0L));
+ RocksDBResourceContainer rocksDBResourceContainer =
+ new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+ RocksDBKvBuilder rocksDBKvBuilder =
+ new RocksDBKvBuilder(
+ tempDir.toFile(),
+ rocksDBResourceContainer,
+ rocksDBResourceContainer.getColumnOptions())
+ .setCompactionFilterFactory(filterFactory);
+
+ try {
+ try (RocksDBKv ignored = rocksDBKvBuilder.build()) {
+ assertThat(filterFactory.isOwningHandle()).isTrue();
+ }
+ assertThat(filterFactory.isOwningHandle()).isFalse();
+ } finally {
+ filterFactory.close();
+ }
+ }
+
+ @Test
+ void testCompactionFilterFactoryClosedWhenBuildFails() {
+ FlinkCompactionFilter.FlinkCompactionFilterFactory filterFactory =
+ RowTtlCompactionFilterFactory.create(
+ KvValueLayout.TAGGED, Duration.ofHours(1L), new ManualClock(0L));
RocksDBResourceContainer rocksDBResourceContainer = new RocksDBResourceContainer();
RocksDBKvBuilder rocksDBKvBuilder =
new RocksDBKvBuilder(
- null,
- rocksDBResourceContainer,
- rocksDBResourceContainer.getColumnOptions());
- assertThatThrownBy(rocksDBKvBuilder::build).isInstanceOf(KvBuildingException.class);
+ null,
+ rocksDBResourceContainer,
+ rocksDBResourceContainer.getColumnOptions())
+ .setCompactionFilterFactory(filterFactory);
+
+ try {
+ assertThatThrownBy(rocksDBKvBuilder::build).isInstanceOf(KvBuildingException.class);
+ assertThat(filterFactory.isOwningHandle()).isFalse();
+ } finally {
+ filterFactory.close();
+ }
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
index 3d0ca53b4..b3c1e210b 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
@@ -21,11 +21,15 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
import org.apache.fluss.exception.NotLeaderOrFollowerException;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.KvRecord;
import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.PbLookupRespForBucket;
import org.apache.fluss.rpc.messages.PutKvRequest;
@@ -36,6 +40,7 @@ import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.server.zk.data.TableAssignment;
+import org.apache.fluss.utils.BytesUtils;
import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.types.Tuple2;
@@ -43,13 +48,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
import static org.apache.fluss.server.testutils.KvTestUtils.assertLookupResponse;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest;
@@ -58,6 +64,9 @@ import static org.apache.fluss.testutils.DataTestUtils.genKvRecords;
import static org.apache.fluss.testutils.DataTestUtils.getKeyValuePairs;
import static org.apache.fluss.testutils.DataTestUtils.toKvRecordBatch;
import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.apache.fluss.types.DataTypes.INT;
+import static org.apache.fluss.types.DataTypes.STRING;
+import static org.apache.fluss.types.DataTypes.TIMESTAMP;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -82,6 +91,15 @@ import static org.assertj.core.api.Assertions.assertThat;
* </ol>
*/
class KvRecoverFromRemoteLogITCase {
+ private static final Schema EVENT_TIME_SCHEMA =
+ Schema.newBuilder()
+ .column("a", INT())
+ .column("event_time", TIMESTAMP(3))
+ .column("b", STRING())
+ .primaryKey("a")
+ .build();
+ private static final LocalDateTime EVENT_TIME = LocalDateTime.of(2099, 1, 1, 0, 0);
+
@RegisterExtension
public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
FlussClusterExtension.builder()
@@ -94,7 +112,12 @@ class KvRecoverFromRemoteLogITCase {
// Step 1: Create a PK table with 1 bucket and replication factor 3.
TablePath tablePath = TablePath.of("test_db", "test_kv_recover_from_remote");
TableDescriptor tableDescriptor =
- TableDescriptor.builder().schema(DATA1_SCHEMA_PK).distributedBy(1, "a").build();
+ TableDescriptor.builder()
+ .schema(EVENT_TIME_SCHEMA)
+ .distributedBy(1, "a")
+ .property(ConfigOptions.TABLE_KV_TTL.key(), "1 h")
+ .property(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), "event_time")
+ .build();
long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor);
TableBucket tableBucket = new TableBucket(tableId, 0);
@@ -139,7 +162,12 @@ class KvRecoverFromRemoteLogITCase {
// Write the same key multiple times per batch to increase log volume,
// ensuring multiple log segments are produced quickly for remote tiering.
for (int j = 0; j < 10; j++) {
- batch.addAll(genKvRecords(new Object[] {i, "value_" + i}));
+ batch.addAll(
+ genKvRecords(
+ EVENT_TIME_SCHEMA.getRowType(),
+ new Object[] {
+ i, TimestampNtz.fromLocalDateTime(EVENT_TIME), "value_" + i
+ }));
}
allRecords.addAll(batch);
putRecordBatchAndAssertSuccess(tableBucket, originalLeader, toKvRecordBatch(batch));
@@ -161,7 +189,12 @@ class KvRecoverFromRemoteLogITCase {
List<KvRecord> batch = new ArrayList<>();
// Same as Phase 1: write the same key multiple times to produce more log data.
for (int j = 0; j < 10; j++) {
- batch.addAll(genKvRecords(new Object[] {i, "value_" + i}));
+ batch.addAll(
+ genKvRecords(
+ EVENT_TIME_SCHEMA.getRowType(),
+ new Object[] {
+ i, TimestampNtz.fromLocalDateTime(EVENT_TIME), "value_" + i
+ }));
}
allRecords.addAll(batch);
putRecordBatchAndAssertSuccess(tableBucket, originalLeader, toKvRecordBatch(batch));
@@ -178,6 +211,14 @@ class KvRecoverFromRemoteLogITCase {
Duration.ofMinutes(2),
"Fail to wait for remote log to cover beyond snapshot offset " + snapshotLogOffset);
+ byte[] encodedKey = BytesUtils.toArray(allRecords.get(allRecords.size() - 1).getKey());
+ assertThat(leaderReplica.getKvTablet()).isNotNull();
+ byte[] leaderValue =
+ leaderReplica.getKvTablet().multiGet(Collections.singletonList(encodedKey)).get(0);
+ long originalTag = KvValueLayout.TAGGED.readValueTag(MemorySegment.wrap(leaderValue));
+ assertThat(originalTag)
+ .isEqualTo(EVENT_TIME.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+
// =====================================================================
// Step 3: Restart follower → catches up via remote fetch path
// =====================================================================
@@ -277,6 +318,15 @@ class KvRecoverFromRemoteLogITCase {
keyValue.f1);
}
+ assertThat(followerReplica.getKvTablet()).isNotNull();
+ byte[] recovered =
+ followerReplica
+ .getKvTablet()
+ .multiGet(Collections.singletonList(encodedKey))
+ .get(0);
+ assertThat(KvValueLayout.TAGGED.readValueTag(MemorySegment.wrap(recovered)))
+ .isEqualTo(originalTag);
+
// Clean up: restart the other follower to restore cluster health.
FLUSS_CLUSTER_EXTENSION.startTabletServer(otherFollower);
FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index da8be8590..52eb7d508 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -35,6 +35,7 @@ import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.ChangeType;
@@ -50,6 +51,7 @@ import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
@@ -91,6 +93,7 @@ import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.types.Tuple2;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -128,6 +131,7 @@ import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
@@ -175,6 +179,11 @@ class ReplicaManagerTest extends ReplicaTestBase {
/** First PUT_KV version that understands the STORAGE_BACKPRESSURE_EXCEPTION error code. */
private static final short PUT_KV_VERSION_WITH_STORAGE_BACKPRESSURE = 2;
+ @BeforeEach
+ void resetDiskUsageForTest() {
+ replicaManager.getDiskUsageMonitor().update(0.5);
+ }
+
@Test
void testProduceLog() throws Exception {
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1);
@@ -1097,7 +1106,8 @@ class ReplicaManagerTest extends ReplicaTestBase {
// Decode values to verify auto-increment column values
TestingSchemaGetter schemaGetter =
new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA3_SCHEMA_PK_AUTO_INC);
- ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, KvFormat.COMPACTED);
+ ValueDecoder valueDecoder =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.PLAIN);
InternalRow row1 = valueDecoder.decodeValue(inserted.get(0)).row;
InternalRow row2 = valueDecoder.decodeValue(inserted.get(1)).row;
@@ -1204,7 +1214,8 @@ class ReplicaManagerTest extends ReplicaTestBase {
// Verify auto-increment values are sequential and unique
TestingSchemaGetter schemaGetter =
new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA3_SCHEMA_PK_AUTO_INC);
- ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, KvFormat.COMPACTED);
+ ValueDecoder valueDecoder =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.PLAIN);
Set<Long> autoIncrementValues = new HashSet<>();
for (byte[] value : threadResults[0]) {
@@ -1445,6 +1456,53 @@ class ReplicaManagerTest extends ReplicaTestBase {
assertThat(future.get().getValues()).isEqualTo(builder.build());
}
+ @Test
+ void testTaggedLookupAndPrefixResultsCarryLayout() throws Exception {
+ TablePath tablePath = TablePath.of("test_db", "ttl_raw_kv_table");
+ long tableId = 150007L;
+ registerTaggedTable(tablePath, tableId);
+
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ makeKvTableAsLeader(tableId, tablePath, tableBucket.getBucket());
+
+ CompletableFuture<List<PutKvResultForBucket>> putFuture = new CompletableFuture<>();
+ replicaManager.putRecordsToKv(
+ 20000,
+ 1,
+ Collections.singletonMap(tableBucket, genKvRecordBatch(DATA_1_WITH_KEY_AND_VALUE)),
+ null,
+ MergeMode.DEFAULT,
+ PUT_KV_VERSION,
+ putFuture::complete);
+ PutKvResultForBucket putResult = putFuture.get().get(0);
+ assertThat(putResult.failed()).isFalse();
+
+ CompactedKeyEncoder keyEncoder = new CompactedKeyEncoder(DATA1_ROW_TYPE, new int[] {0});
+ byte[] keyBytes = keyEncoder.encodeKey(row(DATA_1_WITH_KEY_AND_VALUE.get(0).f0));
+
+ CompletableFuture<Map<TableBucket, LookupResultForBucket>> lookupFuture =
+ new CompletableFuture<>();
+ replicaManager.lookups(
+ Collections.singletonMap(tableBucket, Collections.singletonList(keyBytes)),
+ LOOKUP_KV_VERSION,
+ lookupFuture::complete);
+ LookupResultForBucket lookupResult = lookupFuture.get().get(tableBucket);
+ assertThat(lookupResult.failed()).isFalse();
+ assertThat(lookupResult.getKvValueLayout()).isSameAs(KvValueLayout.TAGGED);
+ assertThat(lookupResult.lookupValues().get(0)).hasSizeGreaterThan(Long.BYTES);
+
+ CompletableFuture<Map<TableBucket, PrefixLookupResultForBucket>> prefixFuture =
+ new CompletableFuture<>();
+ replicaManager.prefixLookups(
+ Collections.singletonMap(tableBucket, Collections.singletonList(keyBytes)),
+ PREFIX_LOOKUP_KV_VERSION,
+ prefixFuture::complete);
+ PrefixLookupResultForBucket prefixResult = prefixFuture.get().get(tableBucket);
+ assertThat(prefixResult.failed()).isFalse();
+ assertThat(prefixResult.getKvValueLayout()).isSameAs(KvValueLayout.TAGGED);
+ assertThat(prefixResult.prefixLookupValues().get(0).get(0)).hasSizeGreaterThan(Long.BYTES);
+ }
+
@Test
void testLimitScanLogTable() throws Exception {
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
@@ -2506,6 +2564,23 @@ class ReplicaManagerTest extends ReplicaTestBase {
}
}
+ private void registerTaggedTable(TablePath tablePath, long tableId) throws Exception {
+ Map<String, String> properties = new HashMap<>(DATA1_TABLE_DESCRIPTOR_PK.getProperties());
+ properties.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+ TableDescriptor descriptor = DATA1_TABLE_DESCRIPTOR_PK.withProperties(properties);
+
+ if (zkClient.tableExist(tablePath)) {
+ zkClient.deleteTable(tablePath);
+ }
+ zkClient.registerTable(
+ tablePath,
+ TableRegistration.newTable(tableId, DEFAULT_REMOTE_DATA_DIR, descriptor));
+ zkClient.registerFirstSchema(tablePath, DATA1_SCHEMA_PK);
+ }
+
@Test
void testKvFormatV2TableRejectV0Client() throws Exception {
// Test that old client version (version 0) is rejected for new format tables
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index 2159e7ca2..66351f408 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -237,7 +237,8 @@ public class ReplicaTestBase {
logManager,
TestingMetricGroups.TABLET_SERVER_METRICS,
localDiskManager,
- createTestKvFlushScheduler(conf));
+ createTestKvFlushScheduler(conf),
+ manualClock);
kvManager.startup();
serverMetadataCache =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java
index 9c69edabe..319cba3a3 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java
@@ -44,6 +44,7 @@ import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.row.BinaryString;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
@@ -84,6 +85,7 @@ import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -239,7 +241,7 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
@Test
void testHistoricalInsertUpdateAndDelete() throws Exception {
- TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader(true);
Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
KvTablet kvTablet = replica.getKvTablet();
assertThat(kvTablet).isNotNull();
@@ -377,7 +379,8 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
BinaryValue lookedUpValue =
new ValueDecoder(
schemaGetter(tableInfo),
- tableInfo.getTableConfig().getKvFormat())
+ tableInfo.getTableConfig().getKvFormat(),
+ lookupResult.getKvValueLayout(0))
.decodeValue(lookupResult.lookupValues().get(0));
assertThat(lookedUpValue.row.getString(3)).isEqualTo(BinaryString.fromString("v2"));
@@ -790,8 +793,18 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
return registerHistoricalTableAndBecomeLeader(ChangelogImage.FULL);
}
+ private TableInfo registerHistoricalTableAndBecomeLeader(boolean rowTtlEnabled)
+ throws Exception {
+ return registerHistoricalTableAndBecomeLeader(ChangelogImage.FULL, rowTtlEnabled);
+ }
+
private TableInfo registerHistoricalTableAndBecomeLeader(ChangelogImage changelogImage)
throws Exception {
+ return registerHistoricalTableAndBecomeLeader(changelogImage, false);
+ }
+
+ private TableInfo registerHistoricalTableAndBecomeLeader(
+ ChangelogImage changelogImage, boolean rowTtlEnabled) throws Exception {
replicaManager.getDiskUsageMonitor().update(0.10);
Schema schema =
Schema.newBuilder()
@@ -801,7 +814,7 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
.column("value", DataTypes.STRING())
.primaryKey("id", "region", "dt")
.build();
- TableDescriptor descriptor =
+ TableDescriptor.Builder descriptorBuilder =
TableDescriptor.builder()
.schema(schema)
.distributedBy(1, "id")
@@ -819,8 +832,15 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
.property(ConfigOptions.TABLE_CHANGELOG_IMAGE, changelogImage)
.property(
ConfigOptions.TABLE_KV_FORMAT_VERSION,
- ConfigOptions.KV_FORMAT_VERSION_2)
- .build();
+ ConfigOptions.KV_FORMAT_VERSION_2);
+ if (rowTtlEnabled) {
+ descriptorBuilder
+ .property(ConfigOptions.TABLE_KV_TTL, Duration.ofDays(1))
+ .property(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION,
+ KvValueLayout.TAGGED.version());
+ }
+ TableDescriptor descriptor = descriptorBuilder.build();
TableInfo tableInfo =
TableInfo.of(TABLE_PATH, TABLE_ID, 1, descriptor, DEFAULT_REMOTE_DATA_DIR, 1L, 1L);
zkClient.registerTable(
@@ -933,7 +953,10 @@ class HistoricalPartitionManagerTest extends ReplicaTestBase {
KvStateLookupResult result = kvTablet.lookupHistoricalLocal(originalPartition, primaryKey);
assertThat(result.isPresent()).isTrue();
BinaryValue value =
- new ValueDecoder(schemaGetter(tableInfo), tableInfo.getTableConfig().getKvFormat())
+ new ValueDecoder(
+ schemaGetter(tableInfo),
+ tableInfo.getTableConfig().getKvFormat(),
+ KvValueLayout.fromTableConfig(tableInfo.getTableConfig()))
.decodeValue(result.value());
assertThatRow(value.row).withSchema(tableInfo.getRowType()).isEqualTo(expectedRow);
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
index 6190f566d..84708d0b6 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
@@ -32,6 +32,7 @@ import org.apache.fluss.record.DefaultValueRecordBatch;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
+import org.apache.fluss.row.encode.KvValueLayout;
import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
@@ -78,8 +79,10 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
@@ -493,7 +496,9 @@ public class TabletServiceITCase {
void testLookup() throws Exception {
long tableId =
createTable(
- FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH_PK, DATA1_TABLE_DESCRIPTOR_PK);
+ FLUSS_CLUSTER_EXTENSION,
+ DATA1_TABLE_PATH_PK,
+ withRowTtl(DATA1_TABLE_DESCRIPTOR_PK));
TableBucket tb = new TableBucket(tableId, 0);
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
@@ -582,7 +587,11 @@ public class TabletServiceITCase {
new DataField("c", DataTypes.BIGINT()));
TableDescriptor descriptor =
- TableDescriptor.builder().schema(schema).distributedBy(3, "a", "b").build();
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(3, "a", "b")
+ .property(ConfigOptions.TABLE_KV_TTL.key(), "1 h")
+ .build();
long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, descriptor);
TableBucket tb = new TableBucket(tableId, 0);
@@ -683,7 +692,9 @@ public class TabletServiceITCase {
void testLimitScanPrimaryKeyTable() throws Exception {
long tableId =
createTable(
- FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH_PK, DATA1_TABLE_DESCRIPTOR_PK);
+ FLUSS_CLUSTER_EXTENSION,
+ DATA1_TABLE_PATH_PK,
+ withRowTtl(DATA1_TABLE_DESCRIPTOR_PK));
TableBucket tb = new TableBucket(tableId, 0);
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
@@ -1049,7 +1060,8 @@ public class TabletServiceITCase {
TabletServerGateway gateway = FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader);
CompactedKeyEncoder keyEncoder = new CompactedKeyEncoder(rowType, new int[] {0});
TestingSchemaGetter schemaGetter = new TestingSchemaGetter(DEFAULT_SCHEMA_ID, schema);
- ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, KvFormat.COMPACTED);
+ ValueDecoder valueDecoder =
+ new ValueDecoder(schemaGetter, KvFormat.COMPACTED, KvValueLayout.PLAIN);
byte[] key1 = keyEncoder.encodeKey(row(new Object[] {100}));
byte[] key2 = keyEncoder.encodeKey(row(new Object[] {200}));
@@ -1082,7 +1094,9 @@ public class TabletServiceITCase {
void testScanKv_newScan_happyPath() throws Exception {
long tableId =
createTable(
- FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH_PK, DATA1_TABLE_DESCRIPTOR_PK);
+ FLUSS_CLUSTER_EXTENSION,
+ DATA1_TABLE_PATH_PK,
+ withRowTtl(DATA1_TABLE_DESCRIPTOR_PK));
TableBucket tb = new TableBucket(tableId, 0);
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
int leader = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb);
@@ -1107,7 +1121,10 @@ public class TabletServiceITCase {
assertThat(response.getLogOffset()).isGreaterThanOrEqualTo(0L);
assertThat(response.hasRecords()).isTrue();
DefaultValueRecordBatch batch = DefaultValueRecordBatch.pointToBytes(response.getRecords());
- assertThat(batch.getRecordCount()).isEqualTo(2);
+ DefaultValueRecordBatch.Builder expected = DefaultValueRecordBatch.builder();
+ expected.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"}));
+ expected.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "b1"}));
+ assertThat(batch).isEqualTo(expected.build());
}
@Test
@@ -1434,6 +1451,12 @@ public class TabletServiceITCase {
return req;
}
+ private static TableDescriptor withRowTtl(TableDescriptor descriptor) {
+ Map<String, String> properties = new HashMap<>(descriptor.getProperties());
+ properties.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ return descriptor.withProperties(properties);
+ }
+
private static ScanKvRequest newScanKvContinueRequest(
byte[] scannerId, int callSeqId, int batchSize) {
ScanKvRequest req = new ScanKvRequest();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/RowTtlUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/RowTtlUtilsTest.java
new file mode 100644
index 000000000..cd865ab5f
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/RowTtlUtilsTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.server.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link RowTtlUtils}. */
+class RowTtlUtilsTest {
+
+ @Test
+ void testConvertTtlDurationToMillis() {
+ assertThat(RowTtlUtils.validateAndConvertTtlDurationToMillis(Duration.ofNanos(1_500_000)))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void testRejectsSubMillisecondTtl() {
+ assertThatThrownBy(
+ () ->
+ RowTtlUtils.validateAndConvertTtlDurationToMillis(
+ Duration.ofNanos(999_999)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testRejectsTtlMillisOverflow() {
+ assertThatThrownBy(
+ () ->
+ RowTtlUtils.validateAndConvertTtlDurationToMillis(
+ Duration.ofSeconds(Long.MAX_VALUE)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
index c9e6f2304..e22567951 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
@@ -19,7 +19,10 @@ package org.apache.fluss.server.utils;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.rpc.entity.LookupResultForBucket;
import org.apache.fluss.rpc.entity.PutKvResultForBucket;
+import org.apache.fluss.rpc.messages.LookupResponse;
import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
import org.apache.fluss.rpc.messages.PbPutKvRespForBucket;
import org.apache.fluss.rpc.messages.PutKvRequest;
@@ -41,6 +44,25 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link ServerRpcMessageUtils}. */
class ServerRpcMessageUtilsTest {
+ @Test
+ void testLookupResponseSupportsMixedValueLayouts() {
+ byte[] plainValue = new byte[] {1, 0, 11, 12};
+ byte[] taggedValue = new byte[] {0, 0, 0, 0, 0, 0, 0, 7, 1, 0, 21, 22};
+ LookupResultForBucket result =
+ new LookupResultForBucket(
+ new TableBucket(1L, 0),
+ Arrays.asList(plainValue, taggedValue),
+ Arrays.asList(KvValueLayout.PLAIN, KvValueLayout.TAGGED),
+ "dt=2025-01-01");
+
+ LookupResponse response =
+ ServerRpcMessageUtils.makeLookupResponse(Collections.singletonList(result));
+
+ assertThat(response.getBucketsRespAt(0).getValuesList())
+ .extracting(value -> value.getValues())
+ .containsExactly(new byte[] {1, 0, 11, 12}, new byte[] {1, 0, 21, 22});
+ }
+
@Test
void testHistoricalPutKvRequestAndResponsePreserveOriginalPartitions() throws Exception {
long tableId = 1L;
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java
new file mode 100644
index 000000000..3772197fb
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java
@@ -0,0 +1,316 @@
+/*
+ * 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.server.utils;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.InvalidAlterTableException;
+import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.exception.InvalidTableException;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.TestData;
+import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TableDescriptorValidation}. */
+class TableDescriptorValidationTest {
+
+ @Test
+ void testCreateLogTableWithKvTTLFails() {
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ logTableWithProperty(
+ ConfigOptions.TABLE_KV_TTL.key(), "1 h"),
+ 100,
+ null))
+ .isInstanceOf(InvalidTableException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL.key())
+ .hasMessageContaining("primary key");
+ }
+
+ @ParameterizedTest
+ @MethodSource("supportedKvTTLTimeColumnTypes")
+ void testCreateTableWithSupportedKvTTLTimeColumn(DataType timeColumnType) {
+ TableDescriptor descriptor = pkTableWithKvTTLTimeColumn("event_time", timeColumnType);
+
+ TableDescriptorValidation.validateTableDescriptor(descriptor, 100, null);
+ }
+
+ @Test
+ void testCreateTableWithMissingKvTTLTimeColumnFails() {
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(TestData.DATA1_SCHEMA_PK)
+ .distributedBy(3)
+ .property(ConfigOptions.TABLE_KV_TTL.key(), "1 h")
+ .property(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), "event_time")
+ .property(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()))
+ .build()
+ .withReplicationFactor(3);
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ descriptor, 100, null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key());
+ }
+
+ @Test
+ void testKvTTLTimeColumnRequiresKvTTL() {
+ TableDescriptor descriptor =
+ pkTableWithProperty(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), "event_time");
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ descriptor, 100, null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key())
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL.key());
+ }
+
+ @ParameterizedTest
+ @MethodSource("unsupportedKvTTLTimeColumnTypes")
+ void testCreateTableWithUnsupportedKvTTLTimeColumnFails(DataType timeColumnType) {
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ pkTableWithKvTTLTimeColumn("event_time", timeColumnType),
+ 100,
+ null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key())
+ .hasMessageContaining("BIGINT")
+ .hasMessageContaining("TIMESTAMP_LTZ");
+ }
+
+ @Test
+ void testCreateTableWithLargeKvTTL() {
+ TableDescriptor descriptor =
+ pkTableWithProperties(
+ ConfigOptions.TABLE_KV_TTL.key(),
+ "3000000000 s",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+
+ TableDescriptorValidation.validateTableDescriptor(descriptor, 100, null);
+ }
+
+ @Test
+ void testCreateTableWithKvTTLMillisOverflowFails() {
+ TableDescriptor descriptor =
+ pkTableWithProperties(
+ ConfigOptions.TABLE_KV_TTL.key(),
+ "9223372036854776 s",
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()));
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ descriptor, 100, null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL.key())
+ .hasMessageContaining("exceeds");
+ }
+
+ @Test
+ void testKvFormatVersionStillRejectsValuesAboveVersionTwo() {
+ TableDescriptor descriptor =
+ pkTableWithProperty(
+ ConfigOptions.TABLE_KV_FORMAT_VERSION.key(),
+ String.valueOf(KV_FORMAT_VERSION_2 + 1));
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ descriptor, 100, null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining("maximum supported version is " + KV_FORMAT_VERSION_2);
+ }
+
+ @ParameterizedTest(name = "layoutVersion={0}, rowTtl={1}, valid={2}")
+ @MethodSource("kvValueLayoutMatrix")
+ void testKvValueLayoutMatrix(Integer layoutVersion, boolean rowTtl, boolean valid) {
+ Map<String, String> properties = new HashMap<>();
+ if (layoutVersion != null) {
+ properties.put(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(layoutVersion));
+ }
+ if (rowTtl) {
+ properties.put(ConfigOptions.TABLE_KV_TTL.key(), "1 h");
+ }
+ TableDescriptor descriptor = pkTableWithProperties(properties);
+
+ if (valid) {
+ TableDescriptorValidation.validateTableDescriptor(descriptor, 100, null);
+ } else {
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateTableDescriptor(
+ descriptor, 100, null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key());
+ }
+ }
+
+ @Test
+ void testAlterKvTTLFails() {
+ TableInfo currentTable =
+ TableInfo.of(
+ TablePath.of("db", "t"),
+ 1L,
+ 1,
+ pkTableWithProperty(ConfigOptions.TABLE_KV_TTL.key(), "1 h"),
+ "file://remote",
+ 1L,
+ 1L);
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateAlterTableProperties(
+ currentTable,
+ Collections.singleton(ConfigOptions.TABLE_KV_TTL.key())))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_TTL.key());
+ }
+
+ @Test
+ void testAlterValueLayoutVersionFails() {
+ TableInfo currentTable = plainTableInfo();
+
+ assertThatThrownBy(
+ () ->
+ TableDescriptorValidation.validateAlterTableProperties(
+ currentTable,
+ Collections.singleton(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key())))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining(ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key());
+ }
+
+ private static Stream<Arguments> supportedKvTTLTimeColumnTypes() {
+ return Stream.of(
+ Arguments.of(DataTypes.BIGINT()),
+ Arguments.of(DataTypes.TIMESTAMP()),
+ Arguments.of(DataTypes.TIMESTAMP_LTZ()));
+ }
+
+ private static Stream<Arguments> unsupportedKvTTLTimeColumnTypes() {
+ return Stream.of(Arguments.of(DataTypes.STRING()));
+ }
+
+ private static Stream<Arguments> kvValueLayoutMatrix() {
+ return Stream.of(
+ Arguments.of(null, false, true),
+ Arguments.of(null, true, false),
+ Arguments.of(KvValueLayout.PLAIN.version(), false, true),
+ Arguments.of(KvValueLayout.PLAIN.version(), true, false),
+ Arguments.of(KvValueLayout.TAGGED.version(), true, true),
+ Arguments.of(KvValueLayout.TAGGED.version(), false, false),
+ Arguments.of(999, true, false));
+ }
+
+ private TableDescriptor pkTableWithProperty(String key, String value) {
+ return TableDescriptor.builder()
+ .schema(TestData.DATA1_SCHEMA_PK)
+ .distributedBy(3)
+ .property(key, value)
+ .build()
+ .withReplicationFactor(3);
+ }
+
+ private TableDescriptor pkTableWithProperties(
+ String key1, String value1, String key2, String value2) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(key1, value1);
+ properties.put(key2, value2);
+ return pkTableWithProperties(properties);
+ }
+
+ private TableDescriptor pkTableWithProperties(Map<String, String> properties) {
+ return TableDescriptor.builder()
+ .schema(TestData.DATA1_SCHEMA_PK)
+ .distributedBy(3)
+ .properties(properties)
+ .build()
+ .withReplicationFactor(3);
+ }
+
+ private TableDescriptor pkTableWithKvTTLTimeColumn(String timeColumn, DataType timeColumnType) {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(timeColumn, timeColumnType)
+ .primaryKey("id")
+ .build();
+ return TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(3)
+ .property(ConfigOptions.TABLE_KV_TTL.key(), "1 h")
+ .property(ConfigOptions.TABLE_KV_TTL_TIME_COLUMN.key(), timeColumn)
+ .property(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.TAGGED.version()))
+ .build()
+ .withReplicationFactor(3);
+ }
+
+ private TableDescriptor logTableWithProperty(String key, String value) {
+ return TableDescriptor.builder()
+ .schema(TestData.DATA1_SCHEMA)
+ .distributedBy(3)
+ .property(key, value)
+ .build()
+ .withReplicationFactor(3);
+ }
+
+ private TableInfo plainTableInfo() {
+ return TableInfo.of(
+ TablePath.of("db", "t"),
+ 1L,
+ 1,
+ pkTableWithProperty(
+ ConfigOptions.TABLE_KV_VALUE_LAYOUT_VERSION.key(),
+ String.valueOf(KvValueLayout.PLAIN.version())),
+ "file://remote",
+ 1L,
+ 1L);
+ }
+}
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index c54464825..8c8dd2b9a 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -68,6 +68,8 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties)
| bucket.key | String | (None) | Specific the distribution policy of the Fluss table. Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). If you specify multiple fields, delimiter is `,`. If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding th [...]
| table.log.ttl | Duration | 7 days | The time to live for log segments. The configuration controls the maximum time log segments are retained before they become eligible for deletion. When remote log tiering is enabled, this value controls the retention of remote log segments. Setting the value to '0ms' disables TTL-based deletion. The default value is 7 days. [...]
| table.log.local-ttl | Duration | (None) | The time to live for local log segments. The configuration controls the maximum time local log segments are retained before they become eligible for deletion. When remote log tiering is enabled, an expired local segment is deleted only after it has been copied to remote storage. Setting the value to '0ms' disables TTL-based deletion. If not configured, the value inherits `table.log.ttl`. When bot [...]
+| table.kv.ttl | Duration | (None) | The best-effort row-level TTL for primary key data. The changelog retention of a primary key table is controlled separately by `table.log.ttl`. If this option is not set, row-level TTL is disabled. The value must be at least 1 millisecond. Expired rows may remain visible until RocksDB compaction removes them. TTL uses processing time by default, or event time when `table.kv.ttl.time-column` is co [...]
+| table.kv.ttl.time-column | String | (None) | The event-time column for row-level TTL. If this option is not set, row TTL uses processing time. If set, `table.kv.ttl` must also be set, and the column must be `BIGINT` epoch milliseconds, `TIMESTAMP`, or `TIMESTAMP_LTZ`. `TIMESTAMP` values are interpreted in the TabletServer's system time zone, so all TabletServers must use the same system time zone. Rows with null event-time values do not exp [...]
| table.auto-partition.enabled | Boolean | false | Whether enable auto partition for the table. Disable by default. When auto partition is enabled, the partitions of the table will be created automatically. [...]
| table.auto-partition.key | String | (None) | This configuration defines the time-based partition key to be used for auto-partitioning when a table is partitioned with multiple keys. Auto-partitioning utilizes a time-based partition key to handle partitions automatically, including creating new ones and removing outdated ones, by comparing the time value of the partition with the current system time. In the case of a table using multiple par [...]
| table.auto-partition.time-unit | ENUM | DAY | The time granularity for auto created partitions. The default value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If the value is `DAY`, the partition format for auto created is yyyyMMdd. If the value is `MONTH`, the partition format for auto created is yyyyMM. If the value is `QUARTER`, the parti [...]
diff --git a/website/docs/table-design/data-distribution/ttl.md b/website/docs/table-design/data-distribution/ttl.md
index b20a423db..06eeee5c0 100644
--- a/website/docs/table-design/data-distribution/ttl.md
+++ b/website/docs/table-design/data-distribution/ttl.md
@@ -11,3 +11,49 @@ For log tables, this attribute indicates the expiration time of the log table da
For primary key tables, this attribute indicates the expiration time of the changelog and does not represent the expiration time of the primary key table data. If you also want the data in the primary key table to expire automatically, please use [auto partitioning](partitioning.md#auto-partitioning).
When tiered storage is enabled, `table.log.local-ttl` can be used to control how long copied local log segments are retained. If it is not configured, it falls back to `table.log.ttl` for backward compatibility. A non-positive local TTL disables TTL-based local cleanup. When both TTLs are positive, the local TTL must be less than or equal to `table.log.ttl`.
+
+## Row TTL for Primary Key Tables
+
+Primary key tables can configure row-level TTL with `'table.kv.ttl' = '<duration>'`. The duration must be at least 1 millisecond. The option has no default value; if it is not configured, row-level TTL is disabled.
+
+```sql title="Flink SQL"
+CREATE TABLE pk_table
+(
+ id BIGINT,
+ name STRING,
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket.num' = '4',
+ 'table.kv.ttl' = '7 d'
+);
+```
+
+Row TTL is best-effort cleanup. A row becomes eligible for cleanup after the configured duration, but expired rows may still be visible until RocksDB compaction removes them. Fluss stores the TTL timestamp in the primary-key table value and uses a compaction filter to remove expired rows during RocksDB compaction.
+
+Auto partitioning is the recommended expiration mechanism when data can be partitioned by time, because expiring whole partitions preserves changelog completeness. Row TTL is intended for cases that require per-row cleanup and can accept weaker changelog semantics.
+
+Row TTL cleanup does not emit delete records. Downstream consumers of `$changelog` or `$binlog` will not receive delete changes when rows expire. After compaction removes an expired row, the next write for the same primary key is treated as an insert and does not emit an `UPDATE_BEFORE` for the expired value.
+
+By default, row TTL uses processing time. To use event time, configure `table.kv.ttl.time-column` when creating the table:
+
+```sql title="Flink SQL"
+CREATE TABLE pk_table_with_event_time
+(
+ id BIGINT,
+ event_time BIGINT,
+ name STRING,
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket.num' = '4',
+ 'table.kv.ttl' = '7 d',
+ 'table.kv.ttl.time-column' = 'event_time'
+);
+```
+
+The event-time column must be `BIGINT` epoch milliseconds, `TIMESTAMP`, or `TIMESTAMP_LTZ`. `TIMESTAMP` values are interpreted in the TabletServer's system time zone, so all TabletServers must use the same system time zone. Rows with null event-time values do not expire through row TTL.
+
+Row TTL must be configured when creating the primary key table. Changing or disabling `table.kv.ttl`, or changing `table.kv.ttl.time-column`, with `ALTER TABLE ... SET` or `ALTER TABLE ... RESET` is not supported in this version.
+
+Before creating a row-TTL primary key table, upgrade every server to Fluss 1.0+ so that all servers understand the tagged value layout used for row TTL. After a row-TTL table exists, downgrading to a version < 1.0 that does not understand that layout is not supported.
+
+See [Flink Connector Options](/engine-flink/options.md#storage-options) for the complete row TTL option definitions.