Copilot commented on code in PR #3383:
URL: https://github.com/apache/fluss/pull/3383#discussion_r3308493921
##########
website/docs/engine-flink/options.md:
##########
@@ -102,6 +102,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 cause frequent requests and increase
server load. In the future, once list partitions is optimized, the default
value of this parameter can be reduced. |
| 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 (FIP-17) in bounded reads of primary-key tables when no KV snapshot file
is available. 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).
See [Full Scan of Primary Key
Tables](engine-flink/reads.md#full-scan-of-primary-key-tables-server-side-kv-scan)
for details. |
Review Comment:
The anchor in this link looks incorrect for the heading added in reads.md
("### Full Scan of Primary Key Tables"), which will likely generate
`#full-scan-of-primary-key-tables` rather than
`#full-scan-of-primary-key-tables-server-side-kv-scan`. This will produce a
broken docs link.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -477,39 +479,65 @@ private void startInBatchMode() {
context.callAsync(
() -> {
List<SourceSplitBase> splits =
generateHybridLakeFlussSplits();
- // No lake snapshot exists, fall back to Fluss-only
splits
if (splits == null) {
LOG.info(
"No lake snapshot found for table {},"
+ " falling back to Fluss-only
splits.",
tablePath);
- if (isPartitioned) {
- Set<PartitionInfo> partitionInfos =
listPartitions();
- Collection<Partition> partitions =
- partitionInfos.stream()
- .map(
- p ->
- new Partition(
-
p.getPartitionId(),
-
p.getPartitionName()))
- .collect(Collectors.toList());
- // Use log-only splits to avoid generating
mixed split
- // types (HybridSnapshotLogSplit + LogSplit)
for
- // primary-key tables, which is not supported.
- splits =
this.initLogTablePartitionSplits(partitions);
- } else {
- splits = this.getLogSplit(null, null);
- }
+ splits =
generateFlussOnlyBatchSplits(kvBatchEnabled);
}
return splits;
},
this::handleSplitsAdd);
+ } else if (kvBatchEnabled && hasPrimaryKey) {
+ context.callAsync(() -> generateFlussOnlyBatchSplits(true),
this::handleSplitsAdd);
} else {
throw new UnsupportedOperationException(
String.format(
- "Batch only supports when table option '%s' is set
to true.",
- ConfigOptions.TABLE_DATALAKE_ENABLED));
+ "Batch mode requires either '%s' = 'true'
(data-lake integration) "
+ + "or '%s' = 'true' (server-side KV scan,
primary-key tables only).",
+ ConfigOptions.TABLE_DATALAKE_ENABLED,
+
ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED));
Review Comment:
The exception message uses `String.format(...,
ConfigOptions.TABLE_DATALAKE_ENABLED, ...)` and
`ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED` directly, which will
format as the `ConfigOption` object's `toString()` rather than the option key.
This makes the user-facing error message noisy/inconsistent with other messages
in this PR that use `.key()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
