You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/08/20 13:01:11 UTC
(fluss) 02/02: [flink] Name the kv batch read strategy, restore startup guard
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 77d132b0d025ff5b61f79ca752b431cb62633776
Author: Anton Borisov <an...@fresha.com>
AuthorDate: Thu Aug 20 13:00:24 2026 +0200
[flink] Name the kv batch read strategy, restore startup guard
---
.../org/apache/fluss/config/ConfigOptions.java | 28 ++++++------
.../org/apache/fluss/config/KvBatchStrategy.java | 48 +++++++++++++++++++++
.../fluss/flink/source/FlussSourceBuilder.java | 19 +++-----
.../source/enumerator/FlinkSourceEnumerator.java | 27 ++++++------
.../fluss/flink/source/split/KvBatchSplit.java | 13 +++---
.../flink/source/FlinkTableSourceBatchITCase.java | 50 +++++++++++++++++++---
.../fluss/flink/source/FlussSourceITCase.java | 3 +-
.../enumerator/FlinkSourceEnumeratorTest.java | 48 ++++++++++++++++++---
website/docs/engine-flink/options.md | 2 +-
website/docs/engine-flink/reads.md | 49 ++++++++++++++-------
10 files changed, 212 insertions(+), 75 deletions(-)
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 35e78a814..3bb4279da 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
@@ -1543,19 +1543,21 @@ public class ConfigOptions {
+ KV_SCANNER_MAX_BATCH_SIZE.key()
+ "'.");
- public static final ConfigOption<Boolean> CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED =
- key("client.scanner.kv.server-side.enabled")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "Master switch for using the server-side KV scan in bounded reads "
- + "of primary-key tables. When true, bounded primary-key reads "
- + "scan the current KV state directly via the server-side KV scan. "
- + "When false (default), bounded primary-key reads fall back to the "
- + "prior behavior (log-only when lake is enabled, or fail when lake "
- + "is disabled). This option has no effect when a lake snapshot "
- + "already exists: in that case the bounded read always performs "
- + "the lake + Fluss-log union read and this flag is not consulted.");
+ public static final ConfigOption<KvBatchStrategy> CLIENT_SCANNER_KV_BATCH_STRATEGY =
+ key("client.scanner.kv.batch-strategy")
+ .enumType(KvBatchStrategy.class)
+ .defaultValue(KvBatchStrategy.SNAPSHOT_MERGE)
+ .withDescription(
+ "The strategy used for bounded reads of primary-key tables. "
+ + "'snapshot-merge' (default) merges the latest kv snapshot with the "
+ + "bounded changelog range that follows it; the scan is resumable and "
+ + "reflects a single point in time. 'server-scan' scans the live kv state "
+ + "on the tablet server instead, which avoids downloading snapshot files "
+ + "and replaying the changelog, but the scan is not resumable and each "
+ + "bucket is read at the point in time its scanner was opened. "
+ + "This option has no effect when a lake snapshot already exists: the "
+ + "bounded read then performs the lake + Fluss-log union read. "
+ + "Currently only the Flink connector honours this option.");
public static final ConfigOption<Integer> CLIENT_LOOKUP_QUEUE_SIZE =
key("client.lookup.queue-size")
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/KvBatchStrategy.java b/fluss-common/src/main/java/org/apache/fluss/config/KvBatchStrategy.java
new file mode 100644
index 000000000..526453609
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/config/KvBatchStrategy.java
@@ -0,0 +1,48 @@
+/*
+ * 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.config;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Strategy for reading the full state of a primary key table bucket in a bounded scan. */
+@PublicEvolving
+public enum KvBatchStrategy {
+ /**
+ * Merge the latest kv snapshot with the bounded changelog range that follows it. The scan is
+ * resumable and reflects a single point in time across all buckets.
+ */
+ SNAPSHOT_MERGE("snapshot-merge"),
+
+ /**
+ * Scan the live kv state on the tablet server. Avoids downloading snapshot files and replaying
+ * the changelog, but the scan is not resumable and each bucket is read at the point in time its
+ * scanner was opened.
+ */
+ SERVER_SCAN("server-scan");
+
+ private final String name;
+
+ KvBatchStrategy(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
index e0fbb1a29..ca6914533 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
@@ -176,10 +176,7 @@ public class FlussSourceBuilder<OUT> {
* Builds a bounded source for batch execution. The source reads up to the latest offsets at job
* startup and then finishes; combined with the default {@link OffsetsInitializer#full()} on a
* datalake-enabled table this performs a bounded union read of the lake snapshot and the Fluss
- * log. Alternatively, a bounded read of a primary-key table is supported via the server-side KV
- * scan by setting {@code client.scanner.kv.server-side.enabled = true} (see {@link
- * ConfigOptions#CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED}). If not called, the source is unbounded
- * (streaming).
+ * log. If not called, the source is unbounded (streaming).
*
* @return this builder
*/
@@ -356,19 +353,13 @@ public class FlussSourceBuilder<OUT> {
boolean lakeEnabled = tableInfo.getTableConfig().isDataLakeEnabled();
boolean fullStartup = offsetsInitializer instanceof SnapshotOffsetsInitializer;
- boolean kvBatchEnabled = flussConf.get(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED);
-
- // The server-side KV scan reads the live KV state directly, so it does not require full
- // (snapshot) mode; only the snapshot-based path has that restriction.
- if (bounded && hasPrimaryKey && !fullStartup && !kvBatchEnabled) {
+ if (bounded && hasPrimaryKey && !fullStartup) {
throw new IllegalArgumentException(
String.format(
"Bounded (batch) read on primary-key tables requires full mode "
- + "(OffsetsInitializer.full()) or the server-side KV scan "
- + "('%s' = 'true'), but table '%s' has full startup mode=%s.",
- ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED.key(),
- tablePath,
- fullStartup));
+ + "(OffsetsInitializer.full()), but table '%s' has full "
+ + "startup mode=%s.",
+ tablePath, fullStartup));
}
LakeSource<LakeSplit> lakeSource = null;
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index 7ce0b7e14..02d08dfd7 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -28,6 +28,7 @@ import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer;
import org.apache.fluss.client.metadata.KvSnapshots;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.KvBatchStrategy;
import org.apache.fluss.exception.UnsupportedVersionException;
import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.lake.LakeSplitGenerator;
@@ -576,12 +577,10 @@ public class FlinkSourceEnumerator
}
private void startInBatchMode() {
- boolean kvBatchEnabled = flussConf.get(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED);
- // The server-side KV scan reads the live KV state directly, so it does not require a
- // snapshot-based full startup mode; only the snapshot-based path has that restriction.
- if (hasPrimaryKey
- && !kvBatchEnabled
- && !(startingOffsetsInitializer instanceof SnapshotOffsetsInitializer)) {
+ boolean serverScan =
+ flussConf.get(ConfigOptions.CLIENT_SCANNER_KV_BATCH_STRATEGY)
+ == KvBatchStrategy.SERVER_SCAN;
+ if (hasPrimaryKey && !(startingOffsetsInitializer instanceof SnapshotOffsetsInitializer)) {
throw new UnsupportedOperationException(
"Batch mode on primary-key tables only supports full startup mode.");
}
@@ -600,16 +599,14 @@ public class FlinkSourceEnumerator
tablePath);
splits =
generateFlussOnlyBatchSplits(
- kvBatchEnabled, flussOnlyBatchSplitGenerator);
+ serverScan, flussOnlyBatchSplitGenerator);
}
return splits;
},
this::handleSplitsAdd);
} else {
context.callAsync(
- () ->
- generateFlussOnlyBatchSplits(
- kvBatchEnabled, flussOnlyBatchSplitGenerator),
+ () -> generateFlussOnlyBatchSplits(serverScan, flussOnlyBatchSplitGenerator),
this::handleSplitsAdd);
}
}
@@ -628,13 +625,13 @@ public class FlinkSourceEnumerator
}
/**
- * Generates the Fluss-only batch splits. When the server-side KV scan is enabled for a
- * primary-key table, emits {@link KvBatchSplit}s that scan the live KV state directly;
- * otherwise delegates to the snapshot-based {@link FlussOnlyBatchSplitGenerator}.
+ * Generates the Fluss-only batch splits. Under {@link KvBatchStrategy#SERVER_SCAN} a
+ * primary-key table emits {@link KvBatchSplit}s that scan the live kv state on the server;
+ * otherwise it delegates to the snapshot-based {@link FlussOnlyBatchSplitGenerator}.
*/
private List<SourceSplitBase> generateFlussOnlyBatchSplits(
- boolean kvBatchEnabled, FlussOnlyBatchSplitGenerator flussOnlyBatchSplitGenerator) {
- if (kvBatchEnabled && hasPrimaryKey) {
+ boolean serverScan, FlussOnlyBatchSplitGenerator flussOnlyBatchSplitGenerator) {
+ if (serverScan && hasPrimaryKey) {
if (isPartitioned) {
Set<PartitionInfo> partitionInfos = listPartitions();
List<SourceSplitBase> splits = new ArrayList<>();
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/KvBatchSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/KvBatchSplit.java
index 0fd73236a..4f3b8f154 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/KvBatchSplit.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/KvBatchSplit.java
@@ -22,15 +22,16 @@ import org.apache.fluss.metadata.TableBucket;
import javax.annotation.Nullable;
/**
- * A bounded split that reads the full primary-key state of a bucket via the server-side KV scan.
+ * A bounded split that reads the full primary-key state of a bucket via the server-side kv scan.
* Emitted by the enumerator for every bucket of a primary-key table when the source is running in
- * bounded mode and the server-side KV scan is enabled (see {@code
- * client.scanner.kv.server-side.enabled}).
+ * bounded mode under {@code client.scanner.kv.batch-strategy = server-scan}, and the table has no
+ * lake snapshot to union-read from.
*
* <p>This split has no resumable position: on Flink task restart the bucket is rescanned from
- * scratch. Snapshot isolation is provided by the server (a consistent point-in-time view of the
- * RocksDB state at the moment {@code ScanKv} opens the scan), but the client cannot resume an
- * expired or invalidated session, so progress is not checkpointed.
+ * scratch, so rows already emitted are emitted again. The server pins one RocksDB snapshot per
+ * scanner session, which makes a bucket internally consistent as of the moment {@code ScanKv} opens
+ * it — buckets are opened as readers reach them, so a full scan is not consistent across buckets.
+ * The client cannot resume an expired or invalidated session, so progress is not checkpointed.
*
* <p>Unlike {@link HybridSnapshotLogSplit}, this split has no log handoff phase: when the bucket is
* drained the split is marked finished.
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 2af9913a1..0e48a01bc 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
@@ -724,7 +724,7 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " primary key (id) NOT ENFORCED)"
+ " with ("
+ " 'bucket.num' = '4',"
- + " 'client.scanner.kv.server-side.enabled' = 'true')",
+ + " 'client.scanner.kv.batch-strategy' = 'server-scan')",
tableName));
TablePath tablePath = TablePath.of(databaseName, tableName);
try (Table table = conn.getTable(tablePath)) {
@@ -759,7 +759,7 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " primary key (id) NOT ENFORCED)"
+ " with ("
+ " 'bucket.num' = '4',"
- + " 'client.scanner.kv.server-side.enabled' = 'true')",
+ + " 'client.scanner.kv.batch-strategy' = 'server-scan')",
tableName));
TablePath tablePath = TablePath.of(databaseName, tableName);
try (Table table = conn.getTable(tablePath)) {
@@ -784,6 +784,46 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
assertResultsIgnoreOrder(collected, expected, true);
}
+ @Test
+ void testKvBatchStrategiesReturnSameState() throws Exception {
+ String tableName = String.format("test_kv_batch_strategies_%s", RandomUtils.nextInt());
+ tEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " id int not null,"
+ + " name varchar,"
+ + " primary key (id) NOT ENFORCED)"
+ + " with ('bucket.num' = '3')",
+ tableName));
+ TablePath tablePath = TablePath.of(databaseName, tableName);
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter upsertWriter = table.newUpsert().createWriter();
+ for (int i = 1; i <= 5; i++) {
+ upsertWriter.upsert(row(i, "name" + i));
+ }
+ upsertWriter.upsert(row(1, "name1-updated"));
+ upsertWriter.delete(row(2, "name2"));
+ upsertWriter.flush();
+ }
+
+ List<String> snapshotMerge =
+ collectBatchRows(
+ tEnv.executeSql(String.format("SELECT * FROM %s", tableName)).collect());
+ List<String> serverScan =
+ collectBatchRows(
+ tEnv.executeSql(
+ String.format(
+ "SELECT * FROM %s /*+ OPTIONS("
+ + "'client.scanner.kv.batch-strategy' = 'server-scan') */",
+ tableName))
+ .collect());
+
+ assertThat(snapshotMerge)
+ .containsExactlyInAnyOrder(
+ "+I[1, name1-updated]", "+I[3, name3]", "+I[4, name4]", "+I[5, name5]");
+ assertThat(serverScan).containsExactlyInAnyOrderElementsOf(snapshotMerge);
+ }
+
@Test
void testKvBatchScanReturnsAllRecords() throws Exception {
String tableName = String.format("test_kv_batch_100_%s", RandomUtils.nextInt());
@@ -795,7 +835,7 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " primary key (id) NOT ENFORCED)"
+ " with ("
+ " 'bucket.num' = '3',"
- + " 'client.scanner.kv.server-side.enabled' = 'true')",
+ + " 'client.scanner.kv.batch-strategy' = 'server-scan')",
tableName));
TablePath tablePath = TablePath.of(databaseName, tableName);
try (Table table = conn.getTable(tablePath)) {
@@ -829,7 +869,7 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " primary key (id) NOT ENFORCED)"
+ " with ("
+ " 'bucket.num' = '3',"
- + " 'client.scanner.kv.server-side.enabled' = 'true')",
+ + " 'client.scanner.kv.batch-strategy' = 'server-scan')",
tableName));
TablePath tablePath = TablePath.of(databaseName, tableName);
try (Table table = conn.getTable(tablePath)) {
@@ -858,7 +898,7 @@ abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
+ " primary key (id) NOT ENFORCED)"
+ " with ("
+ " 'bucket.num' = '3',"
- + " 'client.scanner.kv.server-side.enabled' = 'true')",
+ + " 'client.scanner.kv.batch-strategy' = 'server-scan')",
tableName));
// No rows written — scan must complete naturally with an empty result set.
CloseableIterator<Row> collected =
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceITCase.java
index 283e8bad5..75630b66e 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceITCase.java
@@ -23,6 +23,7 @@ import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.KvBatchStrategy;
import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
import org.apache.fluss.flink.source.testutils.MockDataUtils;
import org.apache.fluss.flink.source.testutils.Order;
@@ -121,7 +122,7 @@ public class FlussSourceITCase extends FlinkTestBase {
writeRowsToTable(ordersPKTablePath);
Configuration flussConf = new Configuration();
- flussConf.set(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED, true);
+ flussConf.set(ConfigOptions.CLIENT_SCANNER_KV_BATCH_STRATEGY, KvBatchStrategy.SERVER_SCAN);
FlussSource<Order> flussSource =
FlussSource.<Order>builder()
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
index 0a85fa74c..b70a4d88e 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
@@ -23,6 +23,7 @@ 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.config.Configuration;
+import org.apache.fluss.config.KvBatchStrategy;
import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
import org.apache.fluss.flink.lake.split.LakeSnapshotSplit;
@@ -158,7 +159,7 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
long tableId = createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR);
int numSubtasks = DEFAULT_BUCKET_NUM;
Configuration enabled = new Configuration(flussConf);
- enabled.set(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED, true);
+ enabled.set(ConfigOptions.CLIENT_SCANNER_KV_BATCH_STRATEGY, KvBatchStrategy.SERVER_SCAN);
try (MockSplitEnumeratorContext<SourceSplitBase> context =
new MockSplitEnumeratorContext<>(numSubtasks)) {
FlinkSourceEnumerator enumerator =
@@ -200,10 +201,47 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
}
}
+ @Test
+ void testBoundedPkTableEmitsSnapshotSplitsByDefault() throws Throwable {
+ createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR);
+ int numSubtasks = DEFAULT_BUCKET_NUM;
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(numSubtasks)) {
+ FlinkSourceEnumerator enumerator =
+ new FlinkSourceEnumerator(
+ DEFAULT_TABLE_PATH,
+ flussConf,
+ true,
+ false,
+ context,
+ OffsetsInitializer.full(),
+ DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS,
+ false, // bounded
+ null,
+ null,
+ LeaseContext.DEFAULT,
+ false);
+
+ enumerator.start();
+ for (int i = 0; i < numSubtasks; i++) {
+ registerReader(context, enumerator, i);
+ }
+ context.runNextOneTimeCallable();
+
+ List<SourceSplitBase> allAssigned =
+ context.getSplitsAssignmentSequence().stream()
+ .flatMap(a -> a.assignment().values().stream())
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ assertThat(allAssigned).isNotEmpty();
+ allAssigned.forEach(s -> assertThat(s.isHybridSnapshotLogSplit()).isTrue());
+ }
+ }
+
/**
- * With {@code kvBatchEnabled=true} and a partitioned primary-key table the enumerator must emit
- * one {@link KvBatchSplit} per (partition, bucket) pair — one for each auto-created partition
- * and each bucket within it.
+ * Under {@code server-scan} and a partitioned primary-key table the enumerator must emit one
+ * {@link KvBatchSplit} per (partition, bucket) pair — one for each auto-created partition and
+ * each bucket within it.
*/
@Test
void testBoundedPartitionedPkTableEmitsKvBatchSplits() throws Throwable {
@@ -215,7 +253,7 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
int numSubtasks = DEFAULT_BUCKET_NUM;
Configuration enabled = new Configuration(flussConf);
- enabled.set(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED, true);
+ enabled.set(ConfigOptions.CLIENT_SCANNER_KV_BATCH_STRATEGY, KvBatchStrategy.SERVER_SCAN);
try (MockSplitEnumeratorContext<SourceSplitBase> context =
new MockSplitEnumeratorContext<>(numSubtasks)) {
FlinkSourceEnumerator enumerator =
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index 9e86c955f..9eef49967 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -106,7 +106,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties)
| scan.partition.discovery.interval | Duration | 1min | The time interval for the Fluss source to discover the new partitions for partitioned table while scanning. A non-positive value disables the partition discovery. The default value is 1 minute. Currently, since Fluss Admin#listPartitions(TablePath tablePath) requires a large number of requests to ZooKeeper in server, this option cannot be set too small, as a small value would [...]
| scan.kv.snapshot.lease.id | String | UUID | The lease ID used to protect acquired KV snapshots from deletion. If specified, the snapshots will be retained until either the consumer finishes processing all of them or the lease duration expires. By default, this value is set to a randomly generated UUID string if not explicitly provided. [...]
| scan.kv.snapshot.lease.duration | Duration | 1day | The time period how long to wait before expiring the kv snapshot lease to avoid kv snapshot blocking to delete. [...]
-| client.scanner.kv.server-side.enabled | Boolean | false | Master switch for using the server-side KV scan in bounded reads of primary-key tables. When true, bounded primary-key reads scan the current KV state directly via the server-side KV scan. When false (default), bounded primary-key reads fall back to the prior behavior (log-only when lake is enabled, or fail when lake is disabled). This option has no effect when a lake snapshot [...]
+| client.scanner.kv.batch-strategy | Enum | snapshot-merge | The strategy used for bounded reads of primary-key tables. `snapshot-merge` (default) merges the latest kv snapshot with the bounded changelog range that follows it; the scan is resumable and reflects a single point in time. `server-scan` scans the live kv state on the tablet server instead, which avoids downloading snapshot files and replaying the changelog, but the scan is n [...]
| client.scanner.log.check-crc | Boolean | true | Automatically check the CRC3 of the read records for LogScanner. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. [...]
| client.scanner.log.max-poll-records | Integer | 500 | The maximum number of records returned in a single call to poll() for LogScanner. Note that this config doesn't impact the underlying fetching behavior. The Scanner will cache the records from each fetch request and returns them incrementally from each poll. [...]
| client.scanner.log.fetch.max-bytes | MemorySize | 16mb | The maximum amount of data the server should return for a fetch request from client. Records are fetched in batches, and if the first record batch in the first non-empty bucket of the fetch is larger than this value, the record batch will still be returned to ensure that the fetch can make progress. As such, this is not a absolute maximum. [...]
diff --git a/website/docs/engine-flink/reads.md b/website/docs/engine-flink/reads.md
index e99eca1ab..1c58cd14b 100644
--- a/website/docs/engine-flink/reads.md
+++ b/website/docs/engine-flink/reads.md
@@ -259,19 +259,22 @@ LIMIT 100;
`$changelog` is available for Primary Key Tables and Log Tables, while `$binlog` is available only for Primary Key Tables. See [Virtual Tables](/table-design/virtual-tables.md#flink-runtime-modes) for schemas, startup modes, and detailed bounded-read semantics.
-### Full Scan of Primary Key Tables
+### Server-Side Scan of Primary Key Tables
-Fluss can perform a bounded full-table scan on a primary-key table directly via the server-side KV scan API.
+A bounded read of a primary-key table merges the latest kv snapshot with the changelog range that
+follows it. Setting `client.scanner.kv.batch-strategy` to `server-scan` reads the live kv state on
+the tablet server instead, which avoids downloading snapshot files and replaying the changelog.
-Enable the feature by setting `client.scanner.kv.server-side.enabled = true` on the table or as a SQL hint:
+This is intended for interactive queries over small primary-key tables — the kind of full scan that
+backs a dashboard of a few thousand rows. It is not a general replacement for the default strategy.
-- This is a **bounded** read. The source finishes once all buckets have been drained and does not continue reading the change-log.
-- On task restart, each bucket is rescanned from scratch. Progress within a scan session is not checkpointed, because an expired or invalidated server-side session cannot be resumed from a mid-point.
-- The feature is disabled by default (`false`). Without it, unbounded (streaming) reads on primary-key tables work as usual, and bounded reads fall back to the snapshot-based batch scan (full startup mode).
+The option applies only to primary-key tables that have no lake snapshot to read from. On a
+lake-enabled primary-key table with a lake snapshot, a bounded read performs the lake + Fluss-log
+union read and this option is not consulted.
#### Example
-**1. Create a primary-key table with the feature enabled:**
+**1. Create a primary-key table:**
```sql title="Flink SQL"
CREATE TABLE pk_table (
id INT NOT NULL,
@@ -279,8 +282,7 @@ CREATE TABLE pk_table (
region STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
- 'bucket.num' = '4',
- 'client.scanner.kv.server-side.enabled' = 'true'
+ 'bucket.num' = '4'
);
```
@@ -295,14 +297,31 @@ INSERT INTO pk_table VALUES
**3. Run a full scan in batch mode:**
```sql title="Flink SQL"
SET 'execution.runtime-mode' = 'batch';
-SELECT * FROM pk_table;
+SELECT * FROM pk_table /*+ OPTIONS('client.scanner.kv.batch-strategy' = 'server-scan') */;
```
-You can also enable the feature dynamically without storing it in the table metadata:
-```sql title="Flink SQL"
-SET 'execution.runtime-mode' = 'batch';
-SELECT * FROM pk_table /*+ OPTIONS('client.scanner.kv.server-side.enabled' = 'true') */;
-```
+The option can also be set in `CREATE TABLE`, but prefer the hint: as a table property it applies to
+every bounded reader of that table, including jobs written later by other users.
+
+#### Trade-offs
+
+Both strategies return the current state of the table. They differ in guarantees and cost:
+
+| | `snapshot-merge` (default) | `server-scan` |
+|---|---|---|
+| Resumable | Yes — the scan resumes from its checkpointed position | No — the bucket is re-read from the start, so rows already emitted are emitted again |
+| Point in time | Fixed when splits are planned: every bucket is cut at its latest offset at that moment | Per bucket, fixed when that bucket's scanner opens — buckets open as readers reach them |
+| Client cost | Downloads snapshot files and replays the changelog | None |
+| Server cost | None beyond serving the changelog | Holds a scan session on the tablet server serving the bucket |
+
+Because `server-scan` is not resumable, a bounded source that is checkpointed (a `setBounded()`
+DataStream source in streaming runtime mode) will emit duplicates after a restore. Batch runtime
+mode does not checkpoint and is unaffected.
+
+Each `server-scan` session lives on the tablet server that leads the bucket and is subject to
+`kv.scanner.ttl` (10 minutes of idleness) and `kv.scanner.max-per-bucket` (8 concurrent sessions per
+bucket). A source that is back-pressured for longer than the TTL fails its split; a bucket read
+concurrently by more queries than the limit fails to open.
### Limit Read
The Fluss source supports limiting reads for both primary-key tables and log tables, making it convenient to preview the latest `N` records in a table.