This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 8ac9c031c07 [FLINK-41263][table-planner] Add CompiledPlan support for
BatchExecAdaptiveJoin
8ac9c031c07 is described below
commit 8ac9c031c073e08a38f5f5a1446613982e2df127
Author: Bonnie Varghese <[email protected]>
AuthorDate: Fri Nov 6 03:56:05 2026 -0800
[FLINK-41263][table-planner] Add CompiledPlan support for
BatchExecAdaptiveJoin
- Build the AdaptiveJoin description prefix in the constructor instead of
getDescription(), since the serialized description is passed back to the
JsonCreator constructor on restore and would otherwise be prefixed twice.
---
.../nodes/exec/batch/BatchExecAdaptiveJoin.java | 87 ++++-
.../planner/plan/utils/ExecNodeMetadataUtil.java | 2 +
.../apache/flink/table/api/CompiledPlanITCase.java | 37 +++
.../exec/batch/AdaptiveJoinBatchRestoreTest.java | 48 +++
.../plan/nodes/exec/common/JoinTestPrograms.java | 29 ++
.../plan/join-inner-join-with-duplicate-key.json | 274 ++++++++++++++++
.../plan/join-inner-join-with-non-equi-join.json | 240 ++++++++++++++
.../join-left-join/plan/join-left-join.json | 231 ++++++++++++++
.../join-non-window-inner-join-with-null-cond.json | 351 +++++++++++++++++++++
.../plan/join-non-window-inner-join.json | 351 +++++++++++++++++++++
.../join-right-join/plan/join-right-join.json | 231 ++++++++++++++
.../join-semi-join/plan/join-semi-join.json | 224 +++++++++++++
.../plan/join-sort-merge-join.json | 245 ++++++++++++++
.../join-with-filter/plan/join-with-filter.json | 293 +++++++++++++++++
14 files changed, 2631 insertions(+), 12 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecAdaptiveJoin.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecAdaptiveJoin.java
index a103a8e349b..a82f15ce9e2 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecAdaptiveJoin.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecAdaptiveJoin.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.batch;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.ReadableConfig;
@@ -30,6 +31,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
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.SingleTransformationTranslator;
import org.apache.flink.table.planner.plan.nodes.exec.spec.JoinSpec;
@@ -43,23 +45,62 @@ import
org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.InstantiationUtil;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.io.IOException;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkState;
/** {@link BatchExecNode} for adaptive join. */
+@ExecNodeMetadata(
+ name = "batch-exec-adaptive-join",
+ version = 1,
+ producedTransformations = BatchExecAdaptiveJoin.JOIN_TRANSFORMATION,
+ consumedOptions = {
+ "table.exec.resource.hash-join.memory",
+ "table.exec.resource.external-buffer-memory",
+ "table.exec.resource.sort.memory"
+ },
+ minPlanVersion = FlinkVersion.v2_5,
+ minStateVersion = FlinkVersion.v2_5)
public class BatchExecAdaptiveJoin extends ExecNodeBase<RowData>
implements BatchExecNode<RowData>,
SingleTransformationTranslator<RowData> {
+ public static final String JOIN_TRANSFORMATION = "join";
+
+ public static final String FIELD_NAME_JOIN_SPEC = "joinSpec";
+ public static final String FIELD_NAME_ESTIMATED_LEFT_AVG_ROW_SIZE =
"estimatedLeftAvgRowSize";
+ public static final String FIELD_NAME_ESTIMATED_RIGHT_AVG_ROW_SIZE =
"estimatedRightAvgRowSize";
+ public static final String FIELD_NAME_ESTIMATED_LEFT_ROW_COUNT =
"estimatedLeftRowCount";
+ public static final String FIELD_NAME_ESTIMATED_RIGHT_ROW_COUNT =
"estimatedRightRowCount";
+ public static final String FIELD_NAME_LEFT_IS_BUILD = "leftIsBuild";
+ public static final String FIELD_NAME_TRY_DISTINCT_BUILD_ROW =
"tryDistinctBuildRow";
+ public static final String FIELD_NAME_ORIGINAL_JOIN = "originalJoin";
+
+ @JsonProperty(FIELD_NAME_JOIN_SPEC)
private final JoinSpec joinSpec;
+
+ @JsonProperty(FIELD_NAME_LEFT_IS_BUILD)
private final boolean leftIsBuild;
+
+ @JsonProperty(FIELD_NAME_ESTIMATED_LEFT_AVG_ROW_SIZE)
private final int estimatedLeftAvgRowSize;
+
+ @JsonProperty(FIELD_NAME_ESTIMATED_RIGHT_AVG_ROW_SIZE)
private final int estimatedRightAvgRowSize;
+
+ @JsonProperty(FIELD_NAME_ESTIMATED_LEFT_ROW_COUNT)
private final long estimatedLeftRowCount;
+
+ @JsonProperty(FIELD_NAME_ESTIMATED_RIGHT_ROW_COUNT)
private final long estimatedRightRowCount;
+
+ @JsonProperty(FIELD_NAME_TRY_DISTINCT_BUILD_ROW)
private final boolean tryDistinctBuildRow;
- private final String description;
+
+ @JsonProperty(FIELD_NAME_ORIGINAL_JOIN)
private final OperatorType originalJoin;
public BatchExecAdaptiveJoin(
@@ -81,7 +122,12 @@ public class BatchExecAdaptiveJoin extends
ExecNodeBase<RowData>
ExecNodeContext.newPersistedConfig(BatchExecAdaptiveJoin.class, tableConfig),
inputProperties,
outputType,
- description);
+ // Built here rather than in getDescription(), which is
serialized and passed
+ // back to the JsonCreator constructor on restore.
+ "AdaptiveJoin(originalJoin=["
+ + originalJoin
+ + "], "
+ + description.substring(description.indexOf('(') + 1));
this.joinSpec = joinSpec;
this.estimatedLeftAvgRowSize = estimatedLeftAvgRowSize;
this.estimatedRightAvgRowSize = estimatedRightAvgRowSize;
@@ -89,7 +135,6 @@ public class BatchExecAdaptiveJoin extends
ExecNodeBase<RowData>
this.estimatedRightRowCount = estimatedRightRowCount;
this.leftIsBuild = leftIsBuild;
this.tryDistinctBuildRow = tryDistinctBuildRow;
- this.description = description;
checkState(
originalJoin == OperatorType.ShuffleHashJoin
|| originalJoin == OperatorType.SortMergeJoin,
@@ -101,6 +146,33 @@ public class BatchExecAdaptiveJoin extends
ExecNodeBase<RowData>
this.originalJoin = originalJoin;
}
+ @JsonCreator
+ public BatchExecAdaptiveJoin(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig
persistedConfig,
+ @JsonProperty(FIELD_NAME_JOIN_SPEC) JoinSpec joinSpec,
+ @JsonProperty(FIELD_NAME_ESTIMATED_LEFT_AVG_ROW_SIZE) int
estimatedLeftAvgRowSize,
+ @JsonProperty(FIELD_NAME_ESTIMATED_RIGHT_AVG_ROW_SIZE) int
estimatedRightAvgRowSize,
+ @JsonProperty(FIELD_NAME_ESTIMATED_LEFT_ROW_COUNT) long
estimatedLeftRowCount,
+ @JsonProperty(FIELD_NAME_ESTIMATED_RIGHT_ROW_COUNT) long
estimatedRightRowCount,
+ @JsonProperty(FIELD_NAME_LEFT_IS_BUILD) boolean leftIsBuild,
+ @JsonProperty(FIELD_NAME_TRY_DISTINCT_BUILD_ROW) boolean
tryDistinctBuildRow,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty>
inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_ORIGINAL_JOIN) OperatorType originalJoin)
{
+ super(id, context, persistedConfig, inputProperties, outputType,
description);
+ this.joinSpec = joinSpec;
+ this.estimatedLeftAvgRowSize = estimatedLeftAvgRowSize;
+ this.estimatedRightAvgRowSize = estimatedRightAvgRowSize;
+ this.estimatedLeftRowCount = estimatedLeftRowCount;
+ this.estimatedRightRowCount = estimatedRightRowCount;
+ this.leftIsBuild = leftIsBuild;
+ this.tryDistinctBuildRow = tryDistinctBuildRow;
+ this.originalJoin = originalJoin;
+ }
+
@Override
@SuppressWarnings("unchecked")
protected Transformation<RowData> translateToPlanInternal(
@@ -177,13 +249,4 @@ public class BatchExecAdaptiveJoin extends
ExecNodeBase<RowData>
throw new TableException("The adaptive join operator failed to
serialize.", e);
}
}
-
- @Override
- public String getDescription() {
- return "AdaptiveJoin("
- + "originalJoin=["
- + originalJoin
- + "], "
- + description.substring(description.indexOf('(') + 1);
- }
}
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 79f2f14720e..f824cb29067 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
@@ -28,6 +28,7 @@ import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.MultipleExecNodeMetadata;
+import
org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecAdaptiveJoin;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecCalc;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecCorrelate;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecExchange;
@@ -215,6 +216,7 @@ public final class ExecNodeMetadataUtil {
add(BatchExecRuntimeFilter.class);
add(BatchExecLocalRuntimeFilterBuilder.class);
add(BatchExecGlobalRuntimeFilterBuilder.class);
+ add(BatchExecAdaptiveJoin.class);
}
};
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
index 0d5c13d48a5..faf5e1e73ad 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
@@ -22,10 +22,12 @@ import org.apache.flink.FlinkVersion;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.table.planner.utils.JsonPlanTestBase;
import org.apache.flink.table.planner.utils.JsonTestUtils;
import org.apache.flink.table.planner.utils.TableTestUtil;
import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.types.Row;
import org.apache.commons.io.FileUtils;
import org.junit.jupiter.api.BeforeEach;
@@ -434,6 +436,41 @@ class CompiledPlanITCase extends JsonPlanTestBase {
assertResult(DATA, sinkPath);
}
+ @Test
+ void testCompileAndExecutePlanBatchModeWithAdaptiveJoin() throws Exception
{
+ tableEnv = TableEnvironment.create(EnvironmentSettings.inBatchMode());
+ tableEnv.executeSql(
+ String.format(
+ "CREATE TABLE t1 (a INT, b BIGINT) WITH ("
+ + "'connector' = 'values', 'bounded' = 'true',
'data-id' = '%s')",
+ TestValuesTableFactory.registerData(
+ Arrays.asList(Row.of(1, 10L), Row.of(2, 20L),
Row.of(3, 30L)))));
+ tableEnv.executeSql(
+ String.format(
+ "CREATE TABLE t2 (x INT, y STRING) WITH ("
+ + "'connector' = 'values', 'bounded' = 'true',
'data-id' = '%s')",
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(1, "one"), Row.of(3, "three"),
Row.of(4, "four")))));
+ tableEnv.executeSql(
+ "CREATE TABLE sink_t (a INT, b BIGINT, y STRING) WITH ("
+ + "'connector' = 'values', 'sink-insert-only' =
'false')");
+
+ final String compiledJson =
+ tableEnv.compilePlanSql(
+ "INSERT INTO sink_t SELECT a, b, y FROM t1
JOIN t2 ON a = x")
+ .asJsonString();
+ assertThat(compiledJson).contains("batch-exec-adaptive-join_1");
+
+ final CompiledPlan restoredPlan =
+ tableEnv.loadPlan(PlanReference.fromJsonString(compiledJson));
+ assertThat(restoredPlan.asJsonString()).isEqualTo(compiledJson);
+
+ restoredPlan.execute().await();
+ assertThat(TestValuesTableFactory.getResultsAsStrings("sink_t"))
+ .containsExactlyInAnyOrder("+I[1, 10, one]", "+I[3, 30,
three]");
+ }
+
private File createSourceSinkTables() throws IOException {
createTestCsvSourceTable("src", DATA, COLUMNS_DEFINITION);
return createTestCsvSinkTable("sink", COLUMNS_DEFINITION);
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/AdaptiveJoinBatchRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/AdaptiveJoinBatchRestoreTest.java
new file mode 100644
index 00000000000..037e52b889f
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/AdaptiveJoinBatchRestoreTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.common.JoinTestPrograms;
+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.List;
+
+/** Batch Compiled Plan tests for {@link BatchExecAdaptiveJoin}. */
+public class AdaptiveJoinBatchRestoreTest extends BatchRestoreTestBase {
+
+ public AdaptiveJoinBatchRestoreTest() {
+ super(BatchExecAdaptiveJoin.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return Arrays.asList(
+ JoinTestPrograms.NON_WINDOW_INNER_JOIN,
+ JoinTestPrograms.NON_WINDOW_INNER_JOIN_WITH_NULL,
+ JoinTestPrograms.JOIN_WITH_FILTER,
+ JoinTestPrograms.INNER_JOIN_WITH_DUPLICATE_KEY,
+ JoinTestPrograms.INNER_JOIN_WITH_NON_EQUI_JOIN,
+ JoinTestPrograms.LEFT_JOIN,
+ JoinTestPrograms.RIGHT_JOIN,
+ JoinTestPrograms.SEMI_JOIN,
+ JoinTestPrograms.SORT_MERGE_JOIN);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
index 7cb27378d12..8b8f720129c 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.common;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecJoin;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
@@ -47,6 +48,7 @@ public class JoinTestPrograms {
public static final TableTestProgram INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI;
public static final TableTestProgram SEMI_JOIN_UPSERT_INPUTS_NON_EQUI;
public static final TableTestProgram
SEMI_JOIN_UPSERT_INPUTS_NON_EQUI_PRE_2_4;
+ public static final TableTestProgram SORT_MERGE_JOIN;
static final SourceTestStep EMPLOYEE =
SourceTestStep.newBuilder("EMPLOYEE")
@@ -579,6 +581,33 @@ public class JoinTestPrograms {
semiJoinWithUpsertInputsAndNonEquiCondition(
"join-semi-join-upsert-inputs-non-equi-pre-2-4",
"semi join with upsert inputs that delete by key and a
non-equi condition restored from a Flink 2.3 plan");
+
+ // batch only, the other join operators are disabled so that the
planner picks SortMergeJoin
+ SORT_MERGE_JOIN =
+ TableTestProgram.of("join-sort-merge-join", "sort merge join")
+ .setupConfig(
+
ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
+
"ShuffleHashJoin,BroadcastHashJoin,NestedLoopJoin")
+ .setupTableSource(EMPLOYEE)
+ .setupTableSource(DEPARTMENT)
+ .setupTableSink(
+ SinkTestStep.newBuilder("MySink")
+ .addSchema("name varchar",
"department_name varchar")
+ .consumedBeforeRestore(
+ Row.of("Baker", "Research"),
+ Row.of("Charlie", "Human
Resources"),
+ Row.of("Charlie", "HR"),
+ Row.of("Don", "Sales"))
+ .consumedAfterRestore(
+ Row.of("Helena",
"Engineering"),
+ Row.of("Juliet",
"Engineering"),
+ Row.of("Ivana", "Research"),
+ Row.of("Charlie", "People
Operations"))
+ .build())
+ .runSql(
+ "insert into MySink "
+ + "SELECT name, department_name FROM
EMPLOYEE JOIN DEPARTMENT ON deptno = department_num")
+ .build();
}
/**
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-duplicate-key/plan/join-inner-join-with-duplicate-key.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-duplicate-key/plan/join-inner-join-with-duplicate-key.json
new file mode 100644
index 00000000000..f7a28c9fbdb
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-duplicate-key/plan/join-inner-join-with-duplicate-key.json
@@ -0,0 +1,274 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 30,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ] ],
+ "producedType" : "ROW<`deptno` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`deptno` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, project=[deptno]]], fields=[deptno])",
+ "dynamicFilteringDataListenerID" : "ce5dec5b-c4da-4ec9-8f7d-e278c7c5c8d8"
+ }, {
+ "id" : 31,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 32,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : "ROW<`department_num` INT, `b3` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT, `b3` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num, b3]]],
fields=[department_num, b3])",
+ "dynamicFilteringDataListenerID" : "55eb3e60-7f64-47c2-8246-61232a045e0c"
+ }, {
+ "id" : 33,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ } ],
+ "condition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `b3` INT>",
+ "description" : "Calc(select=[department_num, b3], where=[(b3 =
department_num)])"
+ }, {
+ "id" : 34,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 1 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `b3` INT>",
+ "description" : "Exchange(distribution=[hash[b3]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 38,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 1 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 4,
+ "estimatedRightAvgRowSize" : 8,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 15000000,
+ "leftIsBuild" : false,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 1 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `department_num` INT, `b3` INT>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[InnerJoin], where=[(deptno = b3)], select=[deptno, department_num,
b3], build=[right])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 36,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `department_num` INT>",
+ "description" : "Calc(select=[deptno, department_num])"
+ }, {
+ "id" : 37,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_num",
+ "dataType" : "INT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `department_num` INT>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[deptno, department_num])"
+ } ],
+ "edges" : [ {
+ "source" : 30,
+ "target" : 31,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 32,
+ "target" : 33,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 33,
+ "target" : 34,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 31,
+ "target" : 38,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 34,
+ "target" : 38,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 38,
+ "target" : 36,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 36,
+ "target" : 37,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-non-equi-join/plan/join-inner-join-with-non-equi-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-non-equi-join/plan/join-inner-join-with-non-equi-join.json
new file mode 100644
index 00000000000..48f42c5478a
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-inner-join-with-non-equi-join/plan/join-inner-join-with-non-equi-join.json
@@ -0,0 +1,240 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 39,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`deptno` INT, `salary` BIGINT, `name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE]], fields=[deptno, salary, name])",
+ "dynamicFilteringDataListenerID" : "81928aba-8a6c-4539-b1f2-b42bed2e36c0"
+ }, {
+ "id" : 40,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `salary` BIGINT, `name`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 41,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ], [ 3 ] ],
+ "producedType" : "ROW<`department_num` INT, `b2` BIGINT,
`department_name` VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT, `b2` BIGINT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num, b2, department_name]]],
fields=[department_num, b2, department_name])",
+ "dynamicFilteringDataListenerID" : "8c383a44-2dc0-4af2-b5df-e5dfa2bde593"
+ }, {
+ "id" : 42,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `b2` BIGINT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[department_num]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 46,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "BIGINT"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "estimatedLeftAvgRowSize" : 24,
+ "estimatedRightAvgRowSize" : 24,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : true,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`deptno` INT, `salary` BIGINT, `name`
VARCHAR(2147483647), `department_num` INT, `b2` BIGINT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[InnerJoin], where=[((deptno = department_num) AND (salary > b2))],
select=[deptno, salary, name, department_num, b2, department_name],
build=[left])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 44,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name, department_name])"
+ }, {
+ "id" : 45,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name, department_name])"
+ } ],
+ "edges" : [ {
+ "source" : 39,
+ "target" : 40,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 41,
+ "target" : 42,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 40,
+ "target" : 46,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 42,
+ "target" : 46,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 46,
+ "target" : 44,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 44,
+ "target" : 45,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-left-join/plan/join-left-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-left-join/plan/join-left-join.json
new file mode 100644
index 00000000000..cad7e69fc16
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-left-join/plan/join-left-join.json
@@ -0,0 +1,231 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 47,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)> NOT
NULL"
+ } ]
+ },
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, project=[deptno, name]]], fields=[deptno, name])",
+ "dynamicFilteringDataListenerID" : "d45f8ae2-775c-43d4-bb42-e6faacced3c8"
+ }, {
+ "id" : 48,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 49,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 3 ] ],
+ "producedType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num, department_name]]],
fields=[department_num, department_name])",
+ "dynamicFilteringDataListenerID" : "4407f827-72cc-47e4-bcbf-00f27801da78"
+ }, {
+ "id" : 50,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[department_num]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 54,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "LEFT",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 16,
+ "estimatedRightAvgRowSize" : 16,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : true,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647),
`department_num` INT, `department_name` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[LeftOuterJoin], where=[(deptno = department_num)], select=[deptno,
name, department_num, department_name], build=[left])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 52,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name, department_name])"
+ }, {
+ "id" : 53,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name, department_name])"
+ } ],
+ "edges" : [ {
+ "source" : 47,
+ "target" : 48,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 49,
+ "target" : 50,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 48,
+ "target" : 54,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 50,
+ "target" : 54,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 54,
+ "target" : 52,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 52,
+ "target" : 53,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join-with-null-cond/plan/join-non-window-inner-join-with-null-cond.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join-with-null-cond/plan/join-non-window-inner-join-with-null-cond.json
new file mode 100644
index 00000000000..b72cd4bece1
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join-with-null-cond/plan/join-non-window-inner-join-with-null-cond.json
@@ -0,0 +1,351 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 11,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`T1`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, T1]], fields=[a, b, c])",
+ "dynamicFilteringDataListenerID" : "b0c704c8-8c70-4b6a-a5dd-39864009120d"
+ }, {
+ "id" : 12,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "CALL",
+ "internalName" : "$IF$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 3,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[IF((a = 3), null:INTEGER, a) AS a, b, c])"
+ }, {
+ "id" : 13,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 14,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`T2`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[a, b, c])",
+ "dynamicFilteringDataListenerID" : "87ef5f1a-939b-4405-993a-12be7d5c1a38"
+ }, {
+ "id" : 15,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "CALL",
+ "internalName" : "$IF$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 3,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[IF((a = 3), null:INTEGER, a) AS a, b, c])"
+ }, {
+ "id" : 16,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 20,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ false ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "BIGINT"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "estimatedLeftAvgRowSize" : 21,
+ "estimatedRightAvgRowSize" : 21,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : true,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `a0`
INT, `b0` BIGINT, `c0` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[InnerJoin], where=[(IS NOT DISTINCT FROM(a, a0) AND (b > b0))],
select=[a, b, c, a0, b0, c0], build=[left])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 18,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `c` VARCHAR(2147483647), `c0`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[a0 AS a, c0 AS c, c AS c0])"
+ }, {
+ "id" : 19,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "c1",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "c2",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `c` VARCHAR(2147483647), `c0`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[a, c, c0])"
+ } ],
+ "edges" : [ {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 14,
+ "target" : 15,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 15,
+ "target" : 16,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 20,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 16,
+ "target" : 20,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 20,
+ "target" : 18,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 18,
+ "target" : 19,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join/plan/join-non-window-inner-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join/plan/join-non-window-inner-join.json
new file mode 100644
index 00000000000..3a72682f8a8
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-non-window-inner-join/plan/join-non-window-inner-join.json
@@ -0,0 +1,351 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`T1`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, T1]], fields=[a, b, c])",
+ "dynamicFilteringDataListenerID" : "97521272-39c5-4351-82a4-d091e183226b"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "CALL",
+ "internalName" : "$IF$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 3,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[IF((a = 3), null:INTEGER, a) AS a, b, c])"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`T2`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[a, b, c])",
+ "dynamicFilteringDataListenerID" : "3b8ed6e4-3ebd-4857-9f13-9692c04a4fde"
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "CALL",
+ "internalName" : "$IF$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 3,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[IF((a = 3), null:INTEGER, a) AS a, b, c])"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 10,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "BIGINT"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "estimatedLeftAvgRowSize" : 21,
+ "estimatedRightAvgRowSize" : 21,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : true,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `a0`
INT, `b0` BIGINT, `c0` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[InnerJoin], where=[((a = a0) AND (b > b0))], select=[a, b, c, a0, b0,
c0], build=[left])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `c` VARCHAR(2147483647), `c0`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[a0 AS a, c0 AS c, c AS c0])"
+ }, {
+ "id" : 9,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "c1",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "c2",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `c` VARCHAR(2147483647), `c0`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[a, c, c0])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-right-join/plan/join-right-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-right-join/plan/join-right-join.json
new file mode 100644
index 00000000000..eb5c840de97
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-right-join/plan/join-right-join.json
@@ -0,0 +1,231 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 55,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)> NOT
NULL"
+ } ]
+ },
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, project=[deptno, name]]], fields=[deptno, name])",
+ "dynamicFilteringDataListenerID" : "6e22a72c-1055-4604-a881-4448efa1a88f"
+ }, {
+ "id" : 56,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 57,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 3 ] ],
+ "producedType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num, department_name]]],
fields=[department_num, department_name])",
+ "dynamicFilteringDataListenerID" : "64a5ea1d-d1a5-44d8-bc41-e919682e7b96"
+ }, {
+ "id" : 58,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[department_num]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 62,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "RIGHT",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 16,
+ "estimatedRightAvgRowSize" : 16,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : true,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647),
`department_num` INT, `department_name` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[RightOuterJoin], where=[(deptno = department_num)], select=[deptno,
name, department_num, department_name], build=[left])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 60,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name, department_name])"
+ }, {
+ "id" : 61,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name, department_name])"
+ } ],
+ "edges" : [ {
+ "source" : 55,
+ "target" : 56,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 57,
+ "target" : 58,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 56,
+ "target" : 62,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 58,
+ "target" : 62,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 62,
+ "target" : 60,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 60,
+ "target" : 61,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-semi-join/plan/join-semi-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-semi-join/plan/join-semi-join.json
new file mode 100644
index 00000000000..8190267e855
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-semi-join/plan/join-semi-join.json
@@ -0,0 +1,224 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 63,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)> NOT
NULL"
+ } ]
+ },
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, project=[deptno, name]]], fields=[deptno, name])",
+ "dynamicFilteringDataListenerID" : "4164dc59-7846-45f1-977e-f49a2b628c85"
+ }, {
+ "id" : 64,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 65,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ] ],
+ "producedType" : "ROW<`department_num` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num]]],
fields=[department_num])",
+ "dynamicFilteringDataListenerID" : "b70eedcf-e204-47b0-9b62-41b480f3b901"
+ }, {
+ "id" : 66,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT>",
+ "description" : "Exchange(distribution=[hash[department_num]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 70,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "SEMI",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 16,
+ "estimatedRightAvgRowSize" : 4,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "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<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[LeftSemiJoin], where=[(deptno = department_num)], select=[deptno,
name], build=[right])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 68,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name])"
+ }, {
+ "id" : 69,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name])"
+ } ],
+ "edges" : [ {
+ "source" : 63,
+ "target" : 64,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 65,
+ "target" : 66,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 64,
+ "target" : 70,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 66,
+ "target" : 70,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 70,
+ "target" : 68,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 68,
+ "target" : 69,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-sort-merge-join/plan/join-sort-merge-join.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-sort-merge-join/plan/join-sort-merge-join.json
new file mode 100644
index 00000000000..7830c3a3c7f
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-sort-merge-join/plan/join-sort-merge-join.json
@@ -0,0 +1,245 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)> NOT
NULL"
+ } ]
+ },
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, project=[deptno, name]]], fields=[deptno, name])",
+ "dynamicFilteringDataListenerID" : "83897844-91e5-4857-9861-eedd372691d8"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[deptno]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 3 ] ],
+ "producedType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, project=[department_num, department_name]]],
fields=[department_num, department_name])",
+ "dynamicFilteringDataListenerID" : "dd889db7-c54d-4624-99aa-e6ca4c428563"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`department_num` INT, `department_name`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[department_num]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "estimatedLeftAvgRowSize" : 16,
+ "estimatedRightAvgRowSize" : 16,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 100000000,
+ "leftIsBuild" : false,
+ "tryDistinctBuildRow" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`deptno` INT, `name` VARCHAR(2147483647),
`department_num` INT, `department_name` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[SortMergeJoin],
joinType=[InnerJoin], where=[(deptno = department_num)], select=[deptno, name,
department_num, department_name])",
+ "originalJoin" : "SortMergeJoin"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name, department_name])"
+ }, {
+ "id" : 7,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name, department_name])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-with-filter/plan/join-with-filter.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-with-filter/plan/join-with-filter.json
new file mode 100644
index 00000000000..a1c0af63e48
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-adaptive-join_1/join-with-filter/plan/join-with-filter.json
@@ -0,0 +1,293 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`EMPLOYEE`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "deptno",
+ "dataType" : "INT"
+ }, {
+ "name" : "salary",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ]
+ }, {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 1 ], [ 2 ] ],
+ "producedType" : "ROW<`salary` BIGINT, `name` VARCHAR(2147483647)> NOT
NULL"
+ } ]
+ },
+ "outputType" : "ROW<`salary` BIGINT, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, EMPLOYEE, filter=[<(salary, 2:BIGINT)], project=[salary,
name]]], fields=[salary, name])",
+ "dynamicFilteringDataListenerID" : "9078470d-f477-47a5-8e2b-24ba5d2ac5c8"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`salary` BIGINT, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[salary]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`DEPARTMENT`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "department_num",
+ "dataType" : "INT"
+ }, {
+ "name" : "b2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b3",
+ "dataType" : "INT"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown"
+ }, {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 1 ], [ 3 ] ],
+ "producedType" : "ROW<`b2` BIGINT, `department_name`
VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`b2` BIGINT, `department_name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, DEPARTMENT, filter=[], project=[b2, department_name]]],
fields=[b2, department_name])",
+ "dynamicFilteringDataListenerID" : "8a410a67-1030-46bf-8ea3-003f0ec62c7d"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b2` BIGINT, `department_name` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[b2, department_name], where=[(b2 < 2)])"
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b2` BIGINT, `department_name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[b2]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 9,
+ "type" : "batch-exec-adaptive-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"
+ },
+ "joinSpec" : {
+ "joinType" : "INNER",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : null
+ },
+ "estimatedLeftAvgRowSize" : 20,
+ "estimatedRightAvgRowSize" : 20,
+ "estimatedLeftRowCount" : 100000000,
+ "estimatedRightRowCount" : 50000000,
+ "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<`salary` BIGINT, `name` VARCHAR(2147483647), `b2`
BIGINT, `department_name` VARCHAR(2147483647)>",
+ "description" : "AdaptiveJoin(originalJoin=[ShuffleHashJoin],
joinType=[InnerJoin], where=[(salary = b2)], select=[salary, name, b2,
department_name], build=[right])",
+ "originalJoin" : "ShuffleHashJoin"
+ }, {
+ "id" : 7,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[name, department_name])"
+ }, {
+ "id" : 8,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "department_name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`name` VARCHAR(2147483647), `department_name`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[name, department_name])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
