[
https://issues.apache.org/jira/browse/FLINK-38774?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
abjfh updated FLINK-38774:
--------------------------
Description:
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).
was:# Since the original issue:FLINK-31975 is too old, I created a new issue
to fix it.
Summary: Filesystem connector: "incomplete partition spec" when
partition discovery descends into hidden directories (was: Fixed incorrect
handling of hidden directories when obtaining partition structure)
> 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)
