This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 3fa9c571 [FLINK-26936] Pushdown watermark to table store source
3fa9c571 is described below
commit 3fa9c571b8d7f5b38aa6bee19d821eca0e40c84a
Author: Nicholas Jiang <[email protected]>
AuthorDate: Tue Jul 5 21:23:45 2022 +0800
[FLINK-26936] Pushdown watermark to table store source
This closes #198
---
.../store/connector/source/FlinkSourceBuilder.java | 11 ++++-
.../store/connector/source/TableStoreSource.java | 48 ++++++++++++++++++----
2 files changed, 50 insertions(+), 9 deletions(-)
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FlinkSourceBuilder.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FlinkSourceBuilder.java
index 2bdaf7c3..38c5c8ce 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FlinkSourceBuilder.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FlinkSourceBuilder.java
@@ -60,6 +60,7 @@ public class FlinkSourceBuilder {
@Nullable private Predicate predicate;
@Nullable private LogSourceProvider logSourceProvider;
@Nullable private Integer parallelism;
+ @Nullable private WatermarkStrategy<RowData> watermarkStrategy;
public FlinkSourceBuilder(ObjectIdentifier tableIdentifier, FileStoreTable
table) {
this.tableIdentifier = tableIdentifier;
@@ -97,6 +98,12 @@ public class FlinkSourceBuilder {
return this;
}
+ public FlinkSourceBuilder withWatermarkStrategy(
+ @Nullable WatermarkStrategy<RowData> watermarkStrategy) {
+ this.watermarkStrategy = watermarkStrategy;
+ return this;
+ }
+
private long discoveryIntervalMills() {
return conf.get(CONTINUOUS_DISCOVERY_INTERVAL).toMillis();
}
@@ -155,7 +162,9 @@ public class FlinkSourceBuilder {
conf.get(COMPACTION_MANUAL_TRIGGERED)
? new FileStoreEmptySource()
: buildSource(),
- WatermarkStrategy.noWatermarks(),
+ watermarkStrategy == null
+ ? WatermarkStrategy.noWatermarks()
+ : watermarkStrategy,
tableIdentifier.asSummaryString(),
InternalTypeInfo.of(produceType));
if (parallelism != null) {
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
index c5545658..b173f1ba 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.connector.source;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.connector.ChangelogMode;
@@ -25,6 +26,8 @@ import
org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
+import
org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.store.CoreOptions.LogChangelogMode;
@@ -57,7 +60,10 @@ import static
org.apache.flink.table.store.CoreOptions.LOG_CONSISTENCY;
* log source created by {@link LogSourceProvider}.
*/
public class TableStoreSource
- implements ScanTableSource, SupportsFilterPushDown,
SupportsProjectionPushDown {
+ implements ScanTableSource,
+ SupportsFilterPushDown,
+ SupportsProjectionPushDown,
+ SupportsWatermarkPushDown {
private final ObjectIdentifier tableIdentifier;
private final FileStoreTable table;
@@ -68,17 +74,34 @@ public class TableStoreSource
@Nullable private Predicate predicate;
@Nullable private int[][] projectFields;
+ @Nullable private WatermarkStrategy<RowData> watermarkStrategy;
+
public TableStoreSource(
ObjectIdentifier tableIdentifier,
FileStoreTable table,
boolean streaming,
DynamicTableFactory.Context context,
@Nullable LogStoreTableFactory logStoreTableFactory) {
+ this(tableIdentifier, table, streaming, context, logStoreTableFactory,
null, null, null);
+ }
+
+ private TableStoreSource(
+ ObjectIdentifier tableIdentifier,
+ FileStoreTable table,
+ boolean streaming,
+ DynamicTableFactory.Context context,
+ @Nullable LogStoreTableFactory logStoreTableFactory,
+ @Nullable Predicate predicate,
+ @Nullable int[][] projectFields,
+ @Nullable WatermarkStrategy<RowData> watermarkStrategy) {
this.tableIdentifier = tableIdentifier;
this.table = table;
this.streaming = streaming;
this.context = context;
this.logStoreTableFactory = logStoreTableFactory;
+ this.predicate = predicate;
+ this.projectFields = projectFields;
+ this.watermarkStrategy = watermarkStrategy;
}
@Override
@@ -122,7 +145,8 @@ public class TableStoreSource
.withPredicate(predicate)
.withParallelism(
Configuration.fromMap(table.schema().options())
-
.get(FlinkConnectorOptions.SCAN_PARALLELISM));
+
.get(FlinkConnectorOptions.SCAN_PARALLELISM))
+ .withWatermarkStrategy(watermarkStrategy);
return new TableStoreDataStreamScanProvider(
!streaming, env -> sourceBuilder.withEnv(env).build());
@@ -130,12 +154,15 @@ public class TableStoreSource
@Override
public DynamicTableSource copy() {
- TableStoreSource copied =
- new TableStoreSource(
- tableIdentifier, table, streaming, context,
logStoreTableFactory);
- copied.predicate = predicate;
- copied.projectFields = projectFields;
- return copied;
+ return new TableStoreSource(
+ tableIdentifier,
+ table,
+ streaming,
+ context,
+ logStoreTableFactory,
+ predicate,
+ projectFields,
+ watermarkStrategy);
}
@Override
@@ -163,4 +190,9 @@ public class TableStoreSource
public void applyProjection(int[][] projectedFields) {
this.projectFields = projectedFields;
}
+
+ @Override
+ public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
+ this.watermarkStrategy = watermarkStrategy;
+ }
}
