[fix-33847][core] fix other sink except kafka table schema not found problem.

Author: a49a

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -332,9 +332,15 @@ public static Set<URL> registerTable(SqlTree sqlTree,
                 URL sourceTablePathUrl = PluginUtil.buildSourceAndSinkPathByLoadMode(tableInfo.getType(), AbstractSourceTableInfo.SOURCE_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
                 pluginClassPathSets.add(sourceTablePathUrl);
             } else if (tableInfo instanceof AbstractTargetTableInfo) {
-
                 TableSink tableSink = StreamSinkFactory.getTableSink((AbstractTargetTableInfo) tableInfo, localSqlPluginPath, pluginLoadMode);
-                tableEnv.registerTableSink(tableInfo.getName(), tableSink);
+                // TODO Kafka Sink直接注册，其他的Sink要修复才可以。
+                if (tableInfo.getType().startsWith("kafka")) {
+                    tableEnv.registerTableSink(tableInfo.getName(), tableSink);
+                } else {
+                    TypeInformation[] flinkTypes = FunctionManager.transformTypes(tableInfo.getFieldClasses());
+                    tableEnv.registerTableSink(tableInfo.getName(), tableInfo.getFields(), flinkTypes, tableSink);
+                }
+
                 URL sinkTablePathUrl = PluginUtil.buildSourceAndSinkPathByLoadMode(tableInfo.getType(), AbstractTargetTableInfo.TARGET_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
                 pluginClassPathSets.add(sinkTablePathUrl);
             } else if (tableInfo instanceof AbstractSideTableInfo) {

core/src/main/java/com/dtstack/flink/sql/function/FunctionManager.java

@@ -124,4 +124,12 @@ public static void registerAggregateUDF(String classPath, String funcName, Table
         }
     }
 
+    public static TypeInformation[] transformTypes(Class[] fieldTypes) {
+        TypeInformation[] types = new TypeInformation[fieldTypes.length];
+        for (int i = 0; i < fieldTypes.length; i++) {
+            types[i] = TypeInformation.of(fieldTypes[i]);
+        }
+        return types;
+    }
+
 }