This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6fe0d1dbfc5 [FLINK-28720][hive][connectors] Add Hive partition when
flink has no data to write (#20394)
6fe0d1dbfc5 is described below
commit 6fe0d1dbfc58b80393972855da9d58c237f29376
Author: zhangmang <[email protected]>
AuthorDate: Thu Aug 4 22:30:04 2022 +0800
[FLINK-28720][hive][connectors] Add Hive partition when flink has no data
to write (#20394)
---
.../connector/file/table/FileSystemCommitter.java | 24 +++---
.../file/table/FileSystemOutputFormat.java | 3 +-
.../connector/file/table/PartitionLoader.java | 26 ++++++
.../file/table/FileSystemCommitterTest.java | 92 +++++++++++++++++++++-
.../flink/connectors/hive/HiveTableSinkITCase.java | 55 +++++++++++++
.../hive/TableEnvHiveConnectorITCase.java | 4 +-
6 files changed, 185 insertions(+), 19 deletions(-)
diff --git
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
index 69ee3c0695b..2af2d71b901 100644
---
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
+++
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
@@ -58,15 +58,7 @@ class FileSystemCommitter implements Serializable {
private final boolean isToLocal;
private final Path tmpPath;
private final int partitionColumnSize;
-
- FileSystemCommitter(
- FileSystemFactory factory,
- TableMetaStoreFactory metaStoreFactory,
- boolean overwrite,
- Path tmpPath,
- int partitionColumnSize) {
- this(factory, metaStoreFactory, overwrite, tmpPath,
partitionColumnSize, false);
- }
+ private final LinkedHashMap<String, String> staticPartitions;
FileSystemCommitter(
FileSystemFactory factory,
@@ -74,13 +66,15 @@ class FileSystemCommitter implements Serializable {
boolean overwrite,
Path tmpPath,
int partitionColumnSize,
- boolean isToLocal) {
+ boolean isToLocal,
+ LinkedHashMap<String, String> staticPartitions) {
this.factory = factory;
this.metaStoreFactory = metaStoreFactory;
this.overwrite = overwrite;
this.tmpPath = tmpPath;
this.partitionColumnSize = partitionColumnSize;
this.isToLocal = isToLocal;
+ this.staticPartitions = staticPartitions;
}
/** For committing job's output after successful batch job completion. */
@@ -91,9 +85,13 @@ class FileSystemCommitter implements Serializable {
try (PartitionLoader loader =
new PartitionLoader(overwrite, fs, metaStoreFactory,
isToLocal)) {
if (partitionColumnSize > 0) {
- for (Map.Entry<LinkedHashMap<String, String>, List<Path>>
entry :
- collectPartSpecToPaths(fs, taskPaths,
partitionColumnSize).entrySet()) {
- loader.loadPartition(entry.getKey(), entry.getValue());
+ if (taskPaths.isEmpty() && !staticPartitions.isEmpty()) {
+ loader.loadEmptyPartition(this.staticPartitions);
+ } else {
+ for (Map.Entry<LinkedHashMap<String, String>, List<Path>>
entry :
+ collectPartSpecToPaths(fs, taskPaths,
partitionColumnSize).entrySet()) {
+ loader.loadPartition(entry.getKey(), entry.getValue());
+ }
}
} else {
loader.loadNonPartition(taskPaths);
diff --git
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
index 006a7b16713..3ec181e7a53 100644
---
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
+++
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
@@ -93,7 +93,8 @@ public class FileSystemOutputFormat<T> implements
OutputFormat<T>, FinalizeOnMas
overwrite,
tmpPath,
partitionColumns.length,
- isToLocal);
+ isToLocal,
+ staticPartitions);
committer.commitPartitions();
} catch (Exception e) {
throw new TableException("Exception in finalizeGlobal", e);
diff --git
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
index a85ec543a63..ae974d24ef8 100644
---
a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
+++
b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
@@ -86,6 +86,32 @@ public class PartitionLoader implements Closeable {
overwriteAndMoveFiles(srcDirs, tableLocation);
}
+ /**
+ * The flink job does not write data to the partition, but the
corresponding partition needs to
+ * be created or updated.
+ *
+ * <p>The partition does not exist, create it.
+ *
+ * <p>The partition exists:
+ *
+ * <pre>
+ * if overwrite is true, delete the path, then create it;
+ * if overwrite is false, do nothing;
+ * </pre>
+ */
+ public void loadEmptyPartition(LinkedHashMap<String, String> partSpec)
throws Exception {
+ Optional<Path> pathFromMeta = metaStore.getPartition(partSpec);
+ if (pathFromMeta.isPresent() && !overwrite) {
+ return;
+ }
+ Path path = new Path(metaStore.getLocationPath(),
generatePartitionPath(partSpec));
+ if (pathFromMeta.isPresent()) {
+ fs.delete(pathFromMeta.get(), true);
+ fs.mkdirs(path);
+ }
+ metaStore.createOrAlterPartition(partSpec, path);
+ }
+
private void overwriteAndMoveFiles(List<Path> srcDirs, Path destDir)
throws Exception {
FileSystem destFileSystem = destDir.getFileSystem();
boolean dirSuccessExist = destFileSystem.exists(destDir) ||
destFileSystem.mkdirs(destDir);
diff --git
a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java
b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java
index 1910bc44ebd..47983f4667f 100644
---
a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java
+++
b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java
@@ -61,7 +61,13 @@ class FileSystemCommitterTest {
void testPartition() throws Exception {
FileSystemCommitter committer =
new FileSystemCommitter(
- fileSystemFactory, metaStoreFactory, true, new
Path(path.toString()), 2);
+ fileSystemFactory,
+ metaStoreFactory,
+ true,
+ new Path(path.toString()),
+ 2,
+ false,
+ new LinkedHashMap<String, String>());
createFile(path, "task-1/p1=0/p2=0/", "f1", "f2");
createFile(path, "task-2/p1=0/p2=0/", "f3");
@@ -81,7 +87,13 @@ class FileSystemCommitterTest {
committer =
new FileSystemCommitter(
- fileSystemFactory, metaStoreFactory, false, new
Path(path.toString()), 2);
+ fileSystemFactory,
+ metaStoreFactory,
+ false,
+ new Path(path.toString()),
+ 2,
+ false,
+ new LinkedHashMap<String, String>());
createFile(path, "task-2/p1=0/p2=1/", "f6");
committer.commitPartitions();
assertThat(new File(outputPath.toFile(), "p1=0/p2=1/f5")).exists();
@@ -92,7 +104,13 @@ class FileSystemCommitterTest {
void testNotPartition() throws Exception {
FileSystemCommitter committer =
new FileSystemCommitter(
- fileSystemFactory, metaStoreFactory, true, new
Path(path.toString()), 0);
+ fileSystemFactory,
+ metaStoreFactory,
+ true,
+ new Path(path.toString()),
+ 0,
+ false,
+ new LinkedHashMap<String, String>());
createFile(path, "task-1/", "f1", "f2");
createFile(path, "task-2/", "f3");
@@ -107,13 +125,79 @@ class FileSystemCommitterTest {
committer =
new FileSystemCommitter(
- fileSystemFactory, metaStoreFactory, false, new
Path(path.toString()), 0);
+ fileSystemFactory,
+ metaStoreFactory,
+ false,
+ new Path(path.toString()),
+ 0,
+ false,
+ new LinkedHashMap<String, String>());
createFile(path, "task-2/", "f5");
committer.commitPartitions();
assertThat(new File(outputPath.toFile(), "f4")).exists();
assertThat(new File(outputPath.toFile(), "f5")).exists();
}
+ @Test
+ void testEmptyPartition() throws Exception {
+ LinkedHashMap<String, String> staticPartitions = new
LinkedHashMap<String, String>();
+ // add new empty partition
+ staticPartitions.put("dt", "2022-08-02");
+ FileSystemCommitter committer =
+ new FileSystemCommitter(
+ fileSystemFactory,
+ metaStoreFactory,
+ true,
+ new Path(path.toString()),
+ 1,
+ false,
+ staticPartitions);
+
+ createFile(path, "task-1/dt=2022-08-02/");
+ createFile(path, "task-2/dt=2022-08-02/");
+
+ committer.commitPartitions();
+
+ File emptyPartitionFile = new File(outputPath.toFile(),
"dt=2022-08-02");
+ assertThat(emptyPartitionFile).exists();
+ assertThat(emptyPartitionFile).isDirectory();
+ assertThat(emptyPartitionFile).isEmptyDirectory();
+
+ // Add new empty partition to overwrite the old one with data
+ createFile(outputPath, "dt=2022-08-02/f1");
+ assertThat(new File(emptyPartitionFile, "f1")).exists();
+
+ createFile(path, "task-1/dt=2022-08-02/");
+ createFile(path, "task-2/dt=2022-08-02/");
+ committer.commitPartitions();
+
+ assertThat(emptyPartitionFile).exists();
+ assertThat(emptyPartitionFile).isDirectory();
+ assertThat(emptyPartitionFile).isEmptyDirectory();
+
+ // Add empty partition to the old one with data
+ createFile(outputPath, "dt=2022-08-02/f1");
+ assertThat(new File(emptyPartitionFile, "f1")).exists();
+
+ createFile(path, "task-1/dt=2022-08-02/");
+ createFile(path, "task-2/dt=2022-08-02/");
+ committer =
+ new FileSystemCommitter(
+ fileSystemFactory,
+ metaStoreFactory,
+ false,
+ new Path(path.toString()),
+ 1,
+ false,
+ staticPartitions);
+ committer.commitPartitions();
+
+ assertThat(emptyPartitionFile).exists();
+ assertThat(emptyPartitionFile).isDirectory();
+ assertThat(emptyPartitionFile).isNotEmptyDirectory();
+ assertThat(new File(emptyPartitionFile, "f1")).exists();
+ }
+
static class TestMetaStoreFactory implements TableMetaStoreFactory {
private final Path outputPath;
diff --git
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
index 26adb22180d..b7c14a3dd78 100644
---
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
+++
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
@@ -441,6 +441,61 @@ public class HiveTableSinkITCase {
testStreamingWriteWithCustomPartitionCommitPolicy(TestCustomCommitPolicy.class.getName());
}
+ @Test
+ public void testWritingNoDataToPartition() throws Exception {
+ TableEnvironment tEnv =
HiveTestUtils.createTableEnvInBatchMode(SqlDialect.HIVE);
+ tEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
+ tEnv.useCatalog(hiveCatalog.getName());
+ tEnv.executeSql("CREATE TABLE src_table (name string) PARTITIONED BY
(`dt` string)");
+ tEnv.executeSql("CREATE TABLE target_table (name string) PARTITIONED
BY (`dt` string)");
+
+ // insert into partition
+ tEnv.executeSql(
+ "INSERT INTO target_table partition (dt='2022-07-27')
SELECT name FROM src_table where dt = '2022-07-27'")
+ .await();
+ List<Row> partitions =
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show partitions
target_table").collect());
+ assertThat(partitions).hasSize(1);
+ assertThat(partitions.toString()).contains("dt=2022-07-27");
+
+ // insert overwrite partition
+ tEnv.executeSql(
+ "INSERT OVERWRITE target_table partition
(dt='2022-07-28') SELECT name FROM src_table where dt = '2022-07-28'")
+ .await();
+ partitions =
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show partitions
target_table").collect());
+ assertThat(partitions).hasSize(2);
+ assertThat(partitions.toString()).contains("dt=2022-07-28");
+
+ // insert into a partition with data
+ tEnv.executeSql("INSERT INTO target_table partition (dt='2022-07-29')
VALUES ('zm')")
+ .await();
+
+ assertBatch("target_table", Arrays.asList("+I[zm, 2022-07-29]"));
+ tEnv.executeSql(
+ "INSERT INTO target_table partition (dt='2022-07-29')
SELECT name FROM src_table where dt = '2022-07-29'")
+ .await();
+ partitions =
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show partitions
target_table").collect());
+ assertThat(partitions).hasSize(3);
+ assertThat(partitions.toString()).contains("dt=2022-07-29");
+ assertBatch("target_table", Arrays.asList("+I[zm, 2022-07-29]"));
+
+ // insert overwrite a partition with data
+ tEnv.executeSql(
+ "INSERT OVERWRITE target_table partition
(dt='2022-07-29') SELECT name FROM src_table where dt = '2022-07-29'")
+ .await();
+ partitions =
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show partitions
target_table").collect());
+ assertThat(partitions).hasSize(3);
+ assertThat(partitions.toString()).contains("dt=2022-07-29");
+ assertBatch("target_table", Arrays.asList());
+ }
+
private static List<String> fetchRows(Iterator<Row> iter, int size) {
List<String> strings = new ArrayList<>(size);
for (int i = 0; i < size; i++) {
diff --git
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
index 0df52e6b191..02928511460 100644
---
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
+++
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
@@ -104,12 +104,14 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("select * from destp order by
x").collect());
assertThat(results.toString()).isEqualTo("[+I[1, 1], +I[2, 2]]");
// static partitioned table
+ // The semantics of overwrite is to overwrite the original data,
so the
+ // p=1 partition is overwritten with an empty partition.
tableEnv.executeSql("insert overwrite table destp partition(p=1)
select x from src")
.await();
results =
CollectionUtil.iteratorToList(
tableEnv.executeSql("select * from destp order by
x").collect());
- assertThat(results.toString()).isEqualTo("[+I[1, 1], +I[2, 2]]");
+ assertThat(results.toString()).isEqualTo("[+I[2, 2]]");
} finally {
tableEnv.executeSql("drop database db1 cascade");
}
