[jira] [Commented] (FLINK-38774) Filesystem connector: "incomplete partition spec" when partition discovery descends into hidden directories

["abjfh (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-38774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127428#comment-18127428 ] 

abjfh commented on FLINK-38774:
-------------------------------

PR #27314 for this issue was closed by the stale bot while waiting for a review. I'm happy to rebase and reopen it.

[~davidradl] [~martijnvisser] could one of you assign this ticket to me (abjfh) so it can be picked up?

> Filesystem connector: "incomplete partition spec" when partition discovery descends into hidden directories
> -----------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-38774
>                 URL: https://issues.apache.org/jira/browse/FLINK-38774
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / FileSystem
>    Affects Versions: 1.16.0, 2.1.1
>            Reporter: abjfh
>            Priority: Major
>              Labels: bug, pull-request-available
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> Problem
> Reading a partitioned table of the filesystem connector fails when the table path contains a hidden directory at partition depth. Flink discovers partitions by walking the table path recursively (PartitionPathUtils.searchPartSpecAndPaths -> getFileStatusRecurse -> listStatusRecursively). Hidden directories are meant to be ignored, but the recursion descends into them: a non-hidden child of a hidden directory - for example {{job-123}} under the {{_temporary}} staging directory of the streaming file sink - is collected at the expected partition depth with an empty partition spec, and FileSystemTableSource.toFullLinkedPartSpec then rejects it. The table stays unreadable until the directory is removed; writing into the table is not affected.
> Reported in FLINK-31975 against Flink 1.16.0 with a filesystem table on S3; the recursion is unchanged on master.
> Error
> {code:java}
> Flink SQL> select * from country_page_view;
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.table.api.TableException: Partition keys are: [date, country], incomplete partition spec: {}
> {code}
> Root cause
> listStatusRecursively lists with fs.listStatus(dir) at every level, which returns hidden entries as well:
> {code:java}
> if (fileStatus.isDir()) {
>     for (FileStatus stat : fs.listStatus(fileStatus.getPath())) {
>         listStatusRecursively(fs, stat, level + 1, expectLevel, results);
>     }
> }
> {code}
> Hidden entries are filtered only on the collected result in searchPartSpecAndPaths:
> {code:java}
> for (FileStatus part : generatedParts) {
>     // ignore hidden file
>     if (isHiddenFile(part)) {
>         continue;
>     }
>     ret.add(new Tuple2<>(extractPartitionSpecFromPath(part.getPath()), part.getPath()));
> }
> {code}
> isHiddenFile checks the name of the entry itself, not that of its ancestors. For {{_temporary/job-123}} the hidden directory {{_temporary}} is dropped at the leaf, but {{job-123}} is not hidden by name and reaches the result with an empty partition spec - the spec that toFullLinkedPartSpec rejects.
> The same method has two further defects: fs.listStatus may return null (listStatusWithoutHidden handles that and returns null), which makes the for-each throw a NullPointerException; and the unfiltered listing is walked although hidden entries are never wanted.
> Reproduction
> Place a hidden staging subtree next to a real partition and read the table:
> {code}
> path/date=2019-8-30/country=China/   <- real partition
> path/_temporary/job-123/             <- hidden staging directory
> {code}
> PartitionPathUtils.searchPartSpecAndPaths(fs, path, 2) then returns two entries: the real partition, and {{job-123}} with an empty partition spec. With hidden directories skipped at every level, only the real partition is returned.
> Proposed fix
> Reuse the existing listStatusWithoutHidden helper at every recursion level and keep its null contract, so that an absent listing ends the recursion instead of failing:
> {code:java}
> if (fileStatus.isDir()) {
>     FileStatus[] children = listStatusWithoutHidden(fs, fileStatus.getPath());
>     if (children == null) {
>         return;
>     }
>     for (FileStatus stat : children) {
>         listStatusRecursively(fs, stat, level + 1, expectLevel, results);
>     }
> }
> {code}
> Verification
> The linked pull request adds a regression test to PartitionPathUtilsTest: a temporary directory holding the real partition {{date=2019-8-30/country=China}} and the hidden {{_temporary/job-123}} subtree, asserting that searchPartSpecAndPaths returns the real partition only; on the unfixed code the hidden child is returned as a second entry with an empty spec. The change is confined to PartitionPathUtils.listStatusRecursively, so paths without hidden entries keep their current behaviour.
> Related
> Supersedes FLINK-31975 (default catalog failed to retrieve partition Spec), which reports the same failure on 1.16.0. Pull requests: #27314 (closed by the stale bot, no technical objection) and #28913 (current).



--
This message was sent by Atlassian Jira
(v8.20.10#820010)