[BugFix] Fix exception that java.lang.String is not a valid external type for DATE/TIMESTAMP type (#111)

Signed-off-by: PengFei Li <[email protected]>

Author: banmoy

src/main/java/com/starrocks/connector/spark/rest/RestService.java

@@ -289,7 +289,7 @@ public static List<PartitionDefinition> findPartitions(Settings cfg, Logger logg
         if (!StringUtils.isEmpty(cfg.getProperty(STARROCKS_FILTER_QUERY))) {
             sql += " where " + cfg.getProperty(STARROCKS_FILTER_QUERY);
         }
-        logger.debug("Query SQL Sending to StarRocks FE is: '{}'.", sql);
+        logger.info("Query SQL Sending to StarRocks FE is: '{}'.", sql);
 
         HttpPost httpPost = new HttpPost(getUriStr(cfg, logger) + QUERY_PLAN);
         String entity = "{\"sql\": \"" + sql + "\"}";

src/main/java/com/starrocks/connector/spark/serialization/RowBatch.java

@@ -168,15 +168,16 @@ public void convertArrowToRowBatch() throws Exception {
             for (int col = 0; col < fieldVectors.size(); col++) {
                 FieldVector curFieldVector = fieldVectors.get(col);
                 Types.MinorType mt = curFieldVector.getMinorType();
-                Field field = fieldMap.get(curFieldVector.getName());
-
-                String currentType;
-
-                if (field != null && field.getType().isPresent()) {
-                    currentType = field.getType().get();
-                } else {
-                    currentType = DataTypeUtils.map(mt);
+                String vectorName = curFieldVector.getName();
+                Field field = schema.get(col);
+                Preconditions.checkNotNull(field,
+                        "Can't find schema for arrow vector [%s] at index [%s]", vectorName, col);
+                if (!vectorName.isEmpty()) {
+                    Preconditions.checkState(vectorName.equals(field.getName()),
+                            "The column at [%s] has inconsistent column names between schema [%s] " +
+                                    "and arrow vector [%s]", col, field.getName(), vectorName);
                 }
+                String currentType = field.getType().orElseGet(() -> DataTypeUtils.map(mt));
 
                 switch (currentType) {
                     case "NULL_TYPE":

src/test/java/com/starrocks/connector/spark/sql/ReadWriteITTest.java

@@ -1107,4 +1107,60 @@ public void testJsonLz4Compression() throws Exception {
 
         spark.stop();
     }
+
+    @Test
+    public void testArrowVectorNameIsEmptyForTimestampType() throws Exception {
+        String tableName = "testArrowVectorNameIsEmptyForTimestampType_" + genRandomUuid();
+        prepareDateAndDateTimeTable(tableName);
+        executeSrSQL(
+                String.format("INSERT INTO `%s`.`%s` VALUES (1, '1', '2024-04-20', '2024-04-20 19:00:00')",
+                    DB_NAME, tableName));
+
+        SparkSession spark = SparkSession
+                .builder()
+                .master("local[1]")
+                .appName("testArrowVectorNameIsEmptyForTimestampType")
+                .getOrCreate();
+
+        String ddl = String.format("CREATE TABLE src \n" +
+                "USING starrocks\n" +
+                "OPTIONS(\n" +
+                "  \"starrocks.table.identifier\"=\"%s\",\n" +
+                "  \"starrocks.fe.http.url\"=\"%s\",\n" +
+                "  \"starrocks.fe.jdbc.url\"=\"%s\",\n" +
+                "  \"starrocks.user\"=\"%s\",\n" +
+                "  \"starrocks.password\"=\"%s\"\n" +
+                ")", String.join(".", DB_NAME, tableName), FE_HTTP, FE_JDBC, USER, PASSWORD);
+        spark.sql(ddl);
+        List<Row> readRows = spark.sql("SELECT c1, c2, c3 FROM src WHERE c0 = 1;").collectAsList();
+
+        List<List<Object>> expectedData = Collections.singletonList(
+                Arrays.asList(
+                        "1",
+                        Date.valueOf("2024-04-20"),
+                        Timestamp.valueOf("2024-04-20 19:00:00")
+                )
+        );
+        verifyRows(expectedData, readRows);
+
+        spark.stop();
+    }
+
+    private void prepareDateAndDateTimeTable(String tableName) throws Exception {
+        String createStarRocksTable =
+                String.format("CREATE TABLE `%s`.`%s` (" +
+                                "c0 INT," +
+                                "c1 STRING," +
+                                "c2 DATE," +
+                                "c3 DATETIME" +
+                                ") ENGINE=OLAP " +
+                                "PRIMARY KEY(`c0`) " +
+                                "DISTRIBUTED BY HASH(`c0`) BUCKETS 2 " +
+                                "PROPERTIES (" +
+                                "\"replication_num\" = \"1\"" +
+                                ")",
+                        DB_NAME, tableName);
+        executeSrSQL(createStarRocksTable);
+    }
+
 }