binary-signal commented on PR #3383:
URL: https://github.com/apache/fluss/pull/3383#issuecomment-4742907159
@polyzos I had a look at the PR and can confirm that bounded reads via the
Table API work as expected.
That said, there is an issue when performing bounded reads via the
DataStream API. It looks like the corresponding logic was missed in
FlussSourceBuilder.java. The required changes were already implemented in
[flink] Add union read support to datastream (#3432) which allows
`setBounded()` to be set in a Fluss Flink Source. The following guard throws:
```java
# FlussSourceBuilder.java
if (bounded && !(lakeEnabled && fullStartup)) {
...
}
```
The following patch addresses the issue and also adds a test case to cover
this scenario.
```
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 2e80067f..eb95e731 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,7 +176,10 @@ 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. If not called, the source is unbounded (streaming).
+ * 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).
*
* @return this builder
*/
@@ -353,13 +356,24 @@ public class FlussSourceBuilder<OUT> {
boolean lakeEnabled =
tableInfo.getTableConfig().isDataLakeEnabled();
boolean fullStartup = offsetsInitializer instanceof
SnapshotOffsetsInitializer;
- if (bounded && !(lakeEnabled && fullStartup)) {
+ // bounded read via the server-side KV scan applies to primary-key
tables when the master
+ // switch is enabled, mirroring the SQL connector (see
FlinkTableSource).
+ boolean kvBatchEnabled =
flussConf.get(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED);
+ boolean kvBatchAllowed = hasPrimaryKey && kvBatchEnabled;
+
+ if (bounded && !(lakeEnabled && fullStartup) && !kvBatchAllowed) {
throw new IllegalArgumentException(
String.format(
- "Bounded (batch) read requires a
datalake-enabled table started in "
- + "full mode
(OffsetsInitializer.full()), but table '%s' has "
- + "datalake enabled=%s and full startup
mode=%s.",
- tablePath, lakeEnabled, fullStartup));
+ "Bounded (batch) read requires either a
datalake-enabled table started "
+ + "in full mode
(OffsetsInitializer.full()), or a primary-key "
+ + "table with '%s' = 'true'. Table '%s'
has datalake enabled=%s, "
+ + "full startup mode=%s, primary
key=%s, server-side KV scan=%s.",
+
ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED.key(),
+ tablePath,
+ lakeEnabled,
+ fullStartup,
+ hasPrimaryKey,
+ kvBatchEnabled));
}
LakeSource<LakeSplit> lakeSource = null;
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 f4b112c4..df211643 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
@@ -21,6 +21,8 @@ import
org.apache.fluss.client.initializer.OffsetsInitializer;
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.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
import
org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
import org.apache.fluss.flink.source.testutils.MockDataUtils;
import org.apache.fluss.flink.source.testutils.Order;
@@ -113,6 +115,37 @@ public class FlussSourceITCase extends FlinkTestBase {
assertThat(collectedElements).hasSameElementsAs(ORDERS);
}
+ @Test
+ public void testBoundedKvBatchPKSource() throws Exception {
+ createTable(ordersPKTablePath, pkTableDescriptor);
+ writeRowsToTable(ordersPKTablePath);
+
+ // Enable the server-side KV scan so a bounded read is allowed on a
primary-key table
+ // without the data-lake integration (mirrors the SQL connector's
behavior).
+ Configuration flussConf = new Configuration();
+ flussConf.set(ConfigOptions.CLIENT_SCANNER_KV_SERVER_SIDE_ENABLED,
true);
+
+ FlussSource<Order> flussSource =
+ FlussSource.<Order>builder()
+ .setBootstrapServers(bootstrapServers)
+ .setDatabase(DEFAULT_DB)
+ .setTable(pkTableName)
+ .setStartingOffsets(OffsetsInitializer.full())
+ .setScanPartitionDiscoveryIntervalMs(1000L)
+ .setDeserializationSchema(new
MockDataUtils.OrderDeserializationSchema())
+ .setFlussConfig(flussConf)
+ .setBounded()
+ .build();
+
+ // env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ DataStreamSource<Order> stream =
+ env.fromSource(flussSource,
WatermarkStrategy.noWatermarks(), "Fluss Source");
+
+ List<Order> collectedElements =
stream.executeAndCollect(ORDERS.size());
+
+ assertThat(collectedElements).hasSameElementsAs(ORDERS);
+ }
+
@Test
public void testTablePKSourceWithProjectionPushdown() throws Exception {
createTable(ordersPKTablePath, pkTableDescriptor);
```
Quick example
```java
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");
}
}
```
--
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]
