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 98842c482d4a72..200ab726b4f899 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 @@ -341,7 +341,11 @@ private static void listStatusRecursively( } if (fileStatus.isDir()) { - for (FileStatus stat : fs.listStatus(fileStatus.getPath())) { + FileStatus[] children = listStatusWithoutHidden(fs, fileStatus.getPath()); + if (children == null) { + return; + } + for (FileStatus stat : children) { listStatusRecursively(fs, stat, level + 1, expectLevel, results); } } diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PartitionPathUtilsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PartitionPathUtilsTest.java index 8aa9816a450df8..99eb394deb1be8 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PartitionPathUtilsTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PartitionPathUtilsTest.java @@ -18,13 +18,26 @@ package org.apache.flink.table.utils; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.core.fs.FileSystem; + import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.LinkedHashMap; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link org.apache.flink.table.utils.PartitionPathUtils}. */ class PartitionPathUtilsTest { + @TempDir + Path tmpDir; + @Test void testEscapeChar() { for (char c = 0; c <= 128; c++) { @@ -87,4 +100,38 @@ void testEscapePathNameWithCurlyBraces() { assertThat(actual).isEqualTo(expected); assertThat(PartitionPathUtils.unescapePathName(actual)).isEqualTo(origin); } + + /** + * FLINK-38774: {@link PartitionPathUtils#searchPartSpecAndPaths} must not descend into hidden + * directories. A hidden dir such as {@code _temporary} may contain non-hidden children (e.g. + * {@code job-123}) at partition depth; without skipping hidden dirs those children leak into + * the result with an empty partition spec, which later surfaces as a {@code TableException: + * incomplete partition spec}. + */ + @Test + void testSearchPartSpecAndPathsSkipsHiddenDirectories() throws IOException { + org.apache.flink.core.fs.Path basePath = + new org.apache.flink.core.fs.Path(tmpDir.toString(), "country_page_view"); + FileSystem fs = basePath.getFileSystem(); + + // Real partition: date=2019-8-30/country=China + Files.createDirectories( + Path.of( + tmpDir.toString(), + "country_page_view", + "date=2019-8-30", + "country=China")); + // Hidden _temporary dir whose non-hidden child sits at partition depth (2). + Files.createDirectories( + Path.of(tmpDir.toString(), "country_page_view", "_temporary", "job-123")); + + List, org.apache.flink.core.fs.Path>> parts = + PartitionPathUtils.searchPartSpecAndPaths(fs, basePath, 2); + + // Only the real partition is returned; the _temporary subtree is skipped entirely. + assertThat(parts).hasSize(1); + assertThat(parts.get(0).f0) + .containsEntry("date", "2019-8-30") + .containsEntry("country", "China"); + } }