[FLINK-38738][table] Fix UIDs for ScanTableSource with multiple transformations

This closes #27294.

Author: twalthr

flink-python/pyflink/table/tests/jsonplan/testGetJsonPlan.out

@@ -2,7 +2,7 @@
   "flinkVersion" : "",
   "nodes" : [ {
     "id" : 0,
-    "type" : "stream-exec-table-source-scan_1",
+    "type" : "stream-exec-table-source-scan_2",
     "scanTableSource" : {
       "table" : {
         "identifier" : "`default_catalog`.`default_database`.`MyTable`",

flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ProviderContext.java

@@ -34,11 +34,72 @@ public interface ProviderContext {
 
     /**
      * Generates a new unique identifier for a {@link Transformation}/{@code DataStream} operator.
-     * The {@code name} must be unique within the provider implementation. The framework will make
-     * sure that the name is unique for the entire topology.
+     *
+     * <p>UIDs are crucial for state management as they identify an operator in the topology. The
+     * planner guarantees that all operators receive a stable UID. However, the planner does not
+     * control the operators and transformations in sources and sinks. From a planner’s perspective,
+     * sources and sinks are black boxes. Implementers must ensure that UIDs are assigned to all
+     * operators.
+     *
+     * <p>The {@param name} argument must be unique within the provider implementation. The
+     * framework will make sure that the name is unique for the entire topology.
+     *
+     * <p>For example, a connector that consists of two transformations (a source and a validator)
+     * must use this method in the following way:
+     *
+     * <pre>{@code
+     * var sourceOperator = ...
+     * providerContext.generateUid("source").ifPresent(sourceOperator::uid);
+     *
+     * var validatorOperator = sourceOperator.process(...);
+     * providerContext.generateUid("validator").ifPresent(validatorOperator::uid);
+     * }</pre>
      *
      * <p>This method returns empty if an identifier cannot be generated, i.e., because the job is
-     * in batch mode, or UIDs cannot be guaranteed to be unique.
+     * in batch mode, or UIDs cannot be guaranteed to be unique because the topology is not created
+     * from a compiled plan.
      */
     Optional<String> generateUid(String name);
+
+    /**
+     * Returns the display name provided by the framework to label the connector.
+     *
+     * <p>If multiple transformations are present, the implementer can decide which one is the
+     * connector. The name will be shown in UI and logs.
+     *
+     * <p>For example, a connector that consists of two transformations (a source and a validator)
+     * can use this method in the following way:
+     *
+     * <pre>{@code
+     * var sourceOperator = ...
+     * sourceOperator.name(providerContext.getName());
+     *
+     * var validatorOperator = sourceOperator.process(...);
+     * validatorOperator.name("Validator for " + providerContext.getName());
+     * }</pre>
+     */
+    default String getName() {
+        return "";
+    }
+
+    /**
+     * Returns the description provided by the framework to label the connector.
+     *
+     * <p>If multiple transformations are present, the implementer can decide which one is the
+     * connector. The description will be shown in UI and logs.
+     *
+     * <p>For example, a connector that consists of two transformations (a source and a validator)
+     * can use this method in the following way:
+     *
+     * <pre>{@code
+     * var sourceOperator = ...
+     * sourceOperator.setDescription(providerContext.getDescription());
+     *
+     * var validatorOperator = sourceOperator.process(...);
+     * validatorOperator.setDescription("Validates non-null values.");
+     * }</pre>
+     */
+    default String getDescription() {
+        return "";
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNode.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.planner.plan.nodes.exec;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitor;
 import org.apache.flink.table.planner.plan.nodes.physical.FlinkPhysicalRel;
@@ -38,6 +39,9 @@
 /**
  * The representation of execution information for a {@link FlinkPhysicalRel}.
  *
+ * <p>Note: Every new node should be annotated with {@link ExecNodeMetadata}. It enables bookkeeping
+ * and affects how a {@link CompiledPlan} is derived from Java objects.
+ *
  * @param <T> The type of the elements that result from this node.
  */
 @JsonTypeInfo(
@@ -61,6 +65,16 @@ public interface ExecNode<T> extends ExecNodeTranslator<T>, FusionCodegenExecNod
     @JsonProperty(value = FIELD_NAME_ID, index = 0)
     int getId();
 
+    /**
+     * The version of the node.
+     *
+     * <p>A new version can be added by declaring a {@link ExecNodeMetadata} annotation, potentially
+     * by copying the old annotation. You can use this method to get the current compiled version
+     * and execute version-specific logic accordingly.
+     */
+    @JsonIgnore
+    int getVersion();
+
     /** Returns a string which describes this node. */
     @JsonProperty(value = FIELD_NAME_DESCRIPTION)
     String getDescription();

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeBase.java

@@ -26,6 +26,7 @@
 import org.apache.flink.streaming.runtime.partitioner.GlobalPartitioner;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.connector.ProviderContext;
 import org.apache.flink.table.delegation.Planner;
 import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
 import org.apache.flink.table.planner.delegation.PlannerBase;
@@ -48,6 +49,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -129,6 +131,11 @@ public final int getId() {
         return context.getId();
     }
 
+    @Override
+    public final int getVersion() {
+        return context.getVersion();
+    }
+
     @Override
     public String getDescription() {
         return description;
@@ -325,4 +332,28 @@ protected OpFusionCodegenSpecGenerator translateToFusionCodegenSpecInternal(
             PlannerBase planner, ExecNodeConfig config, CodeGeneratorContext parentCtx) {
         throw new TableException("This ExecNode doesn't support operator fusion codegen now.");
     }
+
+    /** Context for connectors to configure transformations and operators. */
+    protected ProviderContext createProviderContext(
+            TransformationMetadata metadata, ExecNodeConfig config) {
+        return new ProviderContext() {
+            @Override
+            public Optional<String> generateUid(String name) {
+                if (config.shouldSetUid()) {
+                    return Optional.of(createTransformationUid(name, config));
+                }
+                return Optional.empty();
+            }
+
+            @Override
+            public String getName() {
+                return metadata.getName();
+            }
+
+            @Override
+            public String getDescription() {
+                return metadata.getDescription();
+            }
+        };
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeMetadata.java

@@ -21,6 +21,7 @@
 import org.apache.flink.FlinkVersion;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode;
 
@@ -32,18 +33,18 @@
 import java.lang.annotation.Target;
 
 /**
- * Annotation to be used for {@link ExecNode}s to keep necessary metadata when
- * serializing/deserializing them in a plan. It's used for internal bookkeeping across Flink
- * versions, and to provide necessary information to the testing infrastructure.
+ * Annotation to be used by classes implementing the {@link ExecNode} interface.
  *
- * <p>Each {@link ExecNode} needs to be annotated and provide the necessary metadata info so that it
- * can be correctly serialized and later on instantiated from a string (JSON) plan.
+ * <p>The {@link ExecNodeMetadata} annotation contains crucial information for (de)serializing a
+ * node into a {@link CompiledPlan}. It's used for internal bookkeeping across Flink versions and
+ * powers testing infrastructure.
  *
- * <p>It's possible for one {@link ExecNode} class to use multiple annotations to denote ability to
- * upgrade to more versions. an {@link ExecNode} class can be annotated directly with multiple
- * {@link ExecNodeMetadata} annotations, or with a single {@link MultipleExecNodeMetadata}
- * annotation where the {@link MultipleExecNodeMetadata#value()} is an array of {@link
- * ExecNodeMetadata} annotations.
+ * <p>Each new {@link ExecNode} version needs an annotation. Multiple annotations on top of a single
+ * {@link ExecNode} class are supported and declare multiple versions. When compiling a node into
+ * JSON, the most recent version is persisted. When restoring from JSON, the original version is
+ * available via {@link ExecNode#getVersion()}. Parts of the original configuration (see {@link
+ * #consumedOptions()}) together with recent session configuration will be made available via {@link
+ * ExecNodeConfig}.
  */
 @Documented
 @Target(ElementType.TYPE)

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecTableSourceScan.java

@@ -124,7 +124,7 @@ public String getDynamicFilteringDataListenerID() {
     protected Transformation<RowData> translateToPlanInternal(
             PlannerBase planner, ExecNodeConfig config) {
         final Transformation<RowData> transformation =
-                super.translateToPlanInternal(planner, config);
+                super.createTransformation(planner, config, false);
         // the boundedness has been checked via the runtime provider already, so we can safely
         // declare all legacy transformations as bounded to make the stream graph generator happy
         ExecNodeUtil.makeLegacySourceTransformationsBounded(transformation);

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java

@@ -398,19 +398,21 @@ private Transformation<?> applySinkProvider(
             ExecNodeConfig config,
             ClassLoader classLoader) {
         try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
-
-            TransformationMetadata sinkMeta = createTransformationMeta(SINK_TRANSFORMATION, config);
+            final TransformationMetadata metadata =
+                    createTransformationMeta(SINK_TRANSFORMATION, config);
             if (runtimeProvider instanceof DataStreamSinkProvider) {
                 Transformation<RowData> sinkTransformation =
                         applyRowtimeTransformation(
                                 inputTransform, rowtimeFieldIndex, sinkParallelism, config);
                 final DataStream<RowData> dataStream = new DataStream<>(env, sinkTransformation);
                 final DataStreamSinkProvider provider = (DataStreamSinkProvider) runtimeProvider;
-                return provider.consumeDataStream(createProviderContext(config), dataStream)
+                return provider.consumeDataStream(
+                                createProviderContext(metadata, config), dataStream)
                         .getTransformation();
             } else if (runtimeProvider instanceof TransformationSinkProvider) {
                 final TransformationSinkProvider provider =
                         (TransformationSinkProvider) runtimeProvider;
+                final ProviderContext providerContext = createProviderContext(metadata, config);
                 return provider.createTransformation(
                         new TransformationSinkProvider.Context() {
                             @Override
@@ -425,7 +427,17 @@ public int getRowtimeIndex() {
 
                             @Override
                             public Optional<String> generateUid(String name) {
-                                return createProviderContext(config).generateUid(name);
+                                return providerContext.generateUid(name);
+                            }
+
+                            @Override
+                            public String getName() {
+                                return providerContext.getName();
+                            }
+
+                            @Override
+                            public String getDescription() {
+                                return providerContext.getDescription();
                             }
                         });
             } else if (runtimeProvider instanceof SinkFunctionProvider) {
@@ -436,7 +448,7 @@ public Optional<String> generateUid(String name) {
                         env,
                         inputTransform,
                         rowtimeFieldIndex,
-                        sinkMeta,
+                        metadata,
                         sinkParallelism);
             } else if (runtimeProvider instanceof OutputFormatProvider) {
                 OutputFormat<RowData> outputFormat =
@@ -448,7 +460,7 @@ public Optional<String> generateUid(String name) {
                         env,
                         inputTransform,
                         rowtimeFieldIndex,
-                        sinkMeta,
+                        metadata,
                         sinkParallelism);
             } else if (runtimeProvider instanceof SinkV2Provider) {
                 SinkV2Provider sinkV2Provider = (SinkV2Provider) runtimeProvider;
@@ -466,23 +478,14 @@ public Optional<String> generateUid(String name) {
                         .getAdditionalMetricVariables()
                         .forEach(transformation::addMetricVariable);
                 transformation.setParallelism(sinkParallelism, sinkParallelismConfigured);
-                sinkMeta.fill(transformation);
+                metadata.fill(transformation);
                 return transformation;
             } else {
                 throw new TableException("Unsupported sink runtime provider.");
             }
         }
     }
 
-    private ProviderContext createProviderContext(ExecNodeConfig config) {
-        return name -> {
-            if (config.shouldSetUid()) {
-                return Optional.of(createTransformationUid(name, config));
-            }
-            return Optional.empty();
-        };
-    }
-
     private Transformation<?> createSinkFunctionTransformation(
             SinkFunction<RowData> sinkFunction,
             StreamExecutionEnvironment env,