[Flink] Kvscan flink integration#3383
Open
polyzos wants to merge 16 commits into
Open
Conversation
2 tasks
polyzos
force-pushed
the
kvscan-flink-integration
branch
from
May 26, 2026 12:21
ddb782e to
ecdb181
Compare
|
|
||
| - 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; bounded reads require the data-lake integration to be enabled. |
Contributor
There was a problem hiding this comment.
When disabled, unbounded (streaming) reads on primary-key tables continue to work as usual. Bounded reads require data-lake integration unless server-side KV scanning is enabled.
| ``` | ||
|
|
||
| ### 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. |
Contributor
There was a problem hiding this comment.
making it easy to preview the latest N records in a table.
Contributor
Author
|
@binary-signa Thank you for your comments... Addressed! |
Contributor
|
@polyzos I had a look at the PR and can confirm that bounded reads via the Table API work as expected. # FlussSourceBuilder.java
if (bounded && !(lakeEnabled && fullStartup)) {
...
}The following patch addresses the issue and also adds a test case to cover this scenario. Quick example to reproduce create a primary key table with datalake option disabled CREATE DATABASE cdstream_db;
CREATE TABLE `fluss_catalog`.`cdstream_db`.`timeseries_norm` (
`streamId` INT NOT NULL,
`vector` ARRAY<DOUBLE>,
`firstTimestamp` TIMESTAMP(3),
`lastTimestamp` TIMESTAMP(3),
`lastUpdateTime` TIMESTAMP(3),
CONSTRAINT `PK_streamId` PRIMARY KEY (`streamId`) NOT ENFORCED
)
COMMENT 'This is the normalized timeseries table'
WITH (
'bootstrap.servers' = 'coordinator-server:9123',
'bucket.key' = 'streamId',
'bucket.num' = '2',
'table.datalake.enabled' = 'false',
'table.datalake.format' = 'paimon'
)Then try bounded read via DataStream API it will throw import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.flink.source.FlussSource;
import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
public class FlussDataStreamBoundedReadKvBatch {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
Configuration conf = new Configuration();
conf.setBoolean("client.scanner.kv.server-side.enabled", true);
var source = FlussSource.<RowData>builder()
.setBootstrapServers("localhost:9122")
.setDatabase("cdstream_db")
.setTable("timeseries_norm")
.setDeserializationSchema(new RowDataDeserializationSchema())
.setBounded()
.setFlussConfig(conf)
.build();
DataStream<RowData> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "BoundedFlussKvSource");
stream.print();
env.execute("BoundedFlussKvJob");
}
} |
Contributor
Author
|
@binary-signal yes I need to find some time to rebase, and after we also need to introduce partition pruning and predicate pushdown tot Datastream. I have tickets to track these |
polyzos
force-pushed
the
kvscan-flink-integration
branch
from
July 13, 2026 18:14
962a847 to
9bec12b
Compare
fresh-borzoni
left a comment
fresh-borzoni
left a comment
Member
There was a problem hiding this comment.
@polyzos Thank you for the PR, left a couple of comments, PTAL
| } | ||
|
|
||
| @Test | ||
| void testKvBatchScanOnPkTable() throws Exception { |
Member
There was a problem hiding this comment.
All KV tests write each key exactly once, so a log read would return the same rows and still pass. Shall we add an update + a delete before the scan, so the test can tell KV state from the changelog?
| } else { | ||
| splits = this.getLogSplit(null, null); | ||
| } | ||
| splits = generateFlussOnlyBatchSplits(kvBatchEnabled); |
Member
There was a problem hiding this comment.
With lake enabled but no lake snapshot yet, flag off reads the whole log, so updates and deletes come back as extra rows (-U/+U/-D), and SELECT * returns the history, not the table.
Shall we KV-scan PK tables here regardless of the flag or fail? Silently wrong rows as the default seems wrong. WDYT?
| .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. " |
Member
There was a problem hiding this comment.
"scan the current KV state directly" reads as if the flag forces a KV scan, but when a lake snapshot exists the union read still wins and the flag is never consulted and it only really applies when there's no lake data yet.
Shall we add "has no effect when a lake snapshot already exists" or smth similar? Same- in options.md.
Contributor
Author
|
@fresh-borzoni Thank you for your review and comments. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
#3126 extended the java client to support kvscan for the live rocksdb table.
This PR integrates that functionality inside the flink connector
Introduces
KvBatchSplitandKvBatchSplitStateas a new split type in the Flink source, enabling the connector to perform bounded full-table scans on primary-key tables via the server-side KV scan API (FIP-17), rather than reading from snapshots.