[FLINK-38738][table] Expose more context in table connectors to correct UID

This closes #27352.

Author: twalthr

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

@@ -61,6 +61,16 @@ public interface ProviderContext {
      */
     Optional<String> generateUid(String name);
 
+    /**
+     * Returns the framework's node type in which this connector is embedded.
+     *
+     * <p>In other words: It returns the ExecNode's name and version as contained in the compiled
+     * plan. For example, "stream-exec-table-source-scan_2" or "stream-exec-sink_1".
+     */
+    default String getContainerNodeType() {
+        return "";
+    }
+
     /**
      * Returns the display name provided by the framework to label the connector.
      *

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

@@ -65,12 +65,26 @@ public interface ExecNode<T> extends ExecNodeTranslator<T>, FusionCodegenExecNod
     @JsonProperty(value = FIELD_NAME_ID, index = 0)
     int getId();
 
+    /**
+     * The node's type as contained in {@link CompiledPlan} (e.g. "stream-exec-table-source-scan_2"
+     * consisting of name and version).
+     *
+     * <p>A new type including its version can be added by declaring a {@link ExecNodeMetadata}
+     * annotation.
+     *
+     * @see ExecNodeContext#getTypeAsString()
+     */
+    @JsonIgnore
+    String getTypeAsString();
+
     /**
      * 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.
+     *
+     * @see ExecNodeContext#getVersion()
      */
     @JsonIgnore
     int getVersion();

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

@@ -131,6 +131,11 @@ public final int getId() {
         return context.getId();
     }
 
+    @Override
+    public final String getTypeAsString() {
+        return context.getTypeAsString();
+    }
+
     @Override
     public final int getVersion() {
         return context.getVersion();
@@ -345,6 +350,11 @@ public Optional<String> generateUid(String name) {
                 return Optional.empty();
             }
 
+            @Override
+            public String getContainerNodeType() {
+                return getTypeAsString();
+            }
+
             @Override
             public String getName() {
                 return metadata.getName();

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

@@ -431,6 +431,11 @@ public Optional<String> generateUid(String name) {
                                 return providerContext.generateUid(name);
                             }
 
+                            @Override
+                            public String getContainerNodeType() {
+                                return providerContext.getContainerNodeType();
+                            }
+
                             @Override
                             public String getName() {
                                 return providerContext.getName();

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TestingBatchExecNode.java

@@ -66,6 +66,11 @@ public int getId() {
         return 0;
     }
 
+    @Override
+    public String getTypeAsString() {
+        return "";
+    }
+
     @Override
     public int getVersion() {
         return 0;

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java

@@ -413,21 +413,22 @@ private static CompiledPlan planFromFlink1_15Values(TableEnvironment env) {
     }
 
     private static CompiledPlan planFromFlink2_2MultiTransformSource(TableEnvironment env) {
-        createMultiTransformSource(env);
+        createMultiTransformSource(env, "stream-exec-table-source-scan_1");
         // plan content is compiled from
         // planFromCurrentFlinkMultiTransformSource() using Flink release-2.2
         return env.loadPlan(
                 PlanReference.fromResource("/jsonplan/testMultiTransformSourceUidsFlink2_2.out"));
     }
 
     private static CompiledPlan planFromCurrentFlinkMultiTransformSource(TableEnvironment env) {
-        createMultiTransformSource(env);
+        createMultiTransformSource(env, "stream-exec-table-source-scan_2");
         return env.from("T")
                 .insertInto(TableDescriptor.forConnector("blackhole").build())
                 .compilePlan();
     }
 
-    private static void createMultiTransformSource(TableEnvironment env) {
+    private static void createMultiTransformSource(
+            TableEnvironment env, String expectedSourceExecNode) {
         final DataStreamScanProvider scanProvider =
                 new DataStreamScanProvider() {
                     @Override
@@ -438,6 +439,10 @@ public boolean isBounded() {
                     @Override
                     public DataStream<RowData> produceDataStream(
                             ProviderContext providerContext, StreamExecutionEnvironment execEnv) {
+
+                        assertThat(providerContext.getContainerNodeType())
+                                .isEqualTo(expectedSourceExecNode);
+
                         // UID 1
                         final SingleOutputStreamOperator<Integer> ints = execEnv.fromData(1, 2, 3);
                         providerContext.generateUid("my-source").ifPresent(ints::uid);