From 8bb6c431bdca1a3dc1b13970c99b53e0c776ecdb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Tue, 4 Aug 2026 14:07:38 +0800 Subject: [PATCH] [FLINK-38774][table-common] Skip hidden directories when recursively listing partitions listStatusRecursively descended into every subdirectory returned by fs.listStatus, including hidden ones such as _temporary. A non-hidden child of a hidden dir (e.g. _temporary/job-123) sitting at partition depth was then collected with an empty partition spec, which later surfaces as 'TableException: incomplete partition spec' in the filesystem connector source. Reuse the existing listStatusWithoutHidden helper to filter hidden children at every recursion level (not only at the leaf, where searchPartSpecAndPaths already filtered). Add a null guard consistent with listStatusWithoutHidden's contract (returns null when fs.listStatus returns null), which also removes a latent NPE in the old for-each over a null list. Adds a @TempDir regression test that creates a real partition (date=2019-8-30/country=China) alongside a hidden _temporary/job-123 subtree and asserts only the real partition is returned. This supersedes the abandoned #27314. Co-Authored-By: Claude Fable 5 --- .../flink/table/utils/PartitionPathUtils.java | 6 ++- .../table/utils/PartitionPathUtilsTest.java | 47 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) 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"); + } }