[FLINK-29222][hive] Fix wrong behavior for Hive's load data inpath

This closes apache#20778
huanhaihu · Sep 20, 2022 · 4448d9f · 4448d9f
1 parent 791d839
commit 4448d9f
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 7 deletions.
diff --git a/...e/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java b/...e/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
@@ -153,14 +153,14 @@ private Optional<TableResultInternal> executeHiveLoadDataOperation(
                         hiveLoadDataOperation.getPath(),
                         hiveLoadDataOperation.getTablePath(),
                         hiveLoadDataOperation.getPartitionSpec(),
-                        hiveLoadDataOperation.isSrcLocal(),
-                        hiveLoadDataOperation.isOverwrite());
+                        hiveLoadDataOperation.isOverwrite(),
+                        hiveLoadDataOperation.isSrcLocal());
             } else {
                 hiveCatalog.loadTable(
                         hiveLoadDataOperation.getPath(),
                         hiveLoadDataOperation.getTablePath(),
-                        hiveLoadDataOperation.isSrcLocal(),
-                        hiveLoadDataOperation.isOverwrite());
+                        hiveLoadDataOperation.isOverwrite(),
+                        hiveLoadDataOperation.isSrcLocal());
             }
             return Optional.of(TableResultImpl.TABLE_RESULT_OK);
         } finally {

diff --git a/...connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java b/...connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java
@@ -715,17 +715,22 @@ public void testLoadData() throws Exception {
                                     .replace("$filepath", testLoadCsvFilePath));
 
             // test load data into table
-            tableEnv.executeSql("insert into tab1 values (1, 1), (1, 2), (2, 1), (2, 2)").await();
+            tableEnv.executeSql("insert into tab1 values (1, 1), (1, 2)").await();
             tableEnv.executeSql(
                     String.format(
                             "load data local inpath '%s' INTO TABLE tab2", warehouse + "/tab1"));
             List<Row> result =
                     CollectionUtil.iteratorToList(
                             tableEnv.executeSql("select * from tab2").collect());
-            assertThat(result.toString()).isEqualTo("[+I[1, 1], +I[1, 2], +I[2, 1], +I[2, 2]]");
+            assertThat(result.toString()).isEqualTo("[+I[1, 1], +I[1, 2]]");
+            // there should still exist data in tab1
+            result =
+                    CollectionUtil.iteratorToList(
+                            tableEnv.executeSql("select * from tab1").collect());
+            assertThat(result.toString()).isEqualTo("[+I[1, 1], +I[1, 2]]");
 
             // test load data overwrite
-            tableEnv.executeSql("insert into tab1 values (2, 1), (2, 2)").await();
+            tableEnv.executeSql("insert overwrite table tab1 values (2, 1), (2, 2)").await();
             tableEnv.executeSql(
                     String.format(
                             "load data local inpath '%s' overwrite into table tab2",
@@ -741,6 +746,8 @@ public void testLoadData() throws Exception {
                                     "load data inpath '%s' into table p_table partition (dateint=2022) ",
                                     testLoadCsvFilePath))
                     .await();
+            // the file should be removed
+            assertThat(new File(testLoadCsvFilePath).exists()).isFalse();
             result =
                     CollectionUtil.iteratorToList(
                             tableEnv.executeSql("select * from p_table where dateint=2022")