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 d542aeace2c [FLINK-36725] Add CompiledPlan suppport for
BatchExecRuntimeFilter
d542aeace2c is described below
commit d542aeace2c784974a582ad657e4993e27d7fe45
Author: bvarghese1 <[email protected]>
AuthorDate: Tue Oct 27 20:01:00 2026 -0700
[FLINK-36725] Add CompiledPlan suppport for BatchExecRuntimeFilter
Generated-by: Claude Code (Sonnet)
---
.../BatchExecGlobalRuntimeFilterBuilder.java | 30 +-
.../BatchExecLocalRuntimeFilterBuilder.java | 37 ++
.../runtimefilter/BatchExecRuntimeFilter.java | 27 ++
.../planner/plan/utils/ExecNodeMetadataUtil.java | 6 +
.../exec/batch/RuntimeFilterBatchRestoreTest.java | 47 ++
.../exec/batch/RuntimeFilterTestPrograms.java | 81 ++++
.../runtime/batch/sql/RuntimeFilterITCase.java | 55 +++
.../runtime-filter/plan/runtime-filter.json | 492 +++++++++++++++++++++
8 files changed, 773 insertions(+), 2 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecGlobalRuntimeFilterBuilder.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecGlobalRuntimeFilterBuilder.java
index 82d0b2c70ad..1520d3eae27 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecGlobalRuntimeFilterBuilder.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecGlobalRuntimeFilterBuilder.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
@@ -29,6 +30,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
@@ -36,12 +38,23 @@ import
org.apache.flink.table.runtime.operators.runtimefilter.GlobalRuntimeFilte
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.LogicalType;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.List;
/** Batch {@link ExecNode} for global runtime filter builder. */
+@ExecNodeMetadata(
+ name = "batch-exec-global-runtime-filter-builder",
+ version = 1,
+ minPlanVersion = FlinkVersion.v2_5,
+ minStateVersion = FlinkVersion.v2_5)
public class BatchExecGlobalRuntimeFilterBuilder extends ExecNodeBase<RowData>
implements BatchExecNode<RowData> {
+ public static final String FIELD_NAME_MAX_ROW_COUNT = "maxRowCount";
+
+ @JsonProperty(FIELD_NAME_MAX_ROW_COUNT)
private final int maxRowCount;
public BatchExecGlobalRuntimeFilterBuilder(
@@ -52,15 +65,28 @@ public class BatchExecGlobalRuntimeFilterBuilder extends
ExecNodeBase<RowData>
int maxRowCount) {
super(
ExecNodeContext.newNodeId(),
-
ExecNodeContext.newContext(BatchExecLocalRuntimeFilterBuilder.class),
+
ExecNodeContext.newContext(BatchExecGlobalRuntimeFilterBuilder.class),
ExecNodeContext.newPersistedConfig(
- BatchExecLocalRuntimeFilterBuilder.class, tableConfig),
+ BatchExecGlobalRuntimeFilterBuilder.class,
tableConfig),
inputProperties,
outputType,
description);
this.maxRowCount = maxRowCount;
}
+ @JsonCreator
+ public BatchExecGlobalRuntimeFilterBuilder(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig
persistedConfig,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty>
inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) LogicalType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_MAX_ROW_COUNT) int maxRowCount) {
+ super(id, context, persistedConfig, inputProperties, outputType,
description);
+ this.maxRowCount = maxRowCount;
+ }
+
@Override
@SuppressWarnings("unchecked")
protected Transformation<RowData> translateToPlanInternal(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecLocalRuntimeFilterBuilder.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecLocalRuntimeFilterBuilder.java
index fa242100a1c..15656d07081 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecLocalRuntimeFilterBuilder.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecLocalRuntimeFilterBuilder.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
@@ -31,6 +32,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
@@ -41,13 +43,31 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.typeutils.RowTypeUtils;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.List;
/** Batch {@link ExecNode} for local runtime filter builder. */
+@ExecNodeMetadata(
+ name = "batch-exec-local-runtime-filter-builder",
+ version = 1,
+ minPlanVersion = FlinkVersion.v2_5,
+ minStateVersion = FlinkVersion.v2_5)
public class BatchExecLocalRuntimeFilterBuilder extends ExecNodeBase<RowData>
implements BatchExecNode<RowData> {
+
+ public static final String FIELD_NAME_BUILD_INDICES = "buildIndices";
+ public static final String FIELD_NAME_ESTIMATED_ROW_COUNT =
"estimatedRowCount";
+ public static final String FIELD_NAME_MAX_ROW_COUNT = "maxRowCount";
+
+ @JsonProperty(FIELD_NAME_BUILD_INDICES)
private final int[] buildIndices;
+
+ @JsonProperty(FIELD_NAME_ESTIMATED_ROW_COUNT)
private final int estimatedRowCount;
+
+ @JsonProperty(FIELD_NAME_MAX_ROW_COUNT)
private final int maxRowCount;
public BatchExecLocalRuntimeFilterBuilder(
@@ -71,6 +91,23 @@ public class BatchExecLocalRuntimeFilterBuilder extends
ExecNodeBase<RowData>
this.maxRowCount = maxRowCount;
}
+ @JsonCreator
+ public BatchExecLocalRuntimeFilterBuilder(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig
persistedConfig,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty>
inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) LogicalType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_BUILD_INDICES) int[] buildIndices,
+ @JsonProperty(FIELD_NAME_ESTIMATED_ROW_COUNT) int
estimatedRowCount,
+ @JsonProperty(FIELD_NAME_MAX_ROW_COUNT) int maxRowCount) {
+ super(id, context, persistedConfig, inputProperties, outputType,
description);
+ this.buildIndices = buildIndices;
+ this.estimatedRowCount = estimatedRowCount;
+ this.maxRowCount = maxRowCount;
+ }
+
@Override
@SuppressWarnings("unchecked")
protected Transformation<RowData> translateToPlanInternal(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecRuntimeFilter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecRuntimeFilter.java
index 621ecf81b07..f4f28615042 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecRuntimeFilter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/runtimefilter/BatchExecRuntimeFilter.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.data.RowData;
@@ -32,6 +33,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
@@ -40,11 +42,23 @@ import
org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.List;
/** Batch {@link ExecNode} for runtime filter. */
+@ExecNodeMetadata(
+ name = "batch-exec-runtime-filter",
+ version = 1,
+ minPlanVersion = FlinkVersion.v2_5,
+ minStateVersion = FlinkVersion.v2_5)
public class BatchExecRuntimeFilter extends ExecNodeBase<RowData>
implements BatchExecNode<RowData> {
+
+ public static final String FIELD_NAME_PROBE_INDICES = "probeIndices";
+
+ @JsonProperty(FIELD_NAME_PROBE_INDICES)
private final int[] probeIndices;
public BatchExecRuntimeFilter(
@@ -63,6 +77,19 @@ public class BatchExecRuntimeFilter extends
ExecNodeBase<RowData>
this.probeIndices = probeIndices;
}
+ @JsonCreator
+ public BatchExecRuntimeFilter(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig
persistedConfig,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty>
inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) LogicalType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_PROBE_INDICES) int[] probeIndices) {
+ super(id, context, persistedConfig, inputProperties, outputType,
description);
+ this.probeIndices = probeIndices;
+ }
+
@Override
@SuppressWarnings("unchecked")
protected Transformation<RowData> translateToPlanInternal(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
index 19fdd3b1013..79f2f14720e 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
@@ -50,6 +50,9 @@ import
org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecUnion;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecValues;
import
org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecVectorSearchTableFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecWindowTableFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecGlobalRuntimeFilterBuilder;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecLocalRuntimeFilterBuilder;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecRuntimeFilter;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCalc;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCorrelate;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc;
@@ -209,6 +212,9 @@ public final class ExecNodeMetadataUtil {
add(BatchExecRank.class);
add(BatchExecVectorSearchTableFunction.class);
add(BatchExecMLPredictTableFunction.class);
+ add(BatchExecRuntimeFilter.class);
+ add(BatchExecLocalRuntimeFilterBuilder.class);
+ add(BatchExecGlobalRuntimeFilterBuilder.class);
}
};
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterBatchRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterBatchRestoreTest.java
new file mode 100644
index 00000000000..7d9e17cfd0d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterBatchRestoreTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.planner.plan.nodes.exec.batch;
+
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecGlobalRuntimeFilterBuilder;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecLocalRuntimeFilterBuilder;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecRuntimeFilter;
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.BatchRestoreTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/** Restore tests for {@link BatchExecRuntimeFilter}. */
+public class RuntimeFilterBatchRestoreTest extends BatchRestoreTestBase {
+
+ public RuntimeFilterBatchRestoreTest() {
+ super(
+ BatchExecRuntimeFilter.class,
+ Arrays.<Class<? extends ExecNode<?>>>asList(
+ BatchExecLocalRuntimeFilterBuilder.class,
+ BatchExecGlobalRuntimeFilterBuilder.class));
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return
Collections.singletonList(RuntimeFilterTestPrograms.RUNTIME_FILTER);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterTestPrograms.java
new file mode 100644
index 00000000000..73acd2ef163
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/RuntimeFilterTestPrograms.java
@@ -0,0 +1,81 @@
+/*
+ * 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.planner.plan.nodes.exec.batch;
+
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+/**
+ * {@link TableTestProgram} definitions for testing {@link
+ *
org.apache.flink.table.planner.plan.nodes.exec.batch.runtimefilter.BatchExecRuntimeFilter}.
+ */
+class RuntimeFilterTestPrograms {
+
+ static final TableTestProgram RUNTIME_FILTER =
+ TableTestProgram.of("runtime-filter", "validates runtime filter
nodes")
+ .setupConfig(
+
OptimizerConfigOptions.TABLE_OPTIMIZER_RUNTIME_FILTER_ENABLED, true)
+ // Without table statistics, each source is estimated at
1e8 rows. This puts the
+ // filtered build side at ~1.2 GB and the probe side at
~2.4 GB, so the
+ // thresholds have to bracket these sizes for a runtime
filter to be injected.
+ .setupConfig(
+ OptimizerConfigOptions
+
.TABLE_OPTIMIZER_RUNTIME_FILTER_MAX_BUILD_DATA_SIZE,
+ MemorySize.parse("1500m"))
+ .setupConfig(
+ OptimizerConfigOptions
+
.TABLE_OPTIMIZER_RUNTIME_FILTER_MIN_PROBE_DATA_SIZE,
+ MemorySize.parse("2g"))
+ // BatchExecMultipleInput is not serializable into a
compiled plan yet
+ .setupConfig(
+
OptimizerConfigOptions.TABLE_OPTIMIZER_MULTIPLE_INPUT_ENABLED, false)
+ .setupTableSource(
+ SourceTestStep.newBuilder("fact_t")
+ .addSchema("a INT", "b BIGINT")
+ .producedBeforeRestore(
+ Row.of(1, 10L),
+ Row.of(2, 20L),
+ Row.of(3, 30L),
+ Row.of(4, 40L),
+ Row.of(5, 50L))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("dim_t")
+ .addSchema("x INT", "y BIGINT")
+ .producedBeforeRestore(
+ Row.of(2, 1L),
+ Row.of(3, 2L),
+ Row.of(3, 3L),
+ Row.of(9, 4L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("a INT", "b BIGINT", "s BIGINT")
+ .consumedBeforeRestore("+I[2, 20, 1]",
"+I[3, 30, 5]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT a, b, s FROM fact_t
JOIN "
+ + "(SELECT x, SUM(y) AS s FROM dim_t WHERE
y < 4 GROUP BY x) "
+ + "ON a = x")
+ .build();
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RuntimeFilterITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RuntimeFilterITCase.java
index 6d464489e35..b3e04f8f1bd 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RuntimeFilterITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RuntimeFilterITCase.java
@@ -20,6 +20,8 @@ package org.apache.flink.table.planner.runtime.batch.sql;
import org.apache.flink.api.common.BatchShuffleMode;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.table.api.CompiledPlan;
+import org.apache.flink.table.api.PlanReference;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.config.AggregatePhaseStrategy;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
@@ -35,6 +37,7 @@ import
org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.types.Row;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -142,6 +145,58 @@ class RuntimeFilterITCase extends BatchTestBase {
false);
}
+ @Test
+ void testRuntimeFilterCompiledPlanRoundTrip() throws Exception {
+ // BatchExecMultipleInput and BatchExecAdaptiveJoin are not
serializable into a compiled
+ // plan yet
+
tEnv.getConfig().set(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTIPLE_INPUT_ENABLED,
false);
+ tEnv.getConfig()
+ .set(
+
OptimizerConfigOptions.TABLE_OPTIMIZER_ADAPTIVE_BROADCAST_JOIN_STRATEGY,
+
OptimizerConfigOptions.AdaptiveBroadcastJoinStrategy.NONE);
+ tEnv.getConfig()
+ .set(
+ OptimizerConfigOptions
+
.TABLE_OPTIMIZER_ADAPTIVE_SKEWED_JOIN_OPTIMIZATION_STRATEGY,
+
OptimizerConfigOptions.AdaptiveSkewedJoinOptimizationStrategy.NONE);
+ tEnv.executeSql(
+ "CREATE TABLE sink (\n"
+ + " a INT,\n"
+ + " b BIGINT,\n"
+ + " c INT,\n"
+ + " d VARCHAR,\n"
+ + " e BIGINT,\n"
+ + " x INT,\n"
+ + " y INT,\n"
+ + " z BIGINT\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false'\n"
+ + ")");
+
+ CompiledPlan compiledPlan =
+ tEnv.compilePlanSql(
+ "insert into sink select * from fact, dim where x = a
and z >= 3");
+ assertThat(compiledPlan.asJsonString())
+ .contains(
+ "batch-exec-local-runtime-filter-builder_1",
+ "batch-exec-global-runtime-filter-builder_1",
+ "batch-exec-runtime-filter_1");
+
+
tEnv.loadPlan(PlanReference.fromJsonString(compiledPlan.asJsonString())).execute().await();
+
+ assertThat(TestValuesTableFactory.getResultsAsStrings("sink"))
+ .containsExactlyInAnyOrder(
+ "+I[3, 4, 3, Hallo Welt wie gehts?, 2, 3, 4, 3]",
+ "+I[3, 5, 4, ABC, 2, 3, 4, 3]",
+ "+I[3, 6, 5, BCD, 3, 3, 4, 3]",
+ "+I[5, 11, 10, GHI, 1, 5, 10, 3]",
+ "+I[5, 12, 11, HIJ, 3, 5, 10, 3]",
+ "+I[5, 13, 12, IJK, 3, 5, 10, 3]",
+ "+I[5, 14, 13, JKL, 2, 5, 10, 3]",
+ "+I[5, 15, 14, KLM, 2, 5, 10, 3]");
+ }
+
@ParameterizedTest(name = "mode = {0}, ofcg = {1}")
@MethodSource("parameters")
void testRuntimeFilterWithBuildSidePushDown(BatchShuffleMode shuffleMode,
boolean ofcg) {
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-runtime-filter_1/runtime-filter/plan/runtime-filter.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-runtime-filter_1/runtime-filter/plan/runtime-filter.json
new file mode 100644
index 00000000000..9138160cb71
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-runtime-filter_1/runtime-filter/plan/runtime-filter.json
@@ -0,0 +1,492 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`dim_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "x",
+ "dataType" : "INT"
+ }, {
+ "name" : "y",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ ]
+ } ]
+ },
+ "outputType" : "ROW<`x` INT, `y` BIGINT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, dim_t, filter=[]]], fields=[x, y])",
+ "dynamicFilteringDataListenerID" : "732a7df2-be6f-450c-ae83-afdb225c7fcc"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ } ],
+ "condition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 4,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`x` INT, `y` BIGINT>",
+ "description" : "Calc(select=[x, y], where=[(y < 4)])"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-hash-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.hash-agg.memory" : "128 mb",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "grouping" : [ 0 ],
+ "auxGrouping" : [ ],
+ "aggCalls" : [ {
+ "name" : "s",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ } ],
+ "aggInputRowType" : "ROW<`x` INT, `y` BIGINT>",
+ "isMerge" : false,
+ "isFinal" : false,
+ "supportAdaptiveLocalHashAgg" : true,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`x` INT, `sum$0` BIGINT>",
+ "description" : "LocalHashAggregate(groupBy=[x], select=[x, Partial_SUM(y)
AS sum$0])"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-local-runtime-filter-builder_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`actualRowCount` INT, `filter` VARBINARY(1)>",
+ "description" : "LocalRuntimeFilterBuilder(select=[x],
estimatedRowCount=[4758129], maxRowCount=[65536000])",
+ "buildIndices" : [ 0 ],
+ "estimatedRowCount" : 4758129,
+ "maxRowCount" : 65536000
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`actualRowCount` INT, `filter` VARBINARY(1)>",
+ "description" : "Exchange(distribution=[single])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-global-runtime-filter-builder_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`actualRowCount` INT, `filter` VARBINARY(1)>",
+ "description" : "GlobalRuntimeFilterBuilder(select=[x],
estimatedRowCount=[4758129], maxRowCount=[65536000])",
+ "maxRowCount" : 65536000
+ }, {
+ "id" : 7,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "BROADCAST"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`actualRowCount` INT, `filter` VARBINARY(1)>",
+ "description" : "Exchange(distribution=[broadcast])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`fact_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, fact_t]], fields=[a, b])",
+ "dynamicFilteringDataListenerID" : "bbc85401-eb6f-4adf-8fc9-88a074b1a04c"
+ }, {
+ "id" : 9,
+ "type" : "batch-exec-runtime-filter_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "BROADCAST"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "ANY"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "RuntimeFilter(select=[a],
estimatedFilterRatio=[0.9524187090179798])",
+ "probeIndices" : [ 0 ]
+ }, {
+ "id" : 10,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT>",
+ "description" : "Exchange(distribution=[hash[a]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 11,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`x` INT, `sum$0` BIGINT>",
+ "description" : "Exchange(distribution=[hash[x]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 12,
+ "type" : "batch-exec-hash-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.hash-agg.memory" : "128 mb",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "grouping" : [ 0 ],
+ "auxGrouping" : [ ],
+ "aggCalls" : [ {
+ "name" : "s",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ } ],
+ "aggInputRowType" : "ROW<`x` INT, `y` BIGINT>",
+ "isMerge" : true,
+ "isFinal" : true,
+ "supportAdaptiveLocalHashAgg" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`x` INT, `s` BIGINT>",
+ "description" : "HashAggregate(isMerge=[true], groupBy=[x], select=[x,
Final_SUM(sum$0) AS s])"
+ }, {
+ "id" : 16,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "isStrict" : false
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`x` INT, `s` BIGINT>",
+ "description" : "Exchange(distribution=[keep_input_as_is[hash[x]]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 13,
+ "type" : "batch-exec-join_1",
+ "configuration" : {
+ "table.exec.resource.external-buffer-memory" : "10 mb",
+ "table.exec.resource.hash-join.memory" : "128 mb",
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 12,
+ "estimatedRightAvgRowSize" : 12,
+ "estimatedLeftRowCount" : 4758130,
+ "estimatedRightRowCount" : 4758129,
+ "isBroadcast" : false,
+ "leftIsBuild" : false,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `x` INT, `s` BIGINT>",
+ "description" : "HashJoin(joinType=[InnerJoin], where=[(a = x)],
select=[a, b, x, s], build=[right])",
+ "withJobStrategyHint" : false
+ }, {
+ "id" : 14,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `s` BIGINT>",
+ "description" : "Calc(select=[a, b, s])"
+ }, {
+ "id" : 15,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "s",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `s` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t],
fields=[a, b, s])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 16,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 16,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 14,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 14,
+ "target" : 15,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
