[fix-32875][core]query field size bigger than sink table field need support
package com.dtstack.flink.sql.exec;
import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlInsert;
import org.apache.flink.sql.parser.dml.RichSqlInsert;
import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.api.internal.TableImpl;
import org.apache.flink.table.api.java.StreamTableEnvironment;
@@ -42,6 +40,7 @@ import scala.Tuple2;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.util.Arrays;
/**
@@ -60,38 +59,20 @@ public class FlinkSQLExec {
RichSqlInsert insert = (RichSqlInsert) flinkPlanner.validate(flinkPlanner.parser().parse(stmt));
TableImpl queryResult = extractQueryTableFromInsertCaluse(tableEnvImpl, flinkPlanner, insert);
- String targetTableName = ((SqlIdentifier) ((SqlInsert) insert).getTargetTable()).names.get(0);
+ String targetTableName = ((SqlIdentifier) insert.getTargetTable()).names.get(0);
TableSink tableSink = getTableSinkByPlanner(streamPlanner, targetTableName);
String[] sinkFieldNames = tableSink.getTableSchema().getFieldNames();
String[] queryFieldNames = queryResult.getSchema().getFieldNames();
-
- if (sinkFieldNames.length != queryFieldNames.length) {
- throw new ValidationException(
- "Field name of query result and registered TableSink " + targetTableName + " do not match.\n" +
- "Query result schema: " + String.join(",", queryFieldNames) + "\n" +
- "TableSink schema: " + String.join(",", sinkFieldNames));
- }
-
-
- Table newTable = null;
+ Table newTable;
try {
newTable = queryResult.select(String.join(",", sinkFieldNames));
- } catch (Exception e) {
- throw new ValidationException(
- "Field name of query result and registered TableSink "+targetTableName +" do not match.\n" +
- "Query result schema: " + String.join(",", queryFieldNames) + "\n" +
- "TableSink schema: " + String.join(",", sinkFieldNames));
- }
-
- try {
tableEnv.insertInto(targetTableName, newTable);
} catch (Exception e) {
- LOG.warn("Field name case of query result and registered TableSink do not match. ", e);
+ LOG.warn(String.format("Query result and registered TableSink do not match \n input field list:%s \n output field list:%s ", Arrays.toString(queryFieldNames), Arrays.toString(sinkFieldNames)));
newTable = queryResult.select(String.join(",", ignoreCase(queryFieldNames, sinkFieldNames)));
tableEnv.insertInto(targetTableName, newTable);
}
-
}
private static TableSink getTableSinkByPlanner(StreamPlanner streamPlanner, String targetTableName)
