You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/06/15 01:54:52 UTC
[flink] branch release-1.11 updated (cc6fb34 -> 89c224e)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from cc6fb34 [FLINK-17836][hive][doc] Add document for Hive dim join
new e66279c [FLINK-18265][fs-connector] Hidden files should be ignored when the filesystem table searches for partitions
new 89c224e [FLINK-18265][fs-connector] temp path in FileSystemOutputFormat should be deleted
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/table/utils/PartitionPathUtils.java | 14 ++++++++++----
.../table/planner/runtime/FileSystemITCaseBase.scala | 18 ++++++++++++++++++
.../flink/table/filesystem/FileSystemOutputFormat.java | 3 +++
.../table/filesystem/FileSystemOutputFormatTest.java | 5 +++++
4 files changed, 36 insertions(+), 4 deletions(-)
[flink] 02/02: [FLINK-18265][fs-connector] temp path in
FileSystemOutputFormat should be deleted
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 89c224e12c198d35a303487c4a2ab73e968214e0
Author: godfreyhe <go...@163.com>
AuthorDate: Fri Jun 12 14:31:06 2020 +0800
[FLINK-18265][fs-connector] temp path in FileSystemOutputFormat should be deleted
This closes #12628
---
.../org/apache/flink/table/filesystem/FileSystemOutputFormat.java | 3 +++
.../apache/flink/table/filesystem/FileSystemOutputFormatTest.java | 5 +++++
2 files changed, 8 insertions(+)
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOutputFormat.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOutputFormat.java
index 56011ab..0f35107 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOutputFormat.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOutputFormat.java
@@ -27,6 +27,7 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.table.api.TableException;
+import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.LinkedHashMap;
@@ -94,6 +95,8 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
committer.commitUpToCheckpoint(CHECKPOINT_ID);
} catch (Exception e) {
throw new TableException("Exception in finalizeGlobal", e);
+ } finally {
+ new File(tmpPath.getPath()).delete();
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemOutputFormatTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemOutputFormatTest.java
index 80118ff..520d56d 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemOutputFormatTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemOutputFormatTest.java
@@ -42,6 +42,7 @@ import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
/**
* Test for {@link FileSystemOutputFormat}.
@@ -128,6 +129,7 @@ public class FileSystemOutputFormatTest {
assertEquals(
"a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n",
content.values().iterator().next());
+ assertFalse(new File(tmpFile.toURI()).exists());
}
@Test
@@ -154,6 +156,7 @@ public class FileSystemOutputFormatTest {
assertEquals(
"a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n",
content.values().iterator().next());
+ assertFalse(new File(tmpFile.toURI()).exists());
}
@Test
@@ -173,6 +176,7 @@ public class FileSystemOutputFormatTest {
assertEquals(2, sortedContent.size());
assertEquals("a1,1\n" + "a2,2\n" + "a3,3\n", sortedContent.get("c=p1"));
assertEquals("a2,2\n", sortedContent.get("c=p2"));
+ assertFalse(new File(tmpFile.toURI()).exists());
}
@Test
@@ -198,6 +202,7 @@ public class FileSystemOutputFormatTest {
assertEquals(2, sortedContent.size());
assertEquals("a1,1\n" + "a2,2\n" + "a3,3\n", sortedContent.get("c=p1"));
assertEquals("a2,2\n", sortedContent.get("c=p2"));
+ assertFalse(new File(tmpFile.toURI()).exists());
}
private OneInputStreamOperatorTestHarness<Row, Object> createSink(
[flink] 01/02: [FLINK-18265][fs-connector] Hidden files should be
ignored when the filesystem table searches for partitions
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e66279c1b683dabffe233b37066c42cd21450c59
Author: godfreyhe <go...@163.com>
AuthorDate: Fri Jun 12 14:14:49 2020 +0800
[FLINK-18265][fs-connector] Hidden files should be ignored when the filesystem table searches for partitions
This closes #12628
---
.../apache/flink/table/utils/PartitionPathUtils.java | 14 ++++++++++----
.../table/planner/runtime/FileSystemITCaseBase.scala | 18 ++++++++++++++++++
2 files changed, 28 insertions(+), 4 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PartitionPathUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PartitionPathUtils.java
index 34e7c48..1ed4494 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PartitionPathUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PartitionPathUtils.java
@@ -195,10 +195,7 @@ public class PartitionPathUtils {
if (statuses == null) {
return null;
}
- return Arrays.stream(statuses).filter(fileStatus -> {
- String name = fileStatus.getPath().getName();
- return !name.startsWith("_") && !name.startsWith(".");
- }).toArray(FileStatus[]::new);
+ return Arrays.stream(statuses).filter(fileStatus -> !isHiddenFile(fileStatus)).toArray(FileStatus[]::new);
}
/**
@@ -213,6 +210,10 @@ public class PartitionPathUtils {
FileStatus[] generatedParts = getFileStatusRecurse(path, partitionNumber, fs);
List<Tuple2<LinkedHashMap<String, String>, Path>> ret = new ArrayList<>();
for (FileStatus part : generatedParts) {
+ // ignore hidden file
+ if (isHiddenFile(part)) {
+ continue;
+ }
ret.add(new Tuple2<>(extractPartitionSpecFromPath(part.getPath()), part.getPath()));
}
return ret;
@@ -322,4 +323,9 @@ public class PartitionPathUtils {
}
}
}
+
+ private static boolean isHiddenFile(FileStatus fileStatus) {
+ String name = fileStatus.getPath().getName();
+ return name.startsWith("_") || name.startsWith(".");
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
index 12981e9..28af0e9 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.runtime
import org.apache.flink.api.common.typeinfo.{TypeInformation, Types}
import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.core.fs.Path
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.planner.runtime.FileSystemITCaseBase._
@@ -28,9 +29,12 @@ import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.runtime.utils.TableEnvUtil.execInsertSqlAndWaitResult
import org.apache.flink.types.Row
+import org.junit.Assert.assertTrue
import org.junit.rules.TemporaryFolder
import org.junit.{Rule, Test}
+import java.io.File
+
import scala.collection.{JavaConverters, Seq}
/**
@@ -188,6 +192,20 @@ trait FileSystemITCaseBase {
}
@Test
+ def testPartitionWithHiddenFile(): Unit = {
+ execInsertSqlAndWaitResult(tableEnv, "insert into partitionedTable " +
+ "partition(a='1', b='1') select x, y from originalT where a=1 and b=1")
+
+ // create hidden partition dir
+ assertTrue(new File(new Path(resultPath + "/a=1/.b=2").toUri).mkdir())
+
+ check(
+ "select x, y from partitionedTable",
+ data_partition_1_1
+ )
+ }
+
+ @Test
def testNonPartition(): Unit = {
execInsertSqlAndWaitResult(tableEnv, "insert into nonPartitionedTable " +
"select x, y, a, b from originalT where a=1 and b=1")