You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2024/11/27 13:46:12 UTC
(flink) branch master updated: [FLINK-36718][table-planner] Add CompiledPlan annotations to BatchExecLookupJoin
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 762bb5ab2a6 [FLINK-36718][table-planner] Add CompiledPlan annotations to BatchExecLookupJoin
762bb5ab2a6 is described below
commit 762bb5ab2a6a5ed48f3d60501b9e1c0d362ad8cb
Author: Jim Hughes <jh...@confluent.io>
AuthorDate: Wed Nov 27 08:46:04 2024 -0500
[FLINK-36718][table-planner] Add CompiledPlan annotations to BatchExecLookupJoin
---
.../plan/nodes/exec/batch/BatchExecLookupJoin.java | 58 ++++
.../planner/plan/utils/ExecNodeMetadataUtil.java | 2 +
.../LookupJoinBatchRestoreTest.java} | 21 +-
.../{stream => common}/LookupJoinTestPrograms.java | 20 +-
.../nodes/exec/stream/LookupJoinRestoreTest.java | 1 +
.../plan/lookup-join-filter-pushdown.json | 285 ++++++++++++++++++++
.../plan/lookup-join-left-join.json | 300 +++++++++++++++++++++
.../plan/lookup-join-pre-filter.json | 300 +++++++++++++++++++++
8 files changed, 964 insertions(+), 23 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java
index 42e6da85a71..9dfc2ebf61a 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.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.ReadableConfig;
import org.apache.flink.table.connector.ChangelogMode;
@@ -26,6 +27,7 @@ import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.planner.delegation.PlannerBase;
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.common.CommonExecLookupJoin;
@@ -34,6 +36,9 @@ import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
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 org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.tools.RelBuilder;
@@ -45,8 +50,20 @@ import java.util.List;
import java.util.Map;
/** {@link BatchExecNode} for temporal table join that implemented by lookup. */
+@ExecNodeMetadata(
+ name = "batch-exec-lookup-join",
+ version = 1,
+ producedTransformations = CommonExecLookupJoin.LOOKUP_JOIN_TRANSFORMATION,
+ consumedOptions = {
+ "table.exec.async-lookup.buffer-capacity",
+ "table.exec.async-lookup.timeout",
+ "table.exec.async-lookup.output-mode"
+ },
+ minPlanVersion = FlinkVersion.v2_0,
+ minStateVersion = FlinkVersion.v2_0)
public class BatchExecLookupJoin extends CommonExecLookupJoin
implements BatchExecNode<RowData>, SingleTransformationTranslator<RowData> {
+
public BatchExecLookupJoin(
ReadableConfig tableConfig,
FlinkJoinType joinType,
@@ -80,6 +97,47 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin
description);
}
+ @JsonCreator
+ public BatchExecLookupJoin(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
+ @JsonProperty(FIELD_NAME_JOIN_TYPE) FlinkJoinType joinType,
+ @JsonProperty(FIELD_NAME_PRE_FILTER_CONDITION) @Nullable RexNode preFilterCondition,
+ @JsonProperty(FIELD_NAME_REMAINING_JOIN_CONDITION) @Nullable
+ RexNode remainingJoinCondition,
+ @JsonProperty(FIELD_NAME_TEMPORAL_TABLE)
+ TemporalTableSourceSpec temporalTableSourceSpec,
+ @JsonProperty(FIELD_NAME_LOOKUP_KEYS) Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ @JsonProperty(FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE) @Nullable
+ List<RexNode> projectionOnTemporalTable,
+ @JsonProperty(FIELD_NAME_FILTER_ON_TEMPORAL_TABLE) @Nullable
+ RexNode filterOnTemporalTable,
+ @JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable
+ LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+ super(
+ id,
+ context,
+ persistedConfig,
+ joinType,
+ preFilterCondition,
+ remainingJoinCondition,
+ temporalTableSourceSpec,
+ lookupKeys,
+ projectionOnTemporalTable,
+ filterOnTemporalTable,
+ asyncLookupOptions,
+ // batch lookup join does not support retry hint currently
+ null,
+ ChangelogMode.insertOnly(),
+ inputProperties,
+ outputType,
+ description);
+ }
+
@Override
@SuppressWarnings("unchecked")
public 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 e9b6e6a40ac..a19fc0dbaf9 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
@@ -35,6 +35,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecExpand;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashAggregate;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashJoin;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecLimit;
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecLookupJoin;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecMatch;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecNestedLoopJoin;
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecOverAggregate;
@@ -184,6 +185,7 @@ public final class ExecNodeMetadataUtil {
add(BatchExecSortAggregate.class);
add(BatchExecSortLimit.class);
add(BatchExecWindowTableFunction.class);
+ add(BatchExecLookupJoin.class);
add(BatchExecMatch.class);
add(BatchExecOverAggregate.class);
add(BatchExecRank.class);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LookupJoinBatchRestoreTest.java
similarity index 62%
copy from flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java
copy to flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LookupJoinBatchRestoreTest.java
index 8700a498104..41a12b5b168 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LookupJoinBatchRestoreTest.java
@@ -16,32 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.plan.nodes.exec.stream;
+package org.apache.flink.table.planner.plan.nodes.exec.batch;
-import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms;
+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;
-/** Restore tests for {@link StreamExecLookupJoin}. */
-public class LookupJoinRestoreTest extends RestoreTestBase {
+/** Batch Compiled Plan tests for {@link BatchExecLookupJoin}. */
+public class LookupJoinBatchRestoreTest extends BatchRestoreTestBase {
- public LookupJoinRestoreTest() {
- super(StreamExecLookupJoin.class);
+ public LookupJoinBatchRestoreTest() {
+ super(BatchExecLookupJoin.class);
}
@Override
public List<TableTestProgram> programs() {
return Arrays.asList(
- LookupJoinTestPrograms.LOOKUP_JOIN_PROJECT_PUSHDOWN,
LookupJoinTestPrograms.LOOKUP_JOIN_FILTER_PUSHDOWN,
LookupJoinTestPrograms.LOOKUP_JOIN_LEFT_JOIN,
- LookupJoinTestPrograms.LOOKUP_JOIN_PRE_FILTER,
- LookupJoinTestPrograms.LOOKUP_JOIN_POST_FILTER,
- LookupJoinTestPrograms.LOOKUP_JOIN_PRE_POST_FILTER,
- LookupJoinTestPrograms.LOOKUP_JOIN_ASYNC_HINT,
- LookupJoinTestPrograms.LOOKUP_JOIN_RETRY_HINT,
- LookupJoinTestPrograms.LOOKUP_JOIN_WITH_TRY_RESOLVE);
+ LookupJoinTestPrograms.LOOKUP_JOIN_PRE_FILTER);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java
similarity index 97%
rename from flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinTestPrograms.java
rename to flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java
index 487560f62ca..64f594a5d4d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.plan.nodes.exec.stream;
+package org.apache.flink.table.planner.plan.nodes.exec.common;
import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.test.program.SinkTestStep;
@@ -142,7 +142,7 @@ public class LookupJoinTestPrograms {
"state STRING",
"zipcode INT");
- static final TableTestProgram LOOKUP_JOIN_PROJECT_PUSHDOWN =
+ public static final TableTestProgram LOOKUP_JOIN_PROJECT_PUSHDOWN =
TableTestProgram.of(
"lookup-join-project-pushdown",
"validates lookup join with project pushdown")
@@ -179,7 +179,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id")
.build();
- static final TableTestProgram LOOKUP_JOIN_FILTER_PUSHDOWN =
+ public static final TableTestProgram LOOKUP_JOIN_FILTER_PUSHDOWN =
TableTestProgram.of(
"lookup-join-filter-pushdown",
"validates lookup join with filter pushdown")
@@ -211,7 +211,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id AND C.age > 30")
.build();
- static final TableTestProgram LOOKUP_JOIN_PRE_FILTER =
+ public static final TableTestProgram LOOKUP_JOIN_PRE_FILTER =
TableTestProgram.of("lookup-join-pre-filter", "validates lookup join with pre filter")
.setupTableSource(CUSTOMERS)
.setupTableSource(ORDERS)
@@ -244,7 +244,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id AND C.age > 30 AND O.total > 15.3")
.build();
- static final TableTestProgram LOOKUP_JOIN_POST_FILTER =
+ public static final TableTestProgram LOOKUP_JOIN_POST_FILTER =
TableTestProgram.of("lookup-join-post-filter", "validates lookup join with post filter")
.setupTableSource(CUSTOMERS)
.setupTableSource(ORDERS)
@@ -277,7 +277,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id AND CAST(O.total AS INT) < C.age")
.build();
- static final TableTestProgram LOOKUP_JOIN_PRE_POST_FILTER =
+ public static final TableTestProgram LOOKUP_JOIN_PRE_POST_FILTER =
TableTestProgram.of(
"lookup-join-pre-post-filter",
"validates lookup join with pre and post filters")
@@ -312,7 +312,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id AND O.total > 15.3 AND CAST(O.total AS INT) < C.age")
.build();
- static final TableTestProgram LOOKUP_JOIN_LEFT_JOIN =
+ public static final TableTestProgram LOOKUP_JOIN_LEFT_JOIN =
TableTestProgram.of("lookup-join-left-join", "validates lookup join with left join")
.setupTableSource(CUSTOMERS)
.setupTableSource(ORDERS)
@@ -345,7 +345,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id AND C.age > 30 AND O.customer_id <> 3")
.build();
- static final TableTestProgram LOOKUP_JOIN_ASYNC_HINT =
+ public static final TableTestProgram LOOKUP_JOIN_ASYNC_HINT =
TableTestProgram.of("lookup-join-async-hint", "validates lookup join with async hint")
.setupTableSource(CUSTOMERS_ASYNC)
.setupTableSource(ORDERS)
@@ -379,7 +379,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id")
.build();
- static final TableTestProgram LOOKUP_JOIN_RETRY_HINT =
+ public static final TableTestProgram LOOKUP_JOIN_RETRY_HINT =
TableTestProgram.of("lookup-join-retry-hint", "validates lookup join with retry hint")
.setupTableSource(CUSTOMERS)
.setupTableSource(ORDERS)
@@ -413,7 +413,7 @@ public class LookupJoinTestPrograms {
+ "ON O.customer_id = C.id")
.build();
- static final TableTestProgram LOOKUP_JOIN_WITH_TRY_RESOLVE =
+ public static final TableTestProgram LOOKUP_JOIN_WITH_TRY_RESOLVE =
TableTestProgram.of(
"lookup-join-with-try-resolve",
"validates lookup join with NUD try resolve strategy enabled")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java
index 8700a498104..56fc84ed3aa 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinRestoreTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms;
import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
import org.apache.flink.table.test.program.TableTestProgram;
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-filter-pushdown/plan/lookup-join-filter-pushdown.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-filter-pushdown/plan/lookup-join-filter-pushdown.json
new file mode 100644
index 00000000000..6a418d6e18f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-filter-pushdown/plan/lookup-join-filter-pushdown.json
@@ -0,0 +1,285 @@
+{
+ "flinkVersion" : "2.0",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`orders_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "customer_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "order_time",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "proc_time",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$PROCTIME$1",
+ "operands" : [ ],
+ "type" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ },
+ "serializableString" : "PROCTIME()"
+ }
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ], [ 2 ] ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, orders_t, project=[order_id, customer_id, total], metadata=[]]], fields=[order_id, customer_id, total])",
+ "dynamicFilteringDataListenerID" : "1101966a-5ce6-4069-9e00-83dd0f0750b6",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-lookup-join_1",
+ "configuration" : {
+ "table.exec.async-lookup.buffer-capacity" : "100",
+ "table.exec.async-lookup.output-mode" : "ORDERED",
+ "table.exec.async-lookup.timeout" : "3 min"
+ },
+ "joinType" : "INNER",
+ "joinCondition" : null,
+ "temporalTable" : {
+ "lookupTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`customers_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "id" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT> NOT NULL"
+ },
+ "lookupKeys" : {
+ "0" : {
+ "type" : "FieldRef",
+ "index" : 1
+ }
+ },
+ "projectionOnTemporalTable" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ } ],
+ "filterOnTemporalTable" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 30,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE, `id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "LookupJoin(table=[default_catalog.default_database.customers_t], joinType=[InnerJoin], lookup=[id=customer_id], where=[(age > 30)], select=[order_id, customer_id, total, id, name, age, city, state, zipcode])",
+ "inputChangelogMode" : [ "INSERT" ]
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "INT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Calc(select=[order_id, total, id, name, age, city, state, zipcode])"
+ }, {
+ "id" : 4,
+ "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" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "id",
+ "dataType" : "INT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[order_id, total, id, name, age, city, state, zipcode])"
+ } ],
+ "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"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-left-join/plan/lookup-join-left-join.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-left-join/plan/lookup-join-left-join.json
new file mode 100644
index 00000000000..28e7f3328a0
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-left-join/plan/lookup-join-left-join.json
@@ -0,0 +1,300 @@
+{
+ "flinkVersion" : "2.0",
+ "nodes" : [ {
+ "id" : 5,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`orders_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "customer_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "order_time",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "proc_time",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$PROCTIME$1",
+ "operands" : [ ],
+ "type" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ },
+ "serializableString" : "PROCTIME()"
+ }
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ], [ 2 ] ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, orders_t, project=[order_id, customer_id, total], metadata=[]]], fields=[order_id, customer_id, total])",
+ "dynamicFilteringDataListenerID" : "ec755f01-b1c5-4c36-ba41-dd548bad0a6a",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-lookup-join_1",
+ "configuration" : {
+ "table.exec.async-lookup.buffer-capacity" : "100",
+ "table.exec.async-lookup.output-mode" : "ORDERED",
+ "table.exec.async-lookup.timeout" : "3 min"
+ },
+ "joinType" : "LEFT",
+ "preFilterCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 3,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "joinCondition" : null,
+ "temporalTable" : {
+ "lookupTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`customers_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "id" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT> NOT NULL"
+ },
+ "lookupKeys" : {
+ "0" : {
+ "type" : "FieldRef",
+ "index" : 1
+ }
+ },
+ "projectionOnTemporalTable" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ } ],
+ "filterOnTemporalTable" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 30,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "LookupJoin(table=[default_catalog.default_database.customers_t], joinType=[LeftOuterJoin], lookup=[id=customer_id], where=[(age > 30)], joinCondition=[(customer_id <> 3)], select=[order_id, customer_id, total, id, name, age, city, state, zipcode])",
+ "inputChangelogMode" : [ "INSERT" ]
+ }, {
+ "id" : 7,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "INT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Calc(select=[order_id, total, id, name, age, city, state, zipcode])"
+ }, {
+ "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`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "id",
+ "dataType" : "INT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[order_id, total, id, name, age, city, state, zipcode])"
+ } ],
+ "edges" : [ {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 8,
+ "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-lookup-join_1/lookup-join-pre-filter/plan/lookup-join-pre-filter.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-pre-filter/plan/lookup-join-pre-filter.json
new file mode 100644
index 00000000000..e8d63344e4c
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-lookup-join_1/lookup-join-pre-filter/plan/lookup-join-pre-filter.json
@@ -0,0 +1,300 @@
+{
+ "flinkVersion" : "2.0",
+ "nodes" : [ {
+ "id" : 9,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`orders_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "customer_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "order_time",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "proc_time",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$PROCTIME$1",
+ "operands" : [ ],
+ "type" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ },
+ "serializableString" : "PROCTIME()"
+ }
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ], [ 2 ] ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, orders_t, project=[order_id, customer_id, total], metadata=[]]], fields=[order_id, customer_id, total])",
+ "dynamicFilteringDataListenerID" : "50059ffe-2f74-4987-8249-27b834c74daa",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 10,
+ "type" : "batch-exec-lookup-join_1",
+ "configuration" : {
+ "table.exec.async-lookup.buffer-capacity" : "100",
+ "table.exec.async-lookup.output-mode" : "ORDERED",
+ "table.exec.async-lookup.timeout" : "3 min"
+ },
+ "joinType" : "LEFT",
+ "preFilterCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "15.3",
+ "type" : "DECIMAL(3, 1) NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "joinCondition" : null,
+ "temporalTable" : {
+ "lookupTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`customers_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "id" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`id` INT NOT NULL, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT> NOT NULL"
+ },
+ "lookupKeys" : {
+ "0" : {
+ "type" : "FieldRef",
+ "index" : 1
+ }
+ },
+ "projectionOnTemporalTable" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ } ],
+ "filterOnTemporalTable" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 30,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `customer_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "LookupJoin(table=[default_catalog.default_database.customers_t], joinType=[LeftOuterJoin], lookup=[id=customer_id], where=[(age > 30)], joinCondition=[(total > 15.3)], select=[order_id, customer_id, total, id, name, age, city, state, zipcode])",
+ "inputChangelogMode" : [ "INSERT" ]
+ }, {
+ "id" : 11,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "INT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Calc(select=[order_id, total, id, name, age, city, state, zipcode])"
+ }, {
+ "id" : 12,
+ "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" : "order_id",
+ "dataType" : "INT"
+ }, {
+ "name" : "total",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "id",
+ "dataType" : "INT"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "age",
+ "dataType" : "INT"
+ }, {
+ "name" : "city",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "state",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "zipcode",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT, `total` DOUBLE, `id` INT, `name` VARCHAR(2147483647), `age` INT, `city` VARCHAR(2147483647), `state` VARCHAR(2147483647), `zipcode` INT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[order_id, total, id, name, age, city, state, zipcode])"
+ } ],
+ "edges" : [ {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file