[FLINK-38773][table] Fix batch vector search excnode context (#27311)

Author: lihaosky

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

@@ -28,7 +28,6 @@
 import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecVectorSearchTableFunction;
 import org.apache.flink.table.planner.plan.nodes.exec.spec.VectorSearchSpec;
 import org.apache.flink.table.planner.plan.nodes.exec.spec.VectorSearchTableSourceSpec;
-import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecVectorSearchTableFunction;
 import org.apache.flink.table.planner.plan.utils.FunctionCallUtil;
 import org.apache.flink.table.types.logical.RowType;
 
@@ -65,9 +64,9 @@ public BatchExecVectorSearchTableFunction(
             String description) {
         this(
                 ExecNodeContext.newNodeId(),
-                ExecNodeContext.newContext(StreamExecVectorSearchTableFunction.class),
+                ExecNodeContext.newContext(BatchExecVectorSearchTableFunction.class),
                 ExecNodeContext.newPersistedConfig(
-                        StreamExecVectorSearchTableFunction.class, tableConfig),
+                        BatchExecVectorSearchTableFunction.class, tableConfig),
                 tableSourceSpec,
                 vectorSearchSpec,
                 asyncOptions,

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java

@@ -47,6 +47,7 @@
 import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecTableSourceScan;
 import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecUnion;
 import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecValues;
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecVectorSearchTableFunction;
 import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecWindowTableFunction;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCalc;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCorrelate;
@@ -201,6 +202,7 @@ private ExecNodeMetadataUtil() {
                     add(BatchExecMatch.class);
                     add(BatchExecOverAggregate.class);
                     add(BatchExecRank.class);
+                    add(BatchExecVectorSearchTableFunction.class);
                 }
             };
 

flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-vector-search-table-function_1/async-vector-search/plan/async-vector-search.json

@@ -1,5 +1,5 @@
 {
-  "flinkVersion" : "2.2",
+  "flinkVersion" : "2.3",
   "nodes" : [ {
     "id" : 5,
     "type" : "batch-exec-table-source-scan_1",
@@ -24,10 +24,10 @@
     },
     "outputType" : "ROW<`id` BIGINT, `content` VARCHAR(2147483647), `vector` ARRAY<FLOAT>>",
     "description" : "TableSourceScan(table=[[default_catalog, default_database, src_t]], fields=[id, content, vector])",
-    "dynamicFilteringDataListenerID" : "581f0987-928f-4c00-a1c1-500fdd1c98fb"
+    "dynamicFilteringDataListenerID" : "7068b1b2-f209-4e4b-8f16-77adc1440cb8"
   }, {
     "id" : 6,
-    "type" : "stream-exec-vector-search-table-function_1",
+    "type" : "batch-exec-vector-search-table-function_1",
     "configuration" : {
       "table.exec.async-vector-search.max-concurrent-operations" : "10",
       "table.exec.async-vector-search.output-mode" : "ORDERED",

flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-vector-search-table-function_1/sync-vector-search/plan/sync-vector-search.json

@@ -1,5 +1,5 @@
 {
-  "flinkVersion" : "2.2",
+  "flinkVersion" : "2.3",
   "nodes" : [ {
     "id" : 1,
     "type" : "batch-exec-table-source-scan_1",
@@ -24,10 +24,10 @@
     },
     "outputType" : "ROW<`id` BIGINT, `content` VARCHAR(2147483647), `vector` ARRAY<FLOAT>>",
     "description" : "TableSourceScan(table=[[default_catalog, default_database, src_t]], fields=[id, content, vector])",
-    "dynamicFilteringDataListenerID" : "5e850a71-a459-4906-9822-22c324f6f4c9"
+    "dynamicFilteringDataListenerID" : "19f6b4e7-e167-4785-8b10-4ab9bb286201"
   }, {
     "id" : 2,
-    "type" : "stream-exec-vector-search-table-function_1",
+    "type" : "batch-exec-vector-search-table-function_1",
     "configuration" : {
       "table.exec.async-vector-search.max-concurrent-operations" : "10",
       "table.exec.async-vector-search.output-mode" : "ORDERED",

flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-vector-search-table-function_1/vector-search-with-runtime-config/plan/vector-search-with-runtime-config.json

@@ -1,5 +1,5 @@
 {
-  "flinkVersion" : "2.2",
+  "flinkVersion" : "2.3",
   "nodes" : [ {
     "id" : 9,
     "type" : "batch-exec-table-source-scan_1",
@@ -24,10 +24,10 @@
     },
     "outputType" : "ROW<`id` BIGINT, `content` VARCHAR(2147483647), `vector` ARRAY<FLOAT>>",
     "description" : "TableSourceScan(table=[[default_catalog, default_database, src_t]], fields=[id, content, vector])",
-    "dynamicFilteringDataListenerID" : "21fdf54d-fce6-42da-bcd3-a522ebee5cd6"
+    "dynamicFilteringDataListenerID" : "733d0e86-4dba-4d5e-986a-22f1d385c6ef"
   }, {
     "id" : 10,
-    "type" : "stream-exec-vector-search-table-function_1",
+    "type" : "batch-exec-vector-search-table-function_1",
     "configuration" : {
       "table.exec.async-vector-search.max-concurrent-operations" : "10",
       "table.exec.async-vector-search.output-mode" : "ORDERED",