This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d9af54c7690b871159e6f64dcffa2fc809ad85e5 Author: Hanyu (Peter) Zheng <[email protected]> AuthorDate: Tue Oct 27 11:23:33 2026 -0500 [FLINK-41217][table-planner] Don't emit the rows of VALUES again after a restore In streaming mode VALUES runs as an InputFormatSourceFunction, which has no state. After VALUES emitted its rows, the task waits for one more checkpoint before it exits. That checkpoint is triggered while the task is still running, so it doesn't mark VALUES as finished. A job that restores from it emits all rows again and downstream operators see duplicates. Version 2 of StreamExecValues runs VALUES as a ValuesSource instead, also when it has no rows. Batch mode keeps the legacy source. Unlike the legacy source, a FLIP-27 source can be chained with a yielding operator, such as the writer of a SinkV2 sink or an async operator. The chained operator then has the max parallelism of VALUES, which is 1, so a checkpoint taken with the legacy source can't be restored. A compiled plan with version 1 of the node therefore keeps the legacy source. Without explicit UIDs, the generated IDs of the operator before the yielding one and of all operators after it change as well. --- .../plan/nodes/exec/common/CommonExecValues.java | 47 +++-- .../plan/nodes/exec/stream/StreamExecValues.java | 16 ++ .../plan/nodes/exec/TransformationsTest.java | 77 ++++++-- .../stream/sql/ValuesStreamingRestoreITCase.java | 204 +++++++++++++++++++++ .../values-test/plan/values-test.json | 82 +++++++++ 5 files changed, 402 insertions(+), 24 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecValues.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecValues.java index bd1a9134cc1..bf1135b055c 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecValues.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecValues.java @@ -18,6 +18,7 @@ package org.apache.flink.table.planner.plan.nodes.exec.common; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.dag.Transformation; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.data.RowData; @@ -29,6 +30,8 @@ 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.SingleTransformationTranslator; import org.apache.flink.table.runtime.operators.values.ValuesInputFormat; +import org.apache.flink.table.runtime.operators.values.ValuesSource; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; @@ -62,23 +65,45 @@ public abstract class CommonExecValues extends ExecNodeBase<RowData> @Override protected Transformation<RowData> translateToPlanInternal( PlannerBase planner, ExecNodeConfig config) { - final ValuesInputFormat inputFormat = - ValuesCodeGenerator.generatorInputFormat( - config, - planner.getFlinkContext().getClassLoader(), - (RowType) getOutputType(), - tuples, - getClass().getSimpleName()); - final Transformation<RowData> transformation = - planner.getExecEnv() - .createInput(inputFormat, inputFormat.getProducedType()) - .getTransformation(); + final ClassLoader classLoader = planner.getFlinkContext().getClassLoader(); + final RowType outputType = (RowType) getOutputType(); + final String description = getClass().getSimpleName(); + final Transformation<RowData> transformation; + if (legacySourceEnabled()) { + final ValuesInputFormat inputFormat = + ValuesCodeGenerator.generatorInputFormat( + config, classLoader, outputType, tuples, description); + transformation = + planner.getExecEnv() + .createInput(inputFormat, inputFormat.getProducedType()) + .getTransformation(); + } else { + final ValuesSource source = + new ValuesSource( + ValuesCodeGenerator.generatorFunction( + config, classLoader, outputType, tuples, description), + InternalTypeInfo.of(outputType), + tuples.size()); + transformation = + planner.getExecEnv() + .fromSource( + source, WatermarkStrategy.noWatermarks(), VALUES_TRANSFORMATION) + .getTransformation(); + } createTransformationMeta(VALUES_TRANSFORMATION, config).fill(transformation); transformation.setParallelism(1); transformation.setMaxParallelism(1); return transformation; } + /** + * Whether VALUES runs as an {@code InputFormatSourceFunction} instead of a {@link + * ValuesSource}, which checkpoints the position of the next row to emit. + */ + protected boolean legacySourceEnabled() { + return true; + } + @JsonProperty(value = FIELD_NAME_TUPLES) public List<List<RexLiteral>> getTuples() { return tuples; diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecValues.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecValues.java index 9ace41eaf89..e1132433910 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecValues.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecValues.java @@ -35,12 +35,21 @@ import org.apache.calcite.rex.RexLiteral; import java.util.List; /** Stream {@link ExecNode} that read records from given values. */ +// Version 1: Initial version @ExecNodeMetadata( name = "stream-exec-values", version = 1, producedTransformations = CommonExecValues.VALUES_TRANSFORMATION, minPlanVersion = FlinkVersion.v1_15, minStateVersion = FlinkVersion.v1_15) +// Version 2: VALUES runs as a FLIP-27 source that checkpoints the position of the next row to +// emit, so a restored job doesn't emit the rows again. +@ExecNodeMetadata( + name = "stream-exec-values", + version = 2, + producedTransformations = CommonExecValues.VALUES_TRANSFORMATION, + minPlanVersion = FlinkVersion.v2_5, + minStateVersion = FlinkVersion.v2_5) public class StreamExecValues extends CommonExecValues implements StreamExecNode<RowData> { public StreamExecValues( @@ -67,4 +76,11 @@ public class StreamExecValues extends CommonExecValues implements StreamExecNode @JsonProperty(FIELD_NAME_DESCRIPTION) String description) { super(id, context, persistedConfig, tuples, outputType, description); } + + // A compiled plan with version 1 keeps the legacy source, so that it can still be restored + // from its checkpoints. + @Override + protected final boolean legacySourceEnabled() { + return getVersion() == 1; + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java index 4b970d83b30..b5f16738676 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java @@ -26,6 +26,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.streaming.api.lineage.LineageDataset; import org.apache.flink.streaming.api.transformations.LegacySourceTransformation; +import org.apache.flink.streaming.api.transformations.SourceTransformation; import org.apache.flink.streaming.api.transformations.WithBoundedness; import org.apache.flink.table.api.CompiledPlan; import org.apache.flink.table.api.DataTypes; @@ -45,6 +46,7 @@ import org.apache.flink.table.planner.factories.TableFactoryHarness; import org.apache.flink.table.planner.factories.TableFactoryHarness.ScanSourceBase; import org.apache.flink.table.planner.lineage.TableSourceLineageVertex; import org.apache.flink.table.planner.utils.JsonTestUtils; +import org.apache.flink.table.runtime.operators.values.ValuesSource; import org.apache.flink.util.Collector; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; @@ -90,8 +92,9 @@ class TransformationsTest { .schema(dummySchema()) .build()); - final LegacySourceTransformation<?> sourceTransform = - toLegacySourceTransformation(env, table); + final Transformation<?> leaf = toLeaf(env.toChangelogStream(table).getTransformation()); + assertThat(leaf).isInstanceOf(LegacySourceTransformation.class); + final LegacySourceTransformation<?> sourceTransform = (LegacySourceTransformation<?>) leaf; assertBoundedness(Boundedness.BOUNDED, sourceTransform); assertThat(sourceTransform.getOperator().emitsProgressiveWatermarks()).isFalse(); @@ -120,8 +123,9 @@ class TransformationsTest { .schema(dummySchema()) .build()); - final LegacySourceTransformation<?> sourceTransform = - toLegacySourceTransformation(env, table); + final Transformation<?> leaf = toLeaf(env.toChangelogStream(table).getTransformation()); + assertThat(leaf).isInstanceOf(LegacySourceTransformation.class); + final LegacySourceTransformation<?> sourceTransform = (LegacySourceTransformation<?>) leaf; assertBoundedness(Boundedness.CONTINUOUS_UNBOUNDED, sourceTransform); assertThat(sourceTransform.getOperator().emitsProgressiveWatermarks()).isTrue(); @@ -145,10 +149,58 @@ class TransformationsTest { final Table table = env.fromValues(1, 2, 3); - final LegacySourceTransformation<?> sourceTransform = - toLegacySourceTransformation(env, table); + final Transformation<?> leaf = toLeaf(env.toChangelogStream(table).getTransformation()); + assertThat(leaf).isInstanceOf(LegacySourceTransformation.class); + assertBoundedness(Boundedness.BOUNDED, leaf); + } - assertBoundedness(Boundedness.BOUNDED, sourceTransform); + @Test + void testStreamValues() { + final StreamTableEnvironment env = + StreamTableEnvironment.create( + StreamExecutionEnvironment.getExecutionEnvironment(), + EnvironmentSettings.newInstance().inStreamingMode().build()); + + final Table table = env.fromValues(1, 2, 3); + + final Transformation<?> leaf = toLeaf(env.toChangelogStream(table).getTransformation()); + assertThat(leaf).isInstanceOf(SourceTransformation.class); + assertThat(((SourceTransformation<?, ?, ?>) leaf).getSource()) + .isInstanceOf(ValuesSource.class); + assertBoundedness(Boundedness.CONTINUOUS_UNBOUNDED, leaf); + } + + @Test + void testStreamValuesWithoutRows() { + final StreamTableEnvironment env = + StreamTableEnvironment.create( + StreamExecutionEnvironment.getExecutionEnvironment(), + EnvironmentSettings.newInstance().inStreamingMode().build()); + + final String query = "SELECT * FROM (VALUES (1)) AS v(id) WHERE id < 0"; + // The planner reduces the query to a VALUES without rows. + assertThat(env.explainSql(query)).contains("Values(tuples=[[]])"); + + final Table table = env.sqlQuery(query); + + final Transformation<?> leaf = toLeaf(env.toChangelogStream(table).getTransformation()); + assertThat(leaf).isInstanceOf(SourceTransformation.class); + assertThat(((SourceTransformation<?, ?, ?>) leaf).getSource()) + .isInstanceOf(ValuesSource.class); + assertBoundedness(Boundedness.CONTINUOUS_UNBOUNDED, leaf); + } + + /** A plan with version 1 of the VALUES node keeps the legacy source and its job graph. */ + @Test + void testLegacyStreamValuesFromFlink1_15Plan() { + final TableEnvironment env = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); + + final Transformation<?> fromFlink115 = + CompiledPlanUtils.toTransformations(env, planFromFlink1_15Values(env)).get(0); + assertThat(toLeaf(fromFlink115)).isInstanceOf(LegacySourceTransformation.class); + final Transformation<?> fromCurrentFlink = + CompiledPlanUtils.toTransformations(env, planFromCurrentFlinkValues(env)).get(0); + assertThat(toLeaf(fromCurrentFlink)).isInstanceOf(SourceTransformation.class); } @Test @@ -190,7 +242,7 @@ class TransformationsTest { env -> planFromCurrentFlinkValues(env).asJsonString(), "\\d+_stream-exec-sink_2_sink", "\\d+_stream-exec-sink_2_constraint-validator", - "\\d+_stream-exec-values_1_values"); + "\\d+_stream-exec-values_2_values"); } @Test @@ -486,14 +538,13 @@ class TransformationsTest { env.createTemporaryTable("T", sourceDescriptor); } - private static LegacySourceTransformation<?> toLegacySourceTransformation( - StreamTableEnvironment env, Table table) { - Transformation<?> transform = env.toChangelogStream(table).getTransformation(); + /** Follows the inputs of a transformation with a single input down to the source. */ + private static Transformation<?> toLeaf(Transformation<?> transformation) { + Transformation<?> transform = transformation; while (transform.getInputs().size() == 1) { transform = transform.getInputs().get(0); } - assertThat(transform).isInstanceOf(LegacySourceTransformation.class); - return (LegacySourceTransformation<?>) transform; + return transform; } private static void assertBoundedness(Boundedness boundedness, Transformation<?> transform) { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/ValuesStreamingRestoreITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/ValuesStreamingRestoreITCase.java new file mode 100644 index 00000000000..cdbc5547824 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/ValuesStreamingRestoreITCase.java @@ -0,0 +1,204 @@ +/* + * 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.runtime.stream.sql; + +import org.apache.flink.client.program.rest.RestClusterClient; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.CoreOptions; +import org.apache.flink.configuration.ExternalizedCheckpointRetention; +import org.apache.flink.configuration.RestartStrategyOptions; +import org.apache.flink.configuration.StateRecoveryOptions; +import org.apache.flink.core.execution.CheckpointType; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.test.junit5.InjectClusterClient; +import org.apache.flink.test.junit5.InjectMiniCluster; +import org.apache.flink.test.junit5.MiniClusterExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + +import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning; +import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * VALUES waits for one more checkpoint after emitting its rows before it exits (FLIP-147). That + * checkpoint is triggered while the task is still running, so it doesn't mark VALUES as finished. A + * job restored from it must not read the VALUES rows again. + * + * <p>Both tests restore after VALUES emitted all its rows. A restore halfway through the rows is + * only tested by {@code ValuesCodeGeneratorTest#testRestoreEmitsOnlyTheRemainingRows}. + */ +class ValuesStreamingRestoreITCase { + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(4) + .build()); + + @TempDir private Path tmpDir; + + private final List<JobClient> jobs = new ArrayList<>(); + + @BeforeEach + void before() { + TestValuesTableFactory.clearAllData(); + } + + @AfterEach + void after() throws Exception { + // The datagen source never ends, so a job that failed half way would otherwise leak. + for (JobClient job : jobs) { + if (!job.getJobStatus().get().isGloballyTerminalState()) { + cancel(job); + } + } + TestValuesTableFactory.clearAllData(); + } + + /** Checkpoint N, the one VALUES waits for, doesn't mark it as finished. */ + @Test + void testRestoreFromCheckpointTakenBeforeValuesExits( + @InjectClusterClient RestClusterClient<?> client, + @InjectMiniCluster MiniCluster miniCluster) + throws Exception { + assertNoDuplicatesAfterRestore(client, miniCluster, false); + } + + /** + * Checkpoint N+1 is triggered after VALUES exits, so it marks VALUES as finished. The legacy + * source passes this case, too. + */ + @Test + void testRestoreFromCheckpointTakenAfterValuesExits( + @InjectClusterClient RestClusterClient<?> client, + @InjectMiniCluster MiniCluster miniCluster) + throws Exception { + assertNoDuplicatesAfterRestore(client, miniCluster, true); + } + + private void assertNoDuplicatesAfterRestore( + RestClusterClient<?> client, MiniCluster miniCluster, boolean afterValuesExits) + throws Exception { + JobClient job = startJob(null); + waitUntilCondition(() -> rowsOf("[123]").size() == 3); + assertThat(valuesState(client, job)).isEqualTo(ExecutionState.RUNNING); + + // The first checkpoint after VALUES emitted its rows. VALUES exits once it completes. The + // periodic checkpoints are an hour apart, so VALUES can't exit before this one is taken. + long checkpointId = client.triggerCheckpoint(job.getJobID(), CheckpointType.FULL).get(); + waitUntilCondition(() -> valuesState(client, job) == ExecutionState.FINISHED); + if (afterValuesExits) { + checkpointId = client.triggerCheckpoint(job.getJobID(), CheckpointType.FULL).get(); + } + cancel(job); + + JobClient restored = + startJob( + tmpDir.resolve(job.getJobID().toString()) + .resolve("chk-" + checkpointId) + .toUri() + .toString()); + // VALUES waits for one more checkpoint before it exits. + waitForAllTaskRunning(miniCluster, restored.getJobID(), true); + client.triggerCheckpoint(restored.getJobID(), CheckpointType.FULL).get(); + waitUntilCondition(() -> valuesState(client, restored) == ExecutionState.FINISHED); + cancel(restored); + + // A count of 2 means the restored job read the VALUES rows again. + assertThat(rowsOf("[123]")).containsExactlyInAnyOrder("+I[1, 1]", "+I[2, 1]", "+I[3, 1]"); + } + + private JobClient startJob(String checkpoint) { + Configuration conf = new Configuration(); + conf.set(CoreOptions.DEFAULT_PARALLELISM, 2); + conf.set(RestartStrategyOptions.RESTART_STRATEGY, "none"); + // Checkpoints are triggered by the test only. The scheduler may trigger its first + // checkpoint after the min pause instead of the interval, so both are long. + conf.set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofHours(1)); + conf.set(CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS, Duration.ofHours(1)); + conf.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, tmpDir.toUri().toString()); + conf.set( + CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, + ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION); + if (checkpoint != null) { + conf.set(StateRecoveryOptions.SAVEPOINT_PATH, checkpoint); + } + TableEnvironment env = + TableEnvironment.create( + EnvironmentSettings.newInstance().withConfiguration(conf).build()); + env.executeSql( + "CREATE TABLE gen (id INT) WITH ('connector' = 'datagen', 'rows-per-second' = '5')"); + env.executeSql( + "CREATE TABLE snk (id INT, cnt BIGINT, PRIMARY KEY (id) NOT ENFORCED)" + + " WITH ('connector' = 'values', 'sink-insert-only' = 'false')"); + JobClient job = + env.executeSql( + "INSERT INTO snk SELECT id, COUNT(*) FROM (" + + " SELECT id FROM (VALUES (1), (2), (3)) AS v(id)" + + " UNION ALL SELECT 1000 AS id FROM gen" + + ") GROUP BY id") + .getJobClient() + .orElseThrow(); + jobs.add(job); + return job; + } + + /** The state of the VALUES vertex, or null before the job has its vertices. */ + private static ExecutionState valuesState(RestClusterClient<?> client, JobClient job) + throws Exception { + return client.getJobDetails(job.getJobID()).get().getJobVertexInfos().stream() + .filter(vertex -> vertex.getName().startsWith("Source: Values")) + .findFirst() + .map(JobDetailsInfo.JobVertexDetailsInfo::getExecutionState) + .orElse(null); + } + + /** Raw changelog rows of the sink whose key matches the given pattern. */ + private static List<String> rowsOf(String keyPattern) { + return TestValuesTableFactory.getRawResultsAsStrings("snk").stream() + .filter(row -> row.matches("[+-][IUD]\\[" + keyPattern + ", .*")) + .collect(Collectors.toList()); + } + + private static void cancel(JobClient job) throws Exception { + job.cancel().get(); + waitUntilCondition(() -> job.getJobStatus().get().isGloballyTerminalState()); + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-values_2/values-test/plan/values-test.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-values_2/values-test/plan/values-test.json new file mode 100644 index 00000000000..475c8f8d56e --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-values_2/values-test/plan/values-test.json @@ -0,0 +1,82 @@ +{ + "flinkVersion" : "2.5", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-values_2", + "tuples" : [ [ { + "kind" : "LITERAL", + "value" : 1, + "type" : "INT NOT NULL" + }, { + "kind" : "LITERAL", + "value" : 2, + "type" : "INT NOT NULL" + }, { + "kind" : "LITERAL", + "value" : "Hi", + "type" : "CHAR(2) NOT NULL" + } ], [ { + "kind" : "LITERAL", + "value" : 3, + "type" : "INT NOT NULL" + }, { + "kind" : "LITERAL", + "value" : 4, + "type" : "INT NOT NULL" + }, { + "kind" : "LITERAL", + "value" : "Hello", + "type" : "CHAR(5) NOT NULL" + } ] ], + "outputType" : "ROW<`EXPR$0` INT NOT NULL, `EXPR$1` INT NOT NULL, `EXPR$2` VARCHAR(5) NOT NULL>", + "description" : "Values(tuples=[[{ 1, 2, _UTF-16LE'Hi' }, { 3, 4, _UTF-16LE'Hello' }]])" + }, { + "id" : 2, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`sink_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "b", + "dataType" : "INT" + }, { + "name" : "a", + "dataType" : "INT" + }, { + "name" : "c", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "MAP", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` INT NOT NULL, `EXPR$1` INT NOT NULL, `EXPR$2` VARCHAR(5) NOT NULL>", + "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[EXPR$0, EXPR$1, EXPR$2])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file
