This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a6f1f40dcb0 [FLINK-32695][e2e] Migrate SqlToolbox watermark push-down
source to Source V2
a6f1f40dcb0 is described below
commit a6f1f40dcb09258afa93fba274a4470ea342b790
Author: Martijn Visser <[email protected]>
AuthorDate: Thu Jul 2 14:03:42 2026 +0200
[FLINK-32695][e2e] Migrate SqlToolbox watermark push-down source to Source
V2
TestSourceFunction implemented the legacy SourceFunction and manually
drove the pushed WatermarkStrategy's generator;
TestScanTableSourceWithWatermarkPushDown wrapped it in the deprecated
SourceFunctionProvider.
Replace both with a DataGeneratorSource behind a DataStreamScanProvider
that applies the pushed watermark strategy via fromSource, the same
pattern KafkaDynamicSource uses. A plain SourceProvider is not suitable
because the planner does not apply the pushed strategy on that path, so
the generated watermark code would never be instantiated on the
TaskManager, which is what this source exists to test (FLINK-28693).
Generated-by: Claude Code (Sonnet 4.6)
---
.../flink/table/sql/PlannerScalaFreeITCase.java | 5 +-
.../flink-sql-client-test/pom.xml | 7 ++
.../TestScanTableSourceWithWatermarkPushDown.java | 62 +++++++++--
...canTableSourceWithWatermarkPushDownFactory.java | 2 +-
.../flink/table/toolbox/TestSourceFunction.java | 123 ---------------------
5 files changed, 66 insertions(+), 133 deletions(-)
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/PlannerScalaFreeITCase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/PlannerScalaFreeITCase.java
index 51a0f278cb3..79e28a68c7c 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/PlannerScalaFreeITCase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/PlannerScalaFreeITCase.java
@@ -66,7 +66,10 @@ class PlannerScalaFreeITCase extends SqlITCaseBase {
runAndCheckSQL("scala_free_e2e.sql", Arrays.asList("+I[Bob, 2]",
"+I[Alice, 1]"));
}
- /** The test data is from {@link
org.apache.flink.table.toolbox.TestSourceFunction#DATA}. */
+ /**
+ * The test data is from {@link
+ *
org.apache.flink.table.toolbox.TestScanTableSourceWithWatermarkPushDown#DATA}.
+ */
@TestTemplate
void testWatermarkPushDown() throws Exception {
assumeThat(executionMode).isEqualTo("streaming");
diff --git a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
index 78df21239ee..aac3141ca9a 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
+++ b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
@@ -45,6 +45,13 @@ under the License.
<scope>provided</scope>
</dependency>
+ <dependency>
+ <!-- Bundled into SqlToolbox.jar; not part of
flink-dist -->
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-datagen</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-end-to-end-tests-common</artifactId>
diff --git
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDown.java
b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDown.java
index db51552451b..7152d1eb9c9 100644
---
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDown.java
+++
b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDown.java
@@ -19,23 +19,45 @@
package org.apache.flink.table.toolbox;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.legacy.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import
org.apache.flink.connector.datagen.functions.FromElementsGeneratorFunction;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.source.DataStreamScanProvider;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import
org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
-/**
- * A source used to test {@link SupportsWatermarkPushDown}.
- *
- * <p>For simplicity, the deprecated source function method is used to create
the source.
- */
+import java.util.Arrays;
+import java.util.List;
+
+/** A source used to test {@link SupportsWatermarkPushDown}. */
public class TestScanTableSourceWithWatermarkPushDown
implements ScanTableSource, SupportsWatermarkPushDown {
+ public static final List<RowData> DATA =
+ Arrays.asList(
+ GenericRowData.of(
+ StringData.fromString("Bob"), 1L,
TimestampData.fromEpochMillis(1)),
+ GenericRowData.of(
+ StringData.fromString("Alice"), 2L,
TimestampData.fromEpochMillis(2)));
+
+ private final DataType producedDataType;
+
private WatermarkStrategy<RowData> watermarkStrategy;
+ public TestScanTableSourceWithWatermarkPushDown(DataType producedDataType)
{
+ this.producedDataType = producedDataType;
+ }
+
@Override
public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
@@ -44,7 +66,7 @@ public class TestScanTableSourceWithWatermarkPushDown
@Override
public DynamicTableSource copy() {
final TestScanTableSourceWithWatermarkPushDown newSource =
- new TestScanTableSourceWithWatermarkPushDown();
+ new TestScanTableSourceWithWatermarkPushDown(producedDataType);
newSource.watermarkStrategy = this.watermarkStrategy;
return newSource;
}
@@ -61,6 +83,30 @@ public class TestScanTableSourceWithWatermarkPushDown
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext
runtimeProviderContext) {
- return SourceFunctionProvider.of(new
TestSourceFunction(watermarkStrategy), false);
+ final TypeInformation<RowData> typeInfo =
+ runtimeProviderContext.createTypeInformation(producedDataType);
+ final DataGeneratorSource<RowData> source =
+ new DataGeneratorSource<>(
+ new FromElementsGeneratorFunction<>(typeInfo,
DATA.toArray(new RowData[0])),
+ DATA.size(),
+ typeInfo);
+ // the pushed watermark strategy must be applied by the connector
itself (like
+ // KafkaDynamicSource); the planner does not apply it for
SourceProvider sources
+ return new DataStreamScanProvider() {
+ @Override
+ public DataStream<RowData> produceDataStream(
+ ProviderContext providerContext,
StreamExecutionEnvironment execEnv) {
+ final WatermarkStrategy<RowData> strategy =
+ watermarkStrategy != null
+ ? watermarkStrategy
+ : WatermarkStrategy.noWatermarks();
+ return execEnv.fromSource(source, strategy,
"watermark-push-down-source", typeInfo);
+ }
+
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+ };
}
}
diff --git
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDownFactory.java
b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDownFactory.java
index 22452706c23..b3f4f847668 100644
---
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDownFactory.java
+++
b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestScanTableSourceWithWatermarkPushDownFactory.java
@@ -32,7 +32,7 @@ public class TestScanTableSourceWithWatermarkPushDownFactory
implements DynamicT
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- return new TestScanTableSourceWithWatermarkPushDown();
+ return new
TestScanTableSourceWithWatermarkPushDown(context.getPhysicalRowDataType());
}
@Override
diff --git
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
deleted file mode 100644
index 5d1d499ec49..00000000000
---
a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * 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.flink.table.toolbox;
-
-import org.apache.flink.api.common.eventtime.Watermark;
-import org.apache.flink.api.common.eventtime.WatermarkGenerator;
-import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
-import org.apache.flink.api.common.eventtime.WatermarkOutput;
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
-import org.apache.flink.table.data.GenericRowData;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.util.clock.RelativeClock;
-import org.apache.flink.util.clock.SystemClock;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- * A source function used to test.
- *
- * <p>For simplicity, the deprecated source function method is used to create
the source.
- */
-@SuppressWarnings("deprecation")
-public class TestSourceFunction implements SourceFunction<RowData> {
-
- public static final List<List<Object>> DATA = new ArrayList<>();
-
- static {
- DATA.add(Arrays.asList(StringData.fromString("Bob"), 1L,
TimestampData.fromEpochMillis(1)));
- DATA.add(
- Arrays.asList(
- StringData.fromString("Alice"), 2L,
TimestampData.fromEpochMillis(2)));
- }
-
- private final WatermarkStrategy<RowData> watermarkStrategy;
-
- public TestSourceFunction(WatermarkStrategy<RowData> watermarkStrategy) {
- this.watermarkStrategy = watermarkStrategy;
- }
-
- private volatile boolean isRunning = true;
-
- @Override
- public void run(SourceContext<RowData> ctx) {
- WatermarkGenerator<RowData> generator =
- watermarkStrategy.createWatermarkGenerator(
- new WatermarkGeneratorSupplier.Context() {
- @Override
- public MetricGroup getMetricGroup() {
- return null;
- }
-
- @Override
- public RelativeClock getInputActivityClock() {
- return SystemClock.getInstance();
- }
- });
- WatermarkOutput output = new TestWatermarkOutput(ctx);
-
- int rowDataSize = DATA.get(0).size();
- int index = 0;
- while (index < DATA.size() && isRunning) {
- List<Object> list = DATA.get(index);
- GenericRowData row = new GenericRowData(rowDataSize);
- for (int i = 0; i < rowDataSize; i++) {
- row.setField(i, list.get(i));
- }
- ctx.collect(row);
- index++;
- generator.onEvent(row, Long.MIN_VALUE, output);
- generator.onPeriodicEmit(output);
- }
- ctx.close();
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
- private static class TestWatermarkOutput implements WatermarkOutput {
-
- private final SourceFunction.SourceContext<RowData> ctx;
-
- public TestWatermarkOutput(SourceFunction.SourceContext<RowData> ctx) {
- this.ctx = ctx;
- }
-
- @Override
- public void emitWatermark(Watermark watermark) {
- ctx.emitWatermark(
- new org.apache.flink.streaming.api.watermark.Watermark(
- watermark.getTimestamp()));
- }
-
- @Override
- public void markIdle() {}
-
- @Override
- public void markActive() {}
- }
-}
