[ISSUE-176] bug fix for jdbc sql insert (#506)

* fix: sql insert pattern bug fix

* fix: sql insert pattern bug fix

Author: dukewy123

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-jdbc/src/main/java/com/antgroup/geaflow/dsl/connector/jdbc/util/JDBCUtils.java

@@ -19,7 +19,10 @@
 
 package com.antgroup.geaflow.dsl.connector.jdbc.util;
 
+import com.antgroup.geaflow.common.type.IType;
 import com.antgroup.geaflow.common.type.Types;
+import com.antgroup.geaflow.common.type.primitive.BinaryStringType;
+import com.antgroup.geaflow.common.type.primitive.StringType;
 import com.antgroup.geaflow.dsl.common.data.Row;
 import com.antgroup.geaflow.dsl.common.data.impl.ObjectRow;
 import com.antgroup.geaflow.dsl.common.exception.GeaFlowDSLException;
@@ -29,7 +32,6 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -72,14 +74,35 @@ public static void createTemporaryTable(Statement statement, String tableName,
     public static void insertIntoTable(Statement statement, String tableName,
                                        List<TableField> fields, Row row) throws SQLException {
         Object[] values = new Object[fields.size()];
+        boolean isFirst = true;
+        StringBuilder builder = new StringBuilder();
         for (int i = 0; i < fields.size(); i++) {
+            if (isFirst) {
+                isFirst = false;
+            } else {
+                builder.append(",");
+            }
+            IType type = fields.get(i).getType();
+            Object value = row.getField(i, type);
+            if (value == null) {
+                if (fields.get(i).isNullable()) {
+                    builder.append("null");
+                } else {
+                    throw new RuntimeException("filed " + fields.get(i).getName() + " can not be null");
+                }
+            } else if (type.getClass() == BinaryStringType.class || type.getClass() == StringType.class) {
+                builder.append("'").append(value).append("'");
+            } else {
+                builder.append(value);
+            }
             values[i] = row.getField(i, fields.get(i).getType());
         }
-        String insertIntoValues = StringUtils.join(
-            Arrays.stream(values).map(value -> "'" + value + "'").collect(Collectors.toList()),
-            ",");
-        String insertIntoTableQuery = String.format("INSERT INTO %s VALUES (%s);", tableName,
-            insertIntoValues);
+
+        String insertIntoValues = builder.toString();
+        String insertColumns = StringUtils.join(fields.stream().map(
+            field -> field.getName()).collect(Collectors.toList()), ",");
+        String insertIntoTableQuery = String.format("INSERT INTO %s (%s) VALUES (%s);", tableName, insertColumns,
+                insertIntoValues);
         statement.execute(insertIntoTableQuery);
     }
 

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-jdbc/src/test/java/com/antgroup/geaflow/dsl/connector/jdbc/JDBCTableConnectorTest.java

@@ -83,9 +83,20 @@ public void testCreateTable() throws SQLException {
     public void testInsertIntoTable() throws SQLException {
         List<TableField> tableFieldList = new ArrayList<>();
         tableFieldList.add(new TableField("id", Types.INTEGER, false));
-        tableFieldList.add(new TableField("name", Types.BINARY_STRING, false));
-        Row row = ObjectRow.create(5, "Test5");
+        tableFieldList.add(new TableField("name", Types.BINARY_STRING, true));
+        Row row = ObjectRow.create(new Object[]{5, null});
         JDBCUtils.insertIntoTable(statement, "test_table", tableFieldList, row);
+        List<Row> rowList = JDBCUtils.selectRowsFromTable(statement, "test_table",
+                "", 2, 0, 20, "id");
+        Row resultRow = null;
+        for (Row queryRow : rowList) {
+            if ((Integer) queryRow.getField(0, Types.INTEGER) == 5) {
+                resultRow = queryRow;
+                break;
+            }
+        }
+        assert resultRow != null;
+        assert resultRow.getField(1, Types.BINARY_STRING) == null;
     }
 
     @Test

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/jdbc_write_001.sql

@@ -19,7 +19,7 @@
 
 CREATE TABLE tbl_result (
 	user_name varchar,
-	user_count bigint
+	count bigint
 ) WITH (
 	type='jdbc',
 	geaflow.dsl.jdbc.driver = 'org.h2.Driver',