From 902281c578b1756c975e973082b7f4bb202e5375 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 06:02:09 -0700 Subject: [PATCH 01/15] [core] Parallelize per-partition file listing for catalog-managed format tables Format table split planning lists every partition's files serially, which is slow for internal (catalog-managed) tables with many partitions (thousands of object store LIST calls on the coordinator). Extract the per-partition listing into a PartitionSplitPlanner strategy with two implementations selected by the existing internal/external discriminator (partitionManager != null): - SequentialSplitPlanner (external): unchanged serial path; a missing partition directory is rethrown. - ParallelSplitPlanner (internal): bounded, order-preserving fan-out via ManifestReadThreadPool; a registered partition without a directory reads as empty (matching Hive), any other listing failure fails the whole scan. Split order and per-partition sort/bin-packing are unchanged, so output is identical to the serial path. Parallelism is bounded by a new option format-table.scan.list-parallelism (default #CPU); external tables are untouched. --- .../java/org/apache/paimon/CoreOptions.java | 16 +++ .../paimon/table/format/FormatTableScan.java | 121 ++++++++++++---- .../CatalogManagedPartitionScanTest.java | 129 ++++++++++++++++++ 3 files changed, 242 insertions(+), 24 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index e0c7fd4004c7..ef90977d7a98 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1230,6 +1230,18 @@ public InlineElement getDescription() { + "Note: Scale-up this parameter will increase memory usage while scanning manifest files. " + "We can consider downsize it when we encounter an out of memory exception while scanning"); + public static final ConfigOption FORMAT_TABLE_SCAN_LIST_PARALLELISM = + key("format-table.scan.list-parallelism") + .intType() + .noDefaultValue() + .withDescription( + "The parallelism of listing partition files during format table split " + + "planning for catalog-managed (internal) tables, default value is the " + + "size of cpu processor. Listing files is I/O bound, so a value larger " + + "than the cpu count can further reduce planning time; bound it by the " + + "object store connection limit. External (filesystem-discovered) tables " + + "always list serially and ignore this option."); + public static final ConfigOption STREAMING_READ_SNAPSHOT_DELAY = key("streaming.read.snapshot.delay") .durationType() @@ -3708,6 +3720,10 @@ public Integer scanManifestParallelism() { return options.get(SCAN_MANIFEST_PARALLELISM); } + public Integer formatTableScanListParallelism() { + return options.get(FORMAT_TABLE_SCAN_LIST_PARALLELISM); + } + public Integer scanBucket() { return options.get(SCAN_BUCKET); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 2dee6fe1237b..5458a8829e03 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -52,6 +52,7 @@ import org.apache.paimon.types.VarCharType; import org.apache.paimon.utils.BinPacking; import org.apache.paimon.utils.InternalRowPartitionComputer; +import org.apache.paimon.utils.ManifestReadThreadPool; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; @@ -73,6 +74,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.function.Function; import static org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed; import static org.apache.paimon.format.text.TextLineReader.isDefaultDelimiter; @@ -185,34 +187,24 @@ public List splits() { try { FileIO fileIO = table.fileIO(); if (!table.partitionKeys().isEmpty()) { + List partitions = new ArrayList<>(); for (Pair, Path> pair : findPartitions()) { - LinkedHashMap partitionSpec = pair.getKey(); - BinaryRow partitionRow = toPartitionRow(partitionSpec); + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + // Filter partitions in memory before any file listing, so only the + // IO is parallelized below. if (partitionFilter == null || partitionFilter.test(partitionRow)) { - try { - splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); - } catch (FileNotFoundException e) { - if (partitionManager == null) { - throw e; - } - // A registered partition without a directory reads as empty, - // matching Hive (e.g. ADD PARTITION before the first INSERT). - // Warn so a directory removed behind the catalog's back stays - // discoverable. - LOG.warn( - "Partition '{}' of format table {} is registered in " - + "the catalog but its directory '{}' does not " - + "exist; treating the partition as empty. If the " - + "directory was removed on purpose, drop the " - + "partition or repair the metadata, e.g. with " - + "MSCK REPAIR TABLE.", - PartitionPathUtils.generatePartitionName( - partitionSpec, false), - table.fullName(), - pair.getValue()); - } + partitions.add( + new PartitionToScan( + pair.getKey(), pair.getValue(), partitionRow)); } } + // Internal (catalog-managed) tables list partition files in parallel; + // external (filesystem-discovered) tables keep the sequential path. + PartitionSplitPlanner planner = + partitionManager != null + ? new ParallelSplitPlanner() + : new SequentialSplitPlanner(); + splits.addAll(planner.plan(fileIO, partitions)); } else { splits.addAll(createSplits(fileIO, new Path(table.location()), null)); } @@ -227,6 +219,87 @@ public List splits() { } } + /** A partition selected for scanning: its spec, directory path, and encoded row. */ + private static final class PartitionToScan { + private final LinkedHashMap spec; + private final Path path; + private final BinaryRow row; + + private PartitionToScan(LinkedHashMap spec, Path path, BinaryRow row) { + this.spec = spec; + this.path = path; + this.row = row; + } + } + + /** Lists partition files and builds splits; per-partition sort and bin-packing are shared. */ + private interface PartitionSplitPlanner { + List plan(FileIO fileIO, List partitions) throws IOException; + } + + /** + * Sequential planner for external (filesystem-discovered) tables: a missing partition directory + * is a hard error, matching the pre-parallel behavior. + */ + private class SequentialSplitPlanner implements PartitionSplitPlanner { + @Override + public List plan(FileIO fileIO, List partitions) + throws IOException { + List splits = new ArrayList<>(); + for (PartitionToScan partition : partitions) { + splits.addAll(createSplits(fileIO, partition.path, partition.row)); + } + return splits; + } + } + + /** + * Parallel planner for internal (catalog-managed) tables: lists partitions concurrently with a + * bounded, order-preserving thread pool. A registered partition without a directory reads as + * empty (matching Hive); any other listing failure fails the whole scan. + */ + private class ParallelSplitPlanner implements PartitionSplitPlanner { + @Override + public List plan(FileIO fileIO, List partitions) { + List splits = new ArrayList<>(); + if (partitions.isEmpty()) { + return splits; + } + Function> lister = + partition -> { + try { + return createSplits(fileIO, partition.path, partition.row); + } catch (FileNotFoundException e) { + warnMissingPartition(partition); + return Collections.emptyList(); + } catch (IOException e) { + // Fail the whole scan on any real listing error; never return a + // truncated result. + throw new RuntimeException( + "Failed to list files for partition " + partition.path, e); + } + }; + ManifestReadThreadPool.randomlyExecuteSequentialReturn( + lister, partitions, coreOptions.formatTableScanListParallelism()) + .forEachRemaining(splits::add); + return splits; + } + } + + private void warnMissingPartition(PartitionToScan partition) { + // A registered partition without a directory reads as empty, matching Hive (e.g. ADD + // PARTITION before the first INSERT). Warn so a directory removed behind the catalog's back + // stays discoverable. + LOG.warn( + "Partition '{}' of format table {} is registered in the catalog but its directory " + + "'{}' does not exist; treating the partition as empty. If the directory " + + "was removed on purpose, drop the partition or repair the metadata, e.g. " + + "with MSCK REPAIR TABLE.", + PartitionPathUtils.generatePartitionName(partition.spec, false), + table.fullName(), + partition.path); + } + List, Path>> findPartitions() { if (partitionManager != null) { Optional extracted = extractPartitionPredicate(partitionFilter); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java index de2127f4939a..b97f94cac30e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java @@ -56,6 +56,7 @@ import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; +import static org.apache.paimon.CoreOptions.FORMAT_TABLE_SCAN_LIST_PARALLELISM; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -506,4 +507,132 @@ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { return super.listFiles(path, recursive); } } + + @Test + void testInternalParallelListingMatchesSerialAndIsDeterministic() throws Exception { + TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + List partitions = new ArrayList<>(); + List expected = new ArrayList<>(); + for (int m = 10; m <= 15; m++) { + partitions.add(partition("2025", Integer.toString(m))); + expected.add(writeDataFile(fileIO, tablePath, "year=2025/month=" + m)); + } + + List serial = + plannedFiles( + new FormatTableScan( + managedStringTable(fileIO, tablePath, partitions, 1), + null, + null) + .plan() + .splits()); + List parallel = + plannedFiles( + new FormatTableScan( + managedStringTable(fileIO, tablePath, partitions, 8), + null, + null) + .plan() + .splits()); + + assertThat(parallel).containsExactlyElementsOf(serial); + assertThat(parallel).containsExactlyElementsOf(expected); + } + + @Test + void testInternalMissingPartitionDirectoryTreatedAsEmpty() throws Exception { + InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + List partitions = + Arrays.asList( + partition("2025", "10"), partition("2025", "11"), partition("2025", "12")); + Path oct = writeDataFile(fileIO, tablePath, "year=2025/month=10"); + Path dec = writeDataFile(fileIO, tablePath, "year=2025/month=12"); + // month=11 is registered but its directory is missing -> FileNotFound -> treated as empty. + fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing")); + + List files = + plannedFiles( + new FormatTableScan( + managedStringTable(fileIO, tablePath, partitions, 4), + null, + null) + .plan() + .splits()); + + assertThat(files).containsExactlyInAnyOrder(oct, dec); + } + + @Test + void testInternalListingIoErrorFailsWholeScan() throws Exception { + InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + List partitions = + Arrays.asList(partition("2025", "10"), partition("2025", "11")); + writeDataFile(fileIO, tablePath, "year=2025/month=10"); + writeDataFile(fileIO, tablePath, "year=2025/month=11"); + // A real (non-FileNotFound) listing error must fail the whole scan, not truncate. + fileIO.failListFilesContaining("month=11", new IOException("boom")); + + FormatTable table = managedStringTable(fileIO, tablePath, partitions, 4); + assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits()) + .isInstanceOf(RuntimeException.class); + } + + @Test + void testExternalMissingDirectoryStillFailsScan() throws Exception { + InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + writeDataFile(fileIO, tablePath, "year=2025/month=10"); + writeDataFile(fileIO, tablePath, "year=2025/month=11"); + // External tables (partitionManager == null) keep the rethrow-on-missing behavior. + fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing")); + + FormatTable table = createStringPartitionTable(fileIO, tablePath, null); + assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits()) + .isInstanceOf(RuntimeException.class); + } + + private FormatTable managedStringTable( + LocalFileIO fileIO, Path tablePath, List partitions, int parallelism) { + RowType rowType = + RowType.builder() + .field("year", DataTypes.STRING()) + .field("month", DataTypes.STRING()) + .field("id", DataTypes.INT()) + .build(); + Map options = new LinkedHashMap<>(); + options.put(FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), "false"); + options.put(FORMAT_TABLE_SCAN_LIST_PARALLELISM.key(), Integer.toString(parallelism)); + return FormatTable.builder() + .fileIO(fileIO) + .identifier(IDENTIFIER) + .rowType(rowType) + .partitionKeys(Arrays.asList("year", "month")) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options(options) + .partitionManager(recordingCatalog(partitions)) + .build(); + } + + private static class InjectingLocalFileIO extends LocalFileIO { + + private final Map listFailures = new LinkedHashMap<>(); + + void failListFilesContaining(String marker, IOException failure) { + listFailures.put(marker, failure); + } + + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + for (Map.Entry entry : listFailures.entrySet()) { + if (path.toString().contains(entry.getKey())) { + throw entry.getValue(); + } + } + return super.listFiles(path, recursive); + } + } } From 616f04b4101712452d2e72c6f5571e845668ec6d Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 06:32:38 -0700 Subject: [PATCH 02/15] [core] Drop the PartitionSplitPlanner abstraction Inline the internal/external branch directly in splits(): a private listPartitionFilesInParallel(...) for internal (catalog-managed) tables and the serial loop for external tables. Behavior is unchanged. --- .../paimon/table/format/FormatTableScan.java | 126 +++++++----------- 1 file changed, 47 insertions(+), 79 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 5458a8829e03..54dd8bbe759b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -187,24 +187,29 @@ public List splits() { try { FileIO fileIO = table.fileIO(); if (!table.partitionKeys().isEmpty()) { - List partitions = new ArrayList<>(); + List, Path>> partitions = new ArrayList<>(); for (Pair, Path> pair : findPartitions()) { - BinaryRow partitionRow = toPartitionRow(pair.getKey()); - // Filter partitions in memory before any file listing, so only the - // IO is parallelized below. - if (partitionFilter == null || partitionFilter.test(partitionRow)) { - partitions.add( - new PartitionToScan( - pair.getKey(), pair.getValue(), partitionRow)); + // Filter partitions in memory before any file listing. + if (partitionFilter == null + || partitionFilter.test(toPartitionRow(pair.getKey()))) { + partitions.add(pair); + } + } + if (partitionManager != null) { + // Internal (catalog-managed) table: list partition files in parallel. + splits.addAll(listPartitionFilesInParallel(fileIO, partitions)); + } else { + // External (filesystem-discovered) table: list serially; a missing + // partition + // directory is a hard error (rethrown to fail the whole scan). + for (Pair, Path> pair : partitions) { + splits.addAll( + createSplits( + fileIO, + pair.getValue(), + toPartitionRow(pair.getKey()))); } } - // Internal (catalog-managed) tables list partition files in parallel; - // external (filesystem-discovered) tables keep the sequential path. - PartitionSplitPlanner planner = - partitionManager != null - ? new ParallelSplitPlanner() - : new SequentialSplitPlanner(); - splits.addAll(planner.plan(fileIO, partitions)); } else { splits.addAll(createSplits(fileIO, new Path(table.location()), null)); } @@ -219,74 +224,37 @@ public List splits() { } } - /** A partition selected for scanning: its spec, directory path, and encoded row. */ - private static final class PartitionToScan { - private final LinkedHashMap spec; - private final Path path; - private final BinaryRow row; - - private PartitionToScan(LinkedHashMap spec, Path path, BinaryRow row) { - this.spec = spec; - this.path = path; - this.row = row; - } - } - - /** Lists partition files and builds splits; per-partition sort and bin-packing are shared. */ - private interface PartitionSplitPlanner { - List plan(FileIO fileIO, List partitions) throws IOException; - } - /** - * Sequential planner for external (filesystem-discovered) tables: a missing partition directory - * is a hard error, matching the pre-parallel behavior. + * Lists partition files for an internal (catalog-managed) table in parallel with a bounded, + * order-preserving thread pool. A registered partition without a directory reads as empty + * (matching Hive); any other listing failure fails the whole scan. */ - private class SequentialSplitPlanner implements PartitionSplitPlanner { - @Override - public List plan(FileIO fileIO, List partitions) - throws IOException { - List splits = new ArrayList<>(); - for (PartitionToScan partition : partitions) { - splits.addAll(createSplits(fileIO, partition.path, partition.row)); - } - return splits; - } - } - - /** - * Parallel planner for internal (catalog-managed) tables: lists partitions concurrently with a - * bounded, order-preserving thread pool. A registered partition without a directory reads as - * empty (matching Hive); any other listing failure fails the whole scan. - */ - private class ParallelSplitPlanner implements PartitionSplitPlanner { - @Override - public List plan(FileIO fileIO, List partitions) { - List splits = new ArrayList<>(); - if (partitions.isEmpty()) { - return splits; - } - Function> lister = - partition -> { - try { - return createSplits(fileIO, partition.path, partition.row); - } catch (FileNotFoundException e) { - warnMissingPartition(partition); - return Collections.emptyList(); - } catch (IOException e) { - // Fail the whole scan on any real listing error; never return a - // truncated result. - throw new RuntimeException( - "Failed to list files for partition " + partition.path, e); - } - }; - ManifestReadThreadPool.randomlyExecuteSequentialReturn( - lister, partitions, coreOptions.formatTableScanListParallelism()) - .forEachRemaining(splits::add); + private List listPartitionFilesInParallel( + FileIO fileIO, List, Path>> partitions) { + List splits = new ArrayList<>(); + if (partitions.isEmpty()) { return splits; } + Function, Path>, List> lister = + pair -> { + try { + return createSplits(fileIO, pair.getValue(), toPartitionRow(pair.getKey())); + } catch (FileNotFoundException e) { + warnMissingPartition(pair.getKey(), pair.getValue()); + return Collections.emptyList(); + } catch (IOException e) { + // Fail the whole scan on any real listing error; never truncate. + throw new RuntimeException( + "Failed to list files for partition " + pair.getValue(), e); + } + }; + ManifestReadThreadPool.randomlyExecuteSequentialReturn( + lister, partitions, coreOptions.formatTableScanListParallelism()) + .forEachRemaining(splits::add); + return splits; } - private void warnMissingPartition(PartitionToScan partition) { + private void warnMissingPartition(LinkedHashMap spec, Path path) { // A registered partition without a directory reads as empty, matching Hive (e.g. ADD // PARTITION before the first INSERT). Warn so a directory removed behind the catalog's back // stays discoverable. @@ -295,9 +263,9 @@ private void warnMissingPartition(PartitionToScan partition) { + "'{}' does not exist; treating the partition as empty. If the directory " + "was removed on purpose, drop the partition or repair the metadata, e.g. " + "with MSCK REPAIR TABLE.", - PartitionPathUtils.generatePartitionName(partition.spec, false), + PartitionPathUtils.generatePartitionName(spec, false), table.fullName(), - partition.path); + path); } List, Path>> findPartitions() { From 4a826a079d9078c7bf3827d41c0f9d046da9e8be Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 06:38:47 -0700 Subject: [PATCH 03/15] [core] Remove redundant comments in FormatTableScan split planning --- .../apache/paimon/table/format/FormatTableScan.java | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 54dd8bbe759b..f003f8bdbd16 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -189,19 +189,14 @@ public List splits() { if (!table.partitionKeys().isEmpty()) { List, Path>> partitions = new ArrayList<>(); for (Pair, Path> pair : findPartitions()) { - // Filter partitions in memory before any file listing. if (partitionFilter == null || partitionFilter.test(toPartitionRow(pair.getKey()))) { partitions.add(pair); } } if (partitionManager != null) { - // Internal (catalog-managed) table: list partition files in parallel. splits.addAll(listPartitionFilesInParallel(fileIO, partitions)); } else { - // External (filesystem-discovered) table: list serially; a missing - // partition - // directory is a hard error (rethrown to fail the whole scan). for (Pair, Path> pair : partitions) { splits.addAll( createSplits( @@ -224,11 +219,6 @@ public List splits() { } } - /** - * Lists partition files for an internal (catalog-managed) table in parallel with a bounded, - * order-preserving thread pool. A registered partition without a directory reads as empty - * (matching Hive); any other listing failure fails the whole scan. - */ private List listPartitionFilesInParallel( FileIO fileIO, List, Path>> partitions) { List splits = new ArrayList<>(); @@ -243,7 +233,6 @@ private List listPartitionFilesInParallel( warnMissingPartition(pair.getKey(), pair.getValue()); return Collections.emptyList(); } catch (IOException e) { - // Fail the whole scan on any real listing error; never truncate. throw new RuntimeException( "Failed to list files for partition " + pair.getValue(), e); } From d7e9235d11fe5745f2d6ccc9d23699a4061ae79b Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 06:50:48 -0700 Subject: [PATCH 04/15] [core] Default format-table list parallelism to 64; filter partitions in the worker --- .../java/org/apache/paimon/CoreOptions.java | 10 +++---- .../paimon/table/format/FormatTableScan.java | 26 ++++++++----------- 2 files changed, 15 insertions(+), 21 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index ef90977d7a98..206cc61a1c05 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1233,14 +1233,12 @@ public InlineElement getDescription() { public static final ConfigOption FORMAT_TABLE_SCAN_LIST_PARALLELISM = key("format-table.scan.list-parallelism") .intType() - .noDefaultValue() + .defaultValue(64) .withDescription( "The parallelism of listing partition files during format table split " - + "planning for catalog-managed (internal) tables, default value is the " - + "size of cpu processor. Listing files is I/O bound, so a value larger " - + "than the cpu count can further reduce planning time; bound it by the " - + "object store connection limit. External (filesystem-discovered) tables " - + "always list serially and ignore this option."); + + "planning for catalog-managed (internal) tables. External " + + "(filesystem-discovered) tables always list serially and ignore " + + "this option."); public static final ConfigOption STREAMING_READ_SNAPSHOT_DELAY = key("streaming.read.snapshot.delay") diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index f003f8bdbd16..4350087fba41 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -187,22 +187,14 @@ public List splits() { try { FileIO fileIO = table.fileIO(); if (!table.partitionKeys().isEmpty()) { - List, Path>> partitions = new ArrayList<>(); - for (Pair, Path> pair : findPartitions()) { - if (partitionFilter == null - || partitionFilter.test(toPartitionRow(pair.getKey()))) { - partitions.add(pair); - } - } if (partitionManager != null) { - splits.addAll(listPartitionFilesInParallel(fileIO, partitions)); + splits.addAll(listPartitionFilesInParallel(fileIO, findPartitions())); } else { - for (Pair, Path> pair : partitions) { - splits.addAll( - createSplits( - fileIO, - pair.getValue(), - toPartitionRow(pair.getKey()))); + for (Pair, Path> pair : findPartitions()) { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter == null || partitionFilter.test(partitionRow)) { + splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); + } } } } else { @@ -227,8 +219,12 @@ private List listPartitionFilesInParallel( } Function, Path>, List> lister = pair -> { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter != null && !partitionFilter.test(partitionRow)) { + return Collections.emptyList(); + } try { - return createSplits(fileIO, pair.getValue(), toPartitionRow(pair.getKey())); + return createSplits(fileIO, pair.getValue(), partitionRow); } catch (FileNotFoundException e) { warnMissingPartition(pair.getKey(), pair.getValue()); return Collections.emptyList(); From 38a7f5e326f5b2b12a0c9a097999436432ecff32 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 06:59:20 -0700 Subject: [PATCH 05/15] [core] Use a dedicated format-table list pool (max 1000, 64 per scan) like Trino --- .../paimon/table/format/FormatTableScan.java | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 4350087fba41..a1c7f31ef4aa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -52,9 +52,10 @@ import org.apache.paimon.types.VarCharType; import org.apache.paimon.utils.BinPacking; import org.apache.paimon.utils.InternalRowPartitionComputer; -import org.apache.paimon.utils.ManifestReadThreadPool; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.SemaphoredDelegatingExecutor; +import org.apache.paimon.utils.ThreadPoolUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -74,6 +75,8 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; import java.util.function.Function; import static org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed; @@ -86,6 +89,11 @@ public class FormatTableScan implements InnerTableScan { private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class); + private static final int LIST_POOL_MAX_THREADS = 1000; + private static final ThreadPoolExecutor LIST_POOL = + ThreadPoolUtils.createCachedThreadPool( + LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); + final FormatTable table; final CoreOptions coreOptions; @Nullable private PartitionPredicate partitionFilter; @@ -233,8 +241,12 @@ private List listPartitionFilesInParallel( "Failed to list files for partition " + pair.getValue(), e); } }; - ManifestReadThreadPool.randomlyExecuteSequentialReturn( - lister, partitions, coreOptions.formatTableScanListParallelism()) + int parallelism = coreOptions.formatTableScanListParallelism(); + ExecutorService executor = + parallelism >= LIST_POOL_MAX_THREADS + ? LIST_POOL + : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); + ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) .forEachRemaining(splits::add); return splits; } From fcc41e02f2215340530678321404ec614da5302a Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 07:29:32 -0700 Subject: [PATCH 06/15] [core] Drop the missing-partition warn comment in FormatTableScan --- .../java/org/apache/paimon/table/format/FormatTableScan.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index a1c7f31ef4aa..052f25873566 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -252,9 +252,6 @@ private List listPartitionFilesInParallel( } private void warnMissingPartition(LinkedHashMap spec, Path path) { - // A registered partition without a directory reads as empty, matching Hive (e.g. ADD - // PARTITION before the first INSERT). Warn so a directory removed behind the catalog's back - // stays discoverable. LOG.warn( "Partition '{}' of format table {} is registered in the catalog but its directory " + "'{}' does not exist; treating the partition as empty. If the directory " From cc06a64f697843c9761ef0ecd77ee6f3683932d9 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 18:40:44 -0700 Subject: [PATCH 07/15] [docs] Regenerate core config docs for format-table.scan.list-parallelism --- docs/generated/core_configuration.html | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 2841618e7c34..c165f3877212 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -746,6 +746,12 @@ Boolean Format table file path only contain partition value. + +
format-table.scan.list-parallelism
+ 64 + Integer + The parallelism of listing partition files during format table split planning for catalog-managed (internal) tables. External (filesystem-discovered) tables always list serially and ignore this option. +
full-compaction.delta-commits
(none) From ae90b041a7cf4e5840dfafce797923be35766f72 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 24 Jul 2026 18:52:15 -0700 Subject: [PATCH 08/15] [core] Clamp format-table list parallelism to at least 1 A non-positive format-table.scan.list-parallelism would create a SemaphoredDelegatingExecutor with zero/negative permits, making the listing acquire() block forever. Clamp to a minimum of 1 so a misconfigured value degrades to serial listing instead of hanging. --- .../java/org/apache/paimon/table/format/FormatTableScan.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 052f25873566..f0a370451499 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -241,7 +241,7 @@ private List listPartitionFilesInParallel( "Failed to list files for partition " + pair.getValue(), e); } }; - int parallelism = coreOptions.formatTableScanListParallelism(); + int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); ExecutorService executor = parallelism >= LIST_POOL_MAX_THREADS ? LIST_POOL From a3f3d07f1f52f97d7d4de39c65ccf46ac07dfd39 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 16:04:10 +0800 Subject: [PATCH 09/15] [core] Refactor format table split enumeration Move format-table partition discovery and split generation into a package-private FormatTableSplitEnumerator. Keep FormatTablePartitionManager focused on catalog metadata, parallelize catalog partition file listing, and preserve serial filesystem discovery. --- docs/generated/core_configuration.html | 2 +- .../java/org/apache/paimon/CoreOptions.java | 6 +- .../paimon/table/format/FormatTableScan.java | 320 +-------------- .../format/FormatTableSplitEnumerator.java | 376 ++++++++++++++++++ .../CatalogManagedPartitionScanTest.java | 121 ++++-- 5 files changed, 485 insertions(+), 340 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index c165f3877212..d6f44e9f334c 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -750,7 +750,7 @@
format-table.scan.list-parallelism
64 Integer - The parallelism of listing partition files during format table split planning for catalog-managed (internal) tables. External (filesystem-discovered) tables always list serially and ignore this option. + The parallelism of listing partition files during split planning for a Format Table with catalog-managed partitions.
full-compaction.delta-commits
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 206cc61a1c05..c5989ff9b3fd 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1235,10 +1235,8 @@ public InlineElement getDescription() { .intType() .defaultValue(64) .withDescription( - "The parallelism of listing partition files during format table split " - + "planning for catalog-managed (internal) tables. External " - + "(filesystem-discovered) tables always list serially and ignore " - + "this option."); + "The parallelism of listing partition files during split planning for " + + "a Format Table with catalog-managed partitions."); public static final ConfigOption STREAMING_READ_SNAPSHOT_DELAY = key("streaming.read.snapshot.delay") diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index f0a370451499..90003555d5df 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -23,15 +23,8 @@ import org.apache.paimon.casting.CastExecutors; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; -import org.apache.paimon.data.GenericRow; -import org.apache.paimon.data.serializer.InternalRowSerializer; -import org.apache.paimon.format.csv.CsvOptions; -import org.apache.paimon.format.json.JsonOptions; -import org.apache.paimon.fs.FileIO; -import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.PartitionEntry; -import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.partition.PartitionPredicate.AndPartitionPredicate; @@ -50,24 +43,15 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.RowType; import org.apache.paimon.types.VarCharType; -import org.apache.paimon.utils.BinPacking; import org.apache.paimon.utils.InternalRowPartitionComputer; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; -import org.apache.paimon.utils.SemaphoredDelegatingExecutor; -import org.apache.paimon.utils.ThreadPoolUtils; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; -import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; -import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; @@ -75,33 +59,18 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.function.Function; -import static org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed; -import static org.apache.paimon.format.text.TextLineReader.isDefaultDelimiter; -import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow; import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; /** {@link TableScan} for {@link FormatTable}. */ public class FormatTableScan implements InnerTableScan { - private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class); - - private static final int LIST_POOL_MAX_THREADS = 1000; - private static final ThreadPoolExecutor LIST_POOL = - ThreadPoolUtils.createCachedThreadPool( - LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); - final FormatTable table; final CoreOptions coreOptions; @Nullable private PartitionPredicate partitionFilter; @Nullable private final FormatTablePartitionManager partitionManager; + private final FormatTableSplitEnumerator splitEnumerator; @Nullable private final Integer limit; - private final long targetSplitSize; - private final long openFileCost; - private final FormatTable.Format format; public FormatTableScan( FormatTable table, @@ -112,9 +81,7 @@ public FormatTableScan( this.partitionFilter = partitionFilter; this.limit = limit; this.partitionManager = table.partitionManager(); - this.targetSplitSize = coreOptions.splitTargetSize(); - this.openFileCost = coreOptions.splitOpenFileCost(); - this.format = table.format(); + this.splitEnumerator = new FormatTableSplitEnumerator(table, coreOptions, partitionManager); } @Override @@ -134,7 +101,7 @@ public List listPartitionEntries() { List partitions = partitionManager.listPartitions(Collections.emptyMap(), null); if (partitions.isEmpty()) { - warnIfFilesystemPartitionsExist(); + splitEnumerator.warnIfFilesystemPartitionsExist(); } boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); List entries = new ArrayList<>(partitions.size()); @@ -145,7 +112,9 @@ public List listPartitionEntries() { } entries.add( new PartitionEntry( - toPartitionRow(normalizeSpec(partition.spec(), onlyValueInPath)), + toPartitionRow( + splitEnumerator.normalizeSpec( + partition.spec(), onlyValueInPath)), partition.recordCount(), partition.fileSizeInBytes(), partition.fileCount(), @@ -182,10 +151,7 @@ public static boolean isDataFileName(String fileName) { } BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { - RowType partitionType = table.partitionType(); - GenericRow row = - convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName()); - return new InternalRowSerializer(partitionType).toBinaryRow(row); + return splitEnumerator.toPartitionRow(partitionSpec); } private class FormatTableScanPlan implements Plan { @@ -193,21 +159,7 @@ private class FormatTableScanPlan implements Plan { public List splits() { List splits = new ArrayList<>(); try { - FileIO fileIO = table.fileIO(); - if (!table.partitionKeys().isEmpty()) { - if (partitionManager != null) { - splits.addAll(listPartitionFilesInParallel(fileIO, findPartitions())); - } else { - for (Pair, Path> pair : findPartitions()) { - BinaryRow partitionRow = toPartitionRow(pair.getKey()); - if (partitionFilter == null || partitionFilter.test(partitionRow)) { - splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); - } - } - } - } else { - splits.addAll(createSplits(fileIO, new Path(table.location()), null)); - } + splits.addAll(splitEnumerator.enumerate(partitionFilter)); // Keep all splits for a positive limit because FormatDataSplit has no row count. if (limit != null && limit <= 0) { return new ArrayList<>(); @@ -219,200 +171,8 @@ public List splits() { } } - private List listPartitionFilesInParallel( - FileIO fileIO, List, Path>> partitions) { - List splits = new ArrayList<>(); - if (partitions.isEmpty()) { - return splits; - } - Function, Path>, List> lister = - pair -> { - BinaryRow partitionRow = toPartitionRow(pair.getKey()); - if (partitionFilter != null && !partitionFilter.test(partitionRow)) { - return Collections.emptyList(); - } - try { - return createSplits(fileIO, pair.getValue(), partitionRow); - } catch (FileNotFoundException e) { - warnMissingPartition(pair.getKey(), pair.getValue()); - return Collections.emptyList(); - } catch (IOException e) { - throw new RuntimeException( - "Failed to list files for partition " + pair.getValue(), e); - } - }; - int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); - ExecutorService executor = - parallelism >= LIST_POOL_MAX_THREADS - ? LIST_POOL - : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); - ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) - .forEachRemaining(splits::add); - return splits; - } - - private void warnMissingPartition(LinkedHashMap spec, Path path) { - LOG.warn( - "Partition '{}' of format table {} is registered in the catalog but its directory " - + "'{}' does not exist; treating the partition as empty. If the directory " - + "was removed on purpose, drop the partition or repair the metadata, e.g. " - + "with MSCK REPAIR TABLE.", - PartitionPathUtils.generatePartitionName(spec, false), - table.fullName(), - path); - } - List, Path>> findPartitions() { - if (partitionManager != null) { - Optional extracted = extractPartitionPredicate(partitionFilter); - Map prefix = leadingEqualityPrefix(extracted); - // The whole predicate goes to the manager as a pushdown hint, together with the - // prefix; the catalog may return a superset. - Predicate catalogFilter = extracted.orElse(null); - List partitions = partitionManager.listPartitions(prefix, catalogFilter); - if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) { - warnIfFilesystemPartitionsExist(); - } - // The prefix and filter are coarse pre-filters; the full predicate is applied per - // partition in the plan, as it is for filesystem discovery. - return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); - } - LOG.debug( - "Find partitions for format table {}, partition filter: {}", - table.name(), - partitionFilter); - boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); - if (partitionFilter instanceof MultiplePartitionPredicate) { - // generate partitions directly - Set partitions = ((MultiplePartitionPredicate) partitionFilter).partitions(); - return generatePartitions( - table.partitionKeys(), - table.partitionType(), - table.defaultPartName(), - new Path(table.location()), - partitions, - onlyValueInPath); - } else { - // search paths with partition filter optimization - // This will prune partition directories early during traversal, - // which is especially important for cloud storage like OSS/S3 - Optional predicate = extractPartitionPredicate(partitionFilter); - LOG.debug( - "Extracted predicate for format table {} partition pruning: {}", - table.name(), - predicate.orElse(null)); - - Pair scanPathAndLevel = - computeScanPathAndLevel( - new Path(table.location()), - table.partitionKeys(), - predicate, - table.partitionType(), - onlyValueInPath); - return searchPartSpecAndPaths( - table.fileIO(), - scanPathAndLevel.getLeft(), - scanPathAndLevel.getRight(), - table.partitionKeys(), - onlyValueInPath, - predicate.orElse(null), - table.partitionType(), - table.defaultPartName()); - } - } - - /** - * Turn the partitions a catalog reports into the specs and directories to read. Catalog - * metadata is not trusted for path construction: a spec that cannot form a partition directory - * of this table is rejected rather than resolved to some other directory. - */ - private List, Path>> toSpecsAndPaths( - List partitions, boolean onlyValueInPath) { - List, Path>> result = new ArrayList<>(partitions.size()); - Path tablePath = new Path(table.location()); - // Do not trust the catalog to be duplicate-free: a repeated spec would double every split - // of that partition and silently duplicate query results. - Set seenPartitionPaths = new HashSet<>(partitions.size()); - for (Partition partition : partitions) { - LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath); - String partitionPath = - PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath); - if (seenPartitionPaths.add(partitionPath)) { - result.add(Pair.of(spec, new Path(tablePath, partitionPath))); - } - } - return result; - } - - /** Order a catalog spec by the table partition keys and check it can form a directory. */ - private LinkedHashMap normalizeSpec( - @Nullable Map spec, boolean onlyValueInPath) { - List partitionKeys = table.partitionKeys(); - if (spec == null - || spec.size() != partitionKeys.size() - || !spec.keySet().containsAll(partitionKeys)) { - throw corruptPartitionSpec(spec); - } - LinkedHashMap normalized = new LinkedHashMap<>(); - for (String partitionKey : partitionKeys) { - String value = spec.get(partitionKey); - // In a value-only layout, '.' and '..' are complete path components and would resolve - // to a directory outside the table. - try { - PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath); - } catch (IllegalArgumentException e) { - throw corruptPartitionSpec(spec); - } - normalized.put(partitionKey, value); - } - return normalized; - } - - private IllegalStateException corruptPartitionSpec(@Nullable Map spec) { - return new IllegalStateException( - String.format( - "Catalog returned corrupt partition metadata %s for format table %s; " - + "expected exactly the partition keys %s with values usable as " - + "path components.", - spec, table.fullName(), table.partitionKeys())); - } - - /** - * The leading equality prefix of the partition predicate, in partition-key order, pushed down - * to the partition catalog. Empty when there is no predicate or it does not start with - * equalities on the leading partition keys. - */ - private Map leadingEqualityPrefix(Optional predicate) { - if (!predicate.isPresent()) { - return Collections.emptyMap(); - } - return extractLeadingEqualityPartitionSpecWhenOnlyAnd( - table.partitionKeys(), predicate.get(), table.partitionType()); - } - - /** - * Warn when the catalog knows no partitions but the table directory contains subdirectories: - * typically a table that predates catalog-managed partitions (or was written by a client that - * does not register partitions) and needs a metadata sync before its data becomes visible. - */ - private void warnIfFilesystemPartitionsExist() { - try { - for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) { - if (status.isDir() && !status.getPath().getName().startsWith(".")) { - LOG.warn( - "Format table {} has no partitions registered in the catalog " - + "but its location {} contains directories. Data written " - + "before enabling catalog-managed partitions (or by clients " - + "that do not register partitions) is invisible until the " - + "partition metadata is synced, e.g. with MSCK REPAIR TABLE.", - table.fullName(), - table.location()); - return; - } - } - } catch (IOException ignored) { - // Best-effort hint only; never fail or slow down the scan because of it. - } + return splitEnumerator.findPartitions(partitionFilter); } protected static List, Path>> generatePartitions( @@ -461,7 +221,7 @@ static Optional extractPartitionPredicate( Optional childPredicate = extractPartitionPredicate(child); childPredicate.ifPresent(predicates::add); // Skip children that can't be expressed as Predicate (e.g. Multiple); - // they are still applied by partitionFilter.test() in plan(). + // they are still applied before listing the partition files. } return predicates.isEmpty() ? Optional.empty() @@ -496,66 +256,6 @@ protected static Pair computeScanPathAndLevel( return Pair.of(scanPath, level); } - private List createSplits(FileIO fileIO, Path path, BinaryRow partition) - throws IOException { - List segments = new ArrayList<>(); - FileStatus[] files = fileIO.listFiles(path, true); - Arrays.sort(files, Comparator.comparing(file -> file.getPath().toString())); - for (FileStatus file : files) { - if (isDataFileName(file.getPath().getName())) { - segments.addAll(toSegments(file)); - } - } - - List splits = new ArrayList<>(); - for (List bin : - BinPacking.packForOrdered( - segments, - file -> Math.max(file.readSize(), openFileCost), - targetSplitSize)) { - splits.add(new FormatDataSplit(bin, partition)); - } - return splits; - } - - private List toSegments(FileStatus file) { - if (!preferToSplitFile(file)) { - return Collections.singletonList( - new FormatDataSplit.FileMeta(file.getPath(), file.getLen())); - } - List segments = new ArrayList<>(); - long remainingBytes = file.getLen(); - long currentStart = 0; - - while (remainingBytes > 0) { - long splitSize = Math.min(targetSplitSize, remainingBytes); - segments.add( - new FormatDataSplit.FileMeta( - file.getPath(), file.getLen(), currentStart, splitSize)); - currentStart += splitSize; - remainingBytes -= splitSize; - } - return segments; - } - - private boolean preferToSplitFile(FileStatus file) { - if (file.getLen() <= targetSplitSize) { - return false; - } - - Options options = coreOptions.toConfiguration(); - switch (format) { - case CSV: - return !isCompressed(file.getPath()) - && isDefaultDelimiter(options.get(CsvOptions.LINE_DELIMITER)); - case JSON: - return !isCompressed(file.getPath()) - && isDefaultDelimiter(options.get(JsonOptions.LINE_DELIMITER)); - default: - return false; - } - } - public static Map extractLeadingEqualityPartitionSpecWhenOnlyAnd( List partitionKeys, Predicate predicate, RowType partitionType) { List predicates = PredicateBuilder.splitAnd(predicate); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java new file mode 100644 index 000000000000..149d617ce551 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java @@ -0,0 +1,376 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.format.csv.CsvOptions; +import org.apache.paimon.format.json.JsonOptions; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.options.Options; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.partition.PartitionPredicate.MultiplePartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BinPacking; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.SemaphoredDelegatingExecutor; +import org.apache.paimon.utils.ThreadPoolUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.function.Function; + +import static org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed; +import static org.apache.paimon.format.text.TextLineReader.isDefaultDelimiter; +import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow; +import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; + +/** Enumerates {@link FormatDataSplit}s for a {@link FormatTable}. */ +final class FormatTableSplitEnumerator { + + private static final Logger LOG = LoggerFactory.getLogger(FormatTableSplitEnumerator.class); + + private static final int LIST_POOL_MAX_THREADS = 1000; + private static final ThreadPoolExecutor LIST_POOL = + ThreadPoolUtils.createCachedThreadPool( + LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); + + private final FormatTable table; + private final CoreOptions coreOptions; + @Nullable private final FormatTablePartitionManager partitionManager; + private final long targetSplitSize; + private final long openFileCost; + private final FormatTable.Format format; + + FormatTableSplitEnumerator( + FormatTable table, + CoreOptions coreOptions, + @Nullable FormatTablePartitionManager partitionManager) { + this.table = table; + this.coreOptions = coreOptions; + this.partitionManager = partitionManager; + this.targetSplitSize = coreOptions.splitTargetSize(); + this.openFileCost = coreOptions.splitOpenFileCost(); + this.format = table.format(); + } + + List enumerate(@Nullable PartitionPredicate partitionFilter) throws IOException { + if (table.partitionKeys().isEmpty()) { + return createSplits(table.fileIO(), new Path(table.location()), null); + } + if (partitionManager != null) { + return enumerateCatalogPartitions(partitionManager, partitionFilter); + } + return enumerateFilesystemPartitions(partitionFilter); + } + + List, Path>> findPartitions( + @Nullable PartitionPredicate partitionFilter) { + if (partitionManager != null) { + return findCatalogPartitions(partitionManager, partitionFilter); + } + return findFilesystemPartitions(partitionFilter); + } + + BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { + RowType partitionType = table.partitionType(); + GenericRow row = + convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName()); + return new InternalRowSerializer(partitionType).toBinaryRow(row); + } + + LinkedHashMap normalizeSpec( + @Nullable Map spec, boolean onlyValueInPath) { + List partitionKeys = table.partitionKeys(); + if (spec == null + || spec.size() != partitionKeys.size() + || !spec.keySet().containsAll(partitionKeys)) { + throw corruptPartitionSpec(spec); + } + LinkedHashMap normalized = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + String value = spec.get(partitionKey); + // In a value-only layout, "." and ".." would resolve outside the table. + try { + PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath); + } catch (IllegalArgumentException e) { + throw corruptPartitionSpec(spec); + } + normalized.put(partitionKey, value); + } + return normalized; + } + + void warnIfFilesystemPartitionsExist() { + try { + for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) { + if (status.isDir() && !status.getPath().getName().startsWith(".")) { + LOG.warn( + "Format table {} has no partitions registered in the catalog " + + "but its location {} contains directories. Data written " + + "before enabling catalog-managed partitions (or by clients " + + "that do not register partitions) is invisible until the " + + "partition metadata is synced, e.g. with MSCK REPAIR TABLE.", + table.fullName(), + table.location()); + return; + } + } + } catch (IOException ignored) { + // Best-effort hint only; never fail or slow down the scan because of it. + } + } + + private List enumerateFilesystemPartitions(@Nullable PartitionPredicate partitionFilter) + throws IOException { + List splits = new ArrayList<>(); + FileIO fileIO = table.fileIO(); + for (Pair, Path> pair : + findFilesystemPartitions(partitionFilter)) { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter == null || partitionFilter.test(partitionRow)) { + splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); + } + } + return splits; + } + + private List enumerateCatalogPartitions( + FormatTablePartitionManager partitionManager, + @Nullable PartitionPredicate partitionFilter) { + List, Path>> partitions = + findCatalogPartitions(partitionManager, partitionFilter); + List splits = new ArrayList<>(); + if (partitions.isEmpty()) { + return splits; + } + + FileIO fileIO = table.fileIO(); + Function, Path>, List> lister = + pair -> { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter != null && !partitionFilter.test(partitionRow)) { + return Collections.emptyList(); + } + try { + return createSplits(fileIO, pair.getValue(), partitionRow); + } catch (FileNotFoundException e) { + warnMissingPartition(pair.getKey(), pair.getValue()); + return Collections.emptyList(); + } catch (IOException e) { + throw new RuntimeException( + "Failed to list files for partition " + pair.getValue(), e); + } + }; + int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); + ExecutorService executor = + parallelism >= LIST_POOL_MAX_THREADS + ? LIST_POOL + : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); + ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) + .forEachRemaining(splits::add); + return splits; + } + + private List, Path>> findCatalogPartitions( + FormatTablePartitionManager partitionManager, + @Nullable PartitionPredicate partitionFilter) { + Optional extracted = FormatTableScan.extractPartitionPredicate(partitionFilter); + Map prefix = leadingEqualityPrefix(extracted); + Predicate catalogFilter = extracted.orElse(null); + List partitions = partitionManager.listPartitions(prefix, catalogFilter); + if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) { + warnIfFilesystemPartitionsExist(); + } + return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); + } + + private List, Path>> findFilesystemPartitions( + @Nullable PartitionPredicate partitionFilter) { + LOG.debug( + "Find partitions for format table {}, partition filter: {}", + table.name(), + partitionFilter); + boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); + if (partitionFilter instanceof MultiplePartitionPredicate) { + Set partitions = ((MultiplePartitionPredicate) partitionFilter).partitions(); + return FormatTableScan.generatePartitions( + table.partitionKeys(), + table.partitionType(), + table.defaultPartName(), + new Path(table.location()), + partitions, + onlyValueInPath); + } + + Optional predicate = FormatTableScan.extractPartitionPredicate(partitionFilter); + LOG.debug( + "Extracted predicate for format table {} partition pruning: {}", + table.name(), + predicate.orElse(null)); + + Pair scanPathAndLevel = + FormatTableScan.computeScanPathAndLevel( + new Path(table.location()), + table.partitionKeys(), + predicate, + table.partitionType(), + onlyValueInPath); + return searchPartSpecAndPaths( + table.fileIO(), + scanPathAndLevel.getLeft(), + scanPathAndLevel.getRight(), + table.partitionKeys(), + onlyValueInPath, + predicate.orElse(null), + table.partitionType(), + table.defaultPartName()); + } + + private List, Path>> toSpecsAndPaths( + List partitions, boolean onlyValueInPath) { + List, Path>> result = new ArrayList<>(partitions.size()); + Path tablePath = new Path(table.location()); + // A duplicate catalog entry must not duplicate all records in that partition. + Set seenPartitionPaths = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { + LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath); + String partitionPath = + PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath); + if (seenPartitionPaths.add(partitionPath)) { + result.add(Pair.of(spec, new Path(tablePath, partitionPath))); + } + } + return result; + } + + private IllegalStateException corruptPartitionSpec(@Nullable Map spec) { + return new IllegalStateException( + String.format( + "Catalog returned corrupt partition metadata %s for format table %s; " + + "expected exactly the partition keys %s with values usable as " + + "path components.", + spec, table.fullName(), table.partitionKeys())); + } + + private Map leadingEqualityPrefix(Optional predicate) { + if (!predicate.isPresent()) { + return Collections.emptyMap(); + } + return FormatTableScan.extractLeadingEqualityPartitionSpecWhenOnlyAnd( + table.partitionKeys(), predicate.get(), table.partitionType()); + } + + private List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition) + throws IOException { + List segments = new ArrayList<>(); + FileStatus[] files = fileIO.listFiles(path, true); + Arrays.sort(files, Comparator.comparing(file -> file.getPath().toString())); + for (FileStatus file : files) { + if (FormatTableScan.isDataFileName(file.getPath().getName())) { + segments.addAll(toSegments(file)); + } + } + + List splits = new ArrayList<>(); + for (List bin : + BinPacking.packForOrdered( + segments, + file -> Math.max(file.readSize(), openFileCost), + targetSplitSize)) { + splits.add(new FormatDataSplit(bin, partition)); + } + return splits; + } + + private List toSegments(FileStatus file) { + if (!preferToSplitFile(file)) { + return Collections.singletonList( + new FormatDataSplit.FileMeta(file.getPath(), file.getLen())); + } + List segments = new ArrayList<>(); + long remainingBytes = file.getLen(); + long currentStart = 0; + + while (remainingBytes > 0) { + long splitSize = Math.min(targetSplitSize, remainingBytes); + segments.add( + new FormatDataSplit.FileMeta( + file.getPath(), file.getLen(), currentStart, splitSize)); + currentStart += splitSize; + remainingBytes -= splitSize; + } + return segments; + } + + private boolean preferToSplitFile(FileStatus file) { + if (file.getLen() <= targetSplitSize) { + return false; + } + + Options options = coreOptions.toConfiguration(); + switch (format) { + case CSV: + return !isCompressed(file.getPath()) + && isDefaultDelimiter(options.get(CsvOptions.LINE_DELIMITER)); + case JSON: + return !isCompressed(file.getPath()) + && isDefaultDelimiter(options.get(JsonOptions.LINE_DELIMITER)); + default: + return false; + } + } + + private void warnMissingPartition(LinkedHashMap spec, Path path) { + LOG.warn( + "Partition '{}' of format table {} is registered in the catalog but its directory " + + "'{}' does not exist; treating the partition as empty. If the directory " + + "was removed on purpose, drop the partition or repair the metadata, e.g. " + + "with MSCK REPAIR TABLE.", + PartitionPathUtils.generatePartitionName(spec, false), + table.fullName(), + path); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java index b97f94cac30e..0d86bbb1f178 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java @@ -53,6 +53,9 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; @@ -509,8 +512,8 @@ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { } @Test - void testInternalParallelListingMatchesSerialAndIsDeterministic() throws Exception { - TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + void testCatalogPartitionListingRunsInParallelAndPreservesOrder() throws Exception { + ParallelTrackingLocalFileIO fileIO = new ParallelTrackingLocalFileIO(); Path tablePath = new Path(tempDir.toUri()); List partitions = new ArrayList<>(); List expected = new ArrayList<>(); @@ -519,29 +522,22 @@ void testInternalParallelListingMatchesSerialAndIsDeterministic() throws Excepti expected.add(writeDataFile(fileIO, tablePath, "year=2025/month=" + m)); } - List serial = - plannedFiles( - new FormatTableScan( - managedStringTable(fileIO, tablePath, partitions, 1), - null, - null) - .plan() - .splits()); List parallel = plannedFiles( new FormatTableScan( - managedStringTable(fileIO, tablePath, partitions, 8), + stringPartitionTable( + fileIO, tablePath, recordingCatalog(partitions), 8), null, null) .plan() .splits()); - assertThat(parallel).containsExactlyElementsOf(serial); assertThat(parallel).containsExactlyElementsOf(expected); + assertThat(fileIO.maxConcurrentListings()).isGreaterThan(1); } @Test - void testInternalMissingPartitionDirectoryTreatedAsEmpty() throws Exception { + void testParallelListingSkipsMissingCatalogRegisteredPartition() throws Exception { InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); Path tablePath = new Path(tempDir.toUri()); List partitions = @@ -549,13 +545,13 @@ void testInternalMissingPartitionDirectoryTreatedAsEmpty() throws Exception { partition("2025", "10"), partition("2025", "11"), partition("2025", "12")); Path oct = writeDataFile(fileIO, tablePath, "year=2025/month=10"); Path dec = writeDataFile(fileIO, tablePath, "year=2025/month=12"); - // month=11 is registered but its directory is missing -> FileNotFound -> treated as empty. fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing")); List files = plannedFiles( new FormatTableScan( - managedStringTable(fileIO, tablePath, partitions, 4), + stringPartitionTable( + fileIO, tablePath, recordingCatalog(partitions), 4), null, null) .plan() @@ -565,37 +561,58 @@ void testInternalMissingPartitionDirectoryTreatedAsEmpty() throws Exception { } @Test - void testInternalListingIoErrorFailsWholeScan() throws Exception { + void testCatalogPartitionListingIoErrorFailsWholeScan() throws Exception { InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); Path tablePath = new Path(tempDir.toUri()); List partitions = Arrays.asList(partition("2025", "10"), partition("2025", "11")); writeDataFile(fileIO, tablePath, "year=2025/month=10"); writeDataFile(fileIO, tablePath, "year=2025/month=11"); - // A real (non-FileNotFound) listing error must fail the whole scan, not truncate. fileIO.failListFilesContaining("month=11", new IOException("boom")); - FormatTable table = managedStringTable(fileIO, tablePath, partitions, 4); + FormatTable table = + stringPartitionTable(fileIO, tablePath, recordingCatalog(partitions), 4); assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits()) - .isInstanceOf(RuntimeException.class); + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(IOException.class) + .hasRootCauseMessage("boom"); } @Test - void testExternalMissingDirectoryStillFailsScan() throws Exception { + void testFilesystemDiscoveredMissingDirectoryStillFailsScan() throws Exception { InjectingLocalFileIO fileIO = new InjectingLocalFileIO(); Path tablePath = new Path(tempDir.toUri()); writeDataFile(fileIO, tablePath, "year=2025/month=10"); writeDataFile(fileIO, tablePath, "year=2025/month=11"); - // External tables (partitionManager == null) keep the rethrow-on-missing behavior. fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing")); FormatTable table = createStringPartitionTable(fileIO, tablePath, null); assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits()) - .isInstanceOf(RuntimeException.class); + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(FileNotFoundException.class); + } + + @Test + void testListParallelismDoesNotChangeFilesystemDiscoveredScanning() throws Exception { + SerialTrackingLocalFileIO fileIO = new SerialTrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + List expected = new ArrayList<>(); + for (int month = 10; month <= 12; month++) { + expected.add(writeDataFile(fileIO, tablePath, "year=2025/month=" + month)); + } + + FormatTable table = stringPartitionTable(fileIO, tablePath, null, 8); + List files = plannedFiles(new FormatTableScan(table, null, null).plan().splits()); + + assertThat(files).containsExactlyInAnyOrderElementsOf(expected); + assertThat(fileIO.maxConcurrentListings()).isOne(); } - private FormatTable managedStringTable( - LocalFileIO fileIO, Path tablePath, List partitions, int parallelism) { + private FormatTable stringPartitionTable( + LocalFileIO fileIO, + Path tablePath, + @Nullable FormatTablePartitionManager partitionManager, + int parallelism) { RowType rowType = RowType.builder() .field("year", DataTypes.STRING()) @@ -613,10 +630,64 @@ private FormatTable managedStringTable( .location(tablePath.toString()) .format(FormatTable.Format.CSV) .options(options) - .partitionManager(recordingCatalog(partitions)) + .partitionManager(partitionManager) .build(); } + private static class ParallelTrackingLocalFileIO extends LocalFileIO { + + private final CountDownLatch concurrentListings = new CountDownLatch(2); + private final AtomicInteger activeListings = new AtomicInteger(); + private final AtomicInteger maxConcurrentListings = new AtomicInteger(); + + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + int active = activeListings.incrementAndGet(); + maxConcurrentListings.updateAndGet(current -> Math.max(current, active)); + concurrentListings.countDown(); + try { + if (!concurrentListings.await(10, TimeUnit.SECONDS)) { + throw new IOException("Partition file listings did not run concurrently"); + } + return super.listFiles(path, recursive); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for concurrent listings", e); + } finally { + activeListings.decrementAndGet(); + } + } + + private int maxConcurrentListings() { + return maxConcurrentListings.get(); + } + } + + private static class SerialTrackingLocalFileIO extends LocalFileIO { + + private final AtomicInteger activeListings = new AtomicInteger(); + private final AtomicInteger maxConcurrentListings = new AtomicInteger(); + + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + int active = activeListings.incrementAndGet(); + maxConcurrentListings.updateAndGet(current -> Math.max(current, active)); + try { + Thread.sleep(50); + return super.listFiles(path, recursive); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while tracking file listings", e); + } finally { + activeListings.decrementAndGet(); + } + } + + private int maxConcurrentListings() { + return maxConcurrentListings.get(); + } + } + private static class InjectingLocalFileIO extends LocalFileIO { private final Map listFailures = new LinkedHashMap<>(); From ba680293472df50437b90cff6a1fce30cb064bee Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 02:57:56 -0700 Subject: [PATCH 10/15] [core] Split format table enumerator into catalog and filesystem impls Per review, replace the partitionManager-based branching inside the single FormatTableSplitEnumerator with two implementations chosen once at construction by FormatTableSplitEnumerator.create: CatalogFormatTableSplitEnumerator (catalog-managed partitions, parallel listing) and FileSystemFormatTableSplitEnumerator (filesystem discovery, serial). enumerate/findPartitions/listPartitionEntries are now polymorphic, so the per-call 'partitionManager != null' if/else is gone. --- .../CatalogFormatTableSplitEnumerator.java | 237 +++++++++++++++ .../FileSystemFormatTableSplitEnumerator.java | 133 +++++++++ .../paimon/table/format/FormatTableScan.java | 53 +--- .../format/FormatTableSplitEnumerator.java | 278 +++--------------- 4 files changed, 408 insertions(+), 293 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java new file mode 100644 index 000000000000..0a1ef66577e7 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.SemaphoredDelegatingExecutor; +import org.apache.paimon.utils.ThreadPoolUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.function.Function; + +/** A {@link FormatTableSplitEnumerator} whose partitions are managed by the catalog. */ +final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator { + + private static final Logger LOG = + LoggerFactory.getLogger(CatalogFormatTableSplitEnumerator.class); + + private static final int LIST_POOL_MAX_THREADS = 1000; + private static final ThreadPoolExecutor LIST_POOL = + ThreadPoolUtils.createCachedThreadPool( + LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); + + private final FormatTablePartitionManager partitionManager; + + CatalogFormatTableSplitEnumerator( + FormatTable table, + CoreOptions coreOptions, + FormatTablePartitionManager partitionManager) { + super(table, coreOptions); + this.partitionManager = partitionManager; + } + + @Override + List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) { + List, Path>> partitions = + findPartitions(partitionFilter); + List splits = new ArrayList<>(); + if (partitions.isEmpty()) { + return splits; + } + + FileIO fileIO = table.fileIO(); + Function, Path>, List> lister = + pair -> { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter != null && !partitionFilter.test(partitionRow)) { + return Collections.emptyList(); + } + try { + return createSplits(fileIO, pair.getValue(), partitionRow); + } catch (FileNotFoundException e) { + warnMissingPartition(pair.getKey(), pair.getValue()); + return Collections.emptyList(); + } catch (IOException e) { + throw new RuntimeException( + "Failed to list files for partition " + pair.getValue(), e); + } + }; + int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); + ExecutorService executor = + parallelism >= LIST_POOL_MAX_THREADS + ? LIST_POOL + : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); + ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) + .forEachRemaining(splits::add); + return splits; + } + + @Override + List, Path>> findPartitions( + @Nullable PartitionPredicate partitionFilter) { + Optional extracted = FormatTableScan.extractPartitionPredicate(partitionFilter); + Map prefix = leadingEqualityPrefix(extracted); + Predicate catalogFilter = extracted.orElse(null); + List partitions = partitionManager.listPartitions(prefix, catalogFilter); + if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) { + warnIfFilesystemPartitionsExist(); + } + return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); + } + + @Override + List listPartitionEntries() { + List partitions = partitionManager.listPartitions(Collections.emptyMap(), null); + if (partitions.isEmpty()) { + warnIfFilesystemPartitionsExist(); + } + boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); + List entries = new ArrayList<>(partitions.size()); + Set> seen = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { + if (!seen.add(partition.spec())) { + continue; + } + entries.add( + new PartitionEntry( + toPartitionRow(normalizeSpec(partition.spec(), onlyValueInPath)), + partition.recordCount(), + partition.fileSizeInBytes(), + partition.fileCount(), + partition.lastFileCreationTime(), + partition.totalBuckets())); + } + return entries; + } + + private List, Path>> toSpecsAndPaths( + List partitions, boolean onlyValueInPath) { + List, Path>> result = new ArrayList<>(partitions.size()); + Path tablePath = new Path(table.location()); + // A duplicate catalog entry must not duplicate all records in that partition. + Set seenPartitionPaths = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { + LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath); + String partitionPath = + PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath); + if (seenPartitionPaths.add(partitionPath)) { + result.add(Pair.of(spec, new Path(tablePath, partitionPath))); + } + } + return result; + } + + LinkedHashMap normalizeSpec( + @Nullable Map spec, boolean onlyValueInPath) { + List partitionKeys = table.partitionKeys(); + if (spec == null + || spec.size() != partitionKeys.size() + || !spec.keySet().containsAll(partitionKeys)) { + throw corruptPartitionSpec(spec); + } + LinkedHashMap normalized = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + String value = spec.get(partitionKey); + // In a value-only layout, "." and ".." would resolve outside the table. + try { + PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath); + } catch (IllegalArgumentException e) { + throw corruptPartitionSpec(spec); + } + normalized.put(partitionKey, value); + } + return normalized; + } + + void warnIfFilesystemPartitionsExist() { + try { + for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) { + if (status.isDir() && !status.getPath().getName().startsWith(".")) { + LOG.warn( + "Format table {} has no partitions registered in the catalog " + + "but its location {} contains directories. Data written " + + "before enabling catalog-managed partitions (or by clients " + + "that do not register partitions) is invisible until the " + + "partition metadata is synced, e.g. with MSCK REPAIR TABLE.", + table.fullName(), + table.location()); + return; + } + } + } catch (IOException ignored) { + // Best-effort hint only; never fail or slow down the scan because of it. + } + } + + private Map leadingEqualityPrefix(Optional predicate) { + if (!predicate.isPresent()) { + return Collections.emptyMap(); + } + return FormatTableScan.extractLeadingEqualityPartitionSpecWhenOnlyAnd( + table.partitionKeys(), predicate.get(), table.partitionType()); + } + + private IllegalStateException corruptPartitionSpec(@Nullable Map spec) { + return new IllegalStateException( + String.format( + "Catalog returned corrupt partition metadata %s for format table %s; " + + "expected exactly the partition keys %s with values usable as " + + "path components.", + spec, table.fullName(), table.partitionKeys())); + } + + private void warnMissingPartition(LinkedHashMap spec, Path path) { + LOG.warn( + "Partition '{}' of format table {} is registered in the catalog but its directory " + + "'{}' does not exist; treating the partition as empty. If the directory " + + "was removed on purpose, drop the partition or repair the metadata, e.g. " + + "with MSCK REPAIR TABLE.", + PartitionPathUtils.generatePartitionName(spec, false), + table.fullName(), + path); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java new file mode 100644 index 000000000000..775bc8d76cbf --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.partition.PartitionPredicate.MultiplePartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.utils.Pair; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; + +/** A {@link FormatTableSplitEnumerator} whose partitions are discovered from the filesystem. */ +final class FileSystemFormatTableSplitEnumerator extends FormatTableSplitEnumerator { + + private static final Logger LOG = + LoggerFactory.getLogger(FileSystemFormatTableSplitEnumerator.class); + + FileSystemFormatTableSplitEnumerator(FormatTable table, CoreOptions coreOptions) { + super(table, coreOptions); + } + + @Override + List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) + throws IOException { + List splits = new ArrayList<>(); + FileIO fileIO = table.fileIO(); + for (Pair, Path> pair : findPartitions(partitionFilter)) { + BinaryRow partitionRow = toPartitionRow(pair.getKey()); + if (partitionFilter == null || partitionFilter.test(partitionRow)) { + splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); + } + } + return splits; + } + + @Override + List, Path>> findPartitions( + @Nullable PartitionPredicate partitionFilter) { + LOG.debug( + "Find partitions for format table {}, partition filter: {}", + table.name(), + partitionFilter); + boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); + if (partitionFilter instanceof MultiplePartitionPredicate) { + Set partitions = ((MultiplePartitionPredicate) partitionFilter).partitions(); + return FormatTableScan.generatePartitions( + table.partitionKeys(), + table.partitionType(), + table.defaultPartName(), + new Path(table.location()), + partitions, + onlyValueInPath); + } + + Optional predicate = FormatTableScan.extractPartitionPredicate(partitionFilter); + LOG.debug( + "Extracted predicate for format table {} partition pruning: {}", + table.name(), + predicate.orElse(null)); + + Pair scanPathAndLevel = + FormatTableScan.computeScanPathAndLevel( + new Path(table.location()), + table.partitionKeys(), + predicate, + table.partitionType(), + onlyValueInPath); + return searchPartSpecAndPaths( + table.fileIO(), + scanPathAndLevel.getLeft(), + scanPathAndLevel.getRight(), + table.partitionKeys(), + onlyValueInPath, + predicate.orElse(null), + table.partitionType(), + table.defaultPartName()); + } + + @Override + List listPartitionEntries() { + List, Path>> partition2Paths = + searchPartSpecAndPaths( + table.fileIO(), + new Path(table.location()), + table.partitionKeys().size(), + table.partitionKeys(), + coreOptions.formatTablePartitionOnlyValueInPath(), + null, + table.partitionType(), + table.defaultPartName()); + List partitionEntries = new ArrayList<>(); + for (Pair, Path> partition2Path : partition2Paths) { + BinaryRow row = toPartitionRow(partition2Path.getKey()); + partitionEntries.add(new PartitionEntry(row, -1L, -1L, -1L, -1L, -1)); + } + return partitionEntries; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 90003555d5df..e9be09843d33 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -25,7 +25,6 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.PartitionEntry; -import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.partition.PartitionPredicate.AndPartitionPredicate; import org.apache.paimon.partition.PartitionPredicate.DefaultPartitionPredicate; @@ -51,24 +50,19 @@ import java.io.IOException; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; -import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; - /** {@link TableScan} for {@link FormatTable}. */ public class FormatTableScan implements InnerTableScan { final FormatTable table; final CoreOptions coreOptions; @Nullable private PartitionPredicate partitionFilter; - @Nullable private final FormatTablePartitionManager partitionManager; private final FormatTableSplitEnumerator splitEnumerator; @Nullable private final Integer limit; @@ -80,8 +74,8 @@ public FormatTableScan( this.coreOptions = new CoreOptions(table.options()); this.partitionFilter = partitionFilter; this.limit = limit; - this.partitionManager = table.partitionManager(); - this.splitEnumerator = new FormatTableSplitEnumerator(table, coreOptions, partitionManager); + this.splitEnumerator = + FormatTableSplitEnumerator.create(table, coreOptions, table.partitionManager()); } @Override @@ -97,48 +91,7 @@ public Plan plan() { @Override public List listPartitionEntries() { - if (partitionManager != null) { - List partitions = - partitionManager.listPartitions(Collections.emptyMap(), null); - if (partitions.isEmpty()) { - splitEnumerator.warnIfFilesystemPartitionsExist(); - } - boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); - List entries = new ArrayList<>(partitions.size()); - Set> seen = new HashSet<>(partitions.size()); - for (Partition partition : partitions) { - if (!seen.add(partition.spec())) { - continue; - } - entries.add( - new PartitionEntry( - toPartitionRow( - splitEnumerator.normalizeSpec( - partition.spec(), onlyValueInPath)), - partition.recordCount(), - partition.fileSizeInBytes(), - partition.fileCount(), - partition.lastFileCreationTime(), - partition.totalBuckets())); - } - return entries; - } - List, Path>> partition2Paths = - searchPartSpecAndPaths( - table.fileIO(), - new Path(table.location()), - table.partitionKeys().size(), - table.partitionKeys(), - coreOptions.formatTablePartitionOnlyValueInPath(), - null, - table.partitionType(), - table.defaultPartName()); - List partitionEntries = new ArrayList<>(); - for (Pair, Path> partition2Path : partition2Paths) { - BinaryRow row = toPartitionRow(partition2Path.getKey()); - partitionEntries.add(new PartitionEntry(row, -1L, -1L, -1L, -1L, -1)); - } - return partitionEntries; + return splitEnumerator.listPartitionEntries(); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java index 149d617ce551..79d1431b688f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java @@ -27,93 +27,78 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.options.Options; -import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionPredicate; -import org.apache.paimon.partition.PartitionPredicate.MultiplePartitionPredicate; -import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.source.Split; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.BinPacking; import org.apache.paimon.utils.Pair; -import org.apache.paimon.utils.PartitionPathUtils; -import org.apache.paimon.utils.SemaphoredDelegatingExecutor; -import org.apache.paimon.utils.ThreadPoolUtils; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; -import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; -import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.function.Function; import static org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed; import static org.apache.paimon.format.text.TextLineReader.isDefaultDelimiter; import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow; -import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; - -/** Enumerates {@link FormatDataSplit}s for a {@link FormatTable}. */ -final class FormatTableSplitEnumerator { - private static final Logger LOG = LoggerFactory.getLogger(FormatTableSplitEnumerator.class); - - private static final int LIST_POOL_MAX_THREADS = 1000; - private static final ThreadPoolExecutor LIST_POOL = - ThreadPoolUtils.createCachedThreadPool( - LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); +/** + * Enumerates {@link FormatDataSplit}s for a {@link FormatTable}. + * + *

Partitions come either from the catalog ({@link CatalogFormatTableSplitEnumerator}) or from + * the filesystem ({@link FileSystemFormatTableSplitEnumerator}); the source is chosen once by + * {@link #create} so the enumeration paths stay free of {@code partitionManager != null} branching. + */ +abstract class FormatTableSplitEnumerator { - private final FormatTable table; - private final CoreOptions coreOptions; - @Nullable private final FormatTablePartitionManager partitionManager; - private final long targetSplitSize; - private final long openFileCost; - private final FormatTable.Format format; + protected final FormatTable table; + protected final CoreOptions coreOptions; + protected final long targetSplitSize; + protected final long openFileCost; + protected final FormatTable.Format format; - FormatTableSplitEnumerator( - FormatTable table, - CoreOptions coreOptions, - @Nullable FormatTablePartitionManager partitionManager) { + FormatTableSplitEnumerator(FormatTable table, CoreOptions coreOptions) { this.table = table; this.coreOptions = coreOptions; - this.partitionManager = partitionManager; this.targetSplitSize = coreOptions.splitTargetSize(); this.openFileCost = coreOptions.splitOpenFileCost(); this.format = table.format(); } - List enumerate(@Nullable PartitionPredicate partitionFilter) throws IOException { - if (table.partitionKeys().isEmpty()) { - return createSplits(table.fileIO(), new Path(table.location()), null); - } + static FormatTableSplitEnumerator create( + FormatTable table, + CoreOptions coreOptions, + @Nullable FormatTablePartitionManager partitionManager) { if (partitionManager != null) { - return enumerateCatalogPartitions(partitionManager, partitionFilter); + return new CatalogFormatTableSplitEnumerator(table, coreOptions, partitionManager); } - return enumerateFilesystemPartitions(partitionFilter); + return new FileSystemFormatTableSplitEnumerator(table, coreOptions); } - List, Path>> findPartitions( - @Nullable PartitionPredicate partitionFilter) { - if (partitionManager != null) { - return findCatalogPartitions(partitionManager, partitionFilter); + final List enumerate(@Nullable PartitionPredicate partitionFilter) throws IOException { + if (table.partitionKeys().isEmpty()) { + return createSplits(table.fileIO(), new Path(table.location()), null); } - return findFilesystemPartitions(partitionFilter); + return enumeratePartitions(partitionFilter); } + /** Enumerate splits for a partitioned table; partitions come from the concrete source. */ + abstract List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) + throws IOException; + + abstract List, Path>> findPartitions( + @Nullable PartitionPredicate partitionFilter); + + abstract List listPartitionEntries(); + BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { RowType partitionType = table.partitionType(); GenericRow row = @@ -121,189 +106,7 @@ BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { return new InternalRowSerializer(partitionType).toBinaryRow(row); } - LinkedHashMap normalizeSpec( - @Nullable Map spec, boolean onlyValueInPath) { - List partitionKeys = table.partitionKeys(); - if (spec == null - || spec.size() != partitionKeys.size() - || !spec.keySet().containsAll(partitionKeys)) { - throw corruptPartitionSpec(spec); - } - LinkedHashMap normalized = new LinkedHashMap<>(); - for (String partitionKey : partitionKeys) { - String value = spec.get(partitionKey); - // In a value-only layout, "." and ".." would resolve outside the table. - try { - PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath); - } catch (IllegalArgumentException e) { - throw corruptPartitionSpec(spec); - } - normalized.put(partitionKey, value); - } - return normalized; - } - - void warnIfFilesystemPartitionsExist() { - try { - for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) { - if (status.isDir() && !status.getPath().getName().startsWith(".")) { - LOG.warn( - "Format table {} has no partitions registered in the catalog " - + "but its location {} contains directories. Data written " - + "before enabling catalog-managed partitions (or by clients " - + "that do not register partitions) is invisible until the " - + "partition metadata is synced, e.g. with MSCK REPAIR TABLE.", - table.fullName(), - table.location()); - return; - } - } - } catch (IOException ignored) { - // Best-effort hint only; never fail or slow down the scan because of it. - } - } - - private List enumerateFilesystemPartitions(@Nullable PartitionPredicate partitionFilter) - throws IOException { - List splits = new ArrayList<>(); - FileIO fileIO = table.fileIO(); - for (Pair, Path> pair : - findFilesystemPartitions(partitionFilter)) { - BinaryRow partitionRow = toPartitionRow(pair.getKey()); - if (partitionFilter == null || partitionFilter.test(partitionRow)) { - splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); - } - } - return splits; - } - - private List enumerateCatalogPartitions( - FormatTablePartitionManager partitionManager, - @Nullable PartitionPredicate partitionFilter) { - List, Path>> partitions = - findCatalogPartitions(partitionManager, partitionFilter); - List splits = new ArrayList<>(); - if (partitions.isEmpty()) { - return splits; - } - - FileIO fileIO = table.fileIO(); - Function, Path>, List> lister = - pair -> { - BinaryRow partitionRow = toPartitionRow(pair.getKey()); - if (partitionFilter != null && !partitionFilter.test(partitionRow)) { - return Collections.emptyList(); - } - try { - return createSplits(fileIO, pair.getValue(), partitionRow); - } catch (FileNotFoundException e) { - warnMissingPartition(pair.getKey(), pair.getValue()); - return Collections.emptyList(); - } catch (IOException e) { - throw new RuntimeException( - "Failed to list files for partition " + pair.getValue(), e); - } - }; - int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); - ExecutorService executor = - parallelism >= LIST_POOL_MAX_THREADS - ? LIST_POOL - : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); - ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) - .forEachRemaining(splits::add); - return splits; - } - - private List, Path>> findCatalogPartitions( - FormatTablePartitionManager partitionManager, - @Nullable PartitionPredicate partitionFilter) { - Optional extracted = FormatTableScan.extractPartitionPredicate(partitionFilter); - Map prefix = leadingEqualityPrefix(extracted); - Predicate catalogFilter = extracted.orElse(null); - List partitions = partitionManager.listPartitions(prefix, catalogFilter); - if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) { - warnIfFilesystemPartitionsExist(); - } - return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); - } - - private List, Path>> findFilesystemPartitions( - @Nullable PartitionPredicate partitionFilter) { - LOG.debug( - "Find partitions for format table {}, partition filter: {}", - table.name(), - partitionFilter); - boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); - if (partitionFilter instanceof MultiplePartitionPredicate) { - Set partitions = ((MultiplePartitionPredicate) partitionFilter).partitions(); - return FormatTableScan.generatePartitions( - table.partitionKeys(), - table.partitionType(), - table.defaultPartName(), - new Path(table.location()), - partitions, - onlyValueInPath); - } - - Optional predicate = FormatTableScan.extractPartitionPredicate(partitionFilter); - LOG.debug( - "Extracted predicate for format table {} partition pruning: {}", - table.name(), - predicate.orElse(null)); - - Pair scanPathAndLevel = - FormatTableScan.computeScanPathAndLevel( - new Path(table.location()), - table.partitionKeys(), - predicate, - table.partitionType(), - onlyValueInPath); - return searchPartSpecAndPaths( - table.fileIO(), - scanPathAndLevel.getLeft(), - scanPathAndLevel.getRight(), - table.partitionKeys(), - onlyValueInPath, - predicate.orElse(null), - table.partitionType(), - table.defaultPartName()); - } - - private List, Path>> toSpecsAndPaths( - List partitions, boolean onlyValueInPath) { - List, Path>> result = new ArrayList<>(partitions.size()); - Path tablePath = new Path(table.location()); - // A duplicate catalog entry must not duplicate all records in that partition. - Set seenPartitionPaths = new HashSet<>(partitions.size()); - for (Partition partition : partitions) { - LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath); - String partitionPath = - PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath); - if (seenPartitionPaths.add(partitionPath)) { - result.add(Pair.of(spec, new Path(tablePath, partitionPath))); - } - } - return result; - } - - private IllegalStateException corruptPartitionSpec(@Nullable Map spec) { - return new IllegalStateException( - String.format( - "Catalog returned corrupt partition metadata %s for format table %s; " - + "expected exactly the partition keys %s with values usable as " - + "path components.", - spec, table.fullName(), table.partitionKeys())); - } - - private Map leadingEqualityPrefix(Optional predicate) { - if (!predicate.isPresent()) { - return Collections.emptyMap(); - } - return FormatTableScan.extractLeadingEqualityPartitionSpecWhenOnlyAnd( - table.partitionKeys(), predicate.get(), table.partitionType()); - } - - private List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition) + List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition) throws IOException { List segments = new ArrayList<>(); FileStatus[] files = fileIO.listFiles(path, true); @@ -362,15 +165,4 @@ private boolean preferToSplitFile(FileStatus file) { return false; } } - - private void warnMissingPartition(LinkedHashMap spec, Path path) { - LOG.warn( - "Partition '{}' of format table {} is registered in the catalog but its directory " - + "'{}' does not exist; treating the partition as empty. If the directory " - + "was removed on purpose, drop the partition or repair the metadata, e.g. " - + "with MSCK REPAIR TABLE.", - PartitionPathUtils.generatePartitionName(spec, false), - table.fullName(), - path); - } } From f06e3fb595fe74721a7581897aa05fae5f3464d8 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 03:38:18 -0700 Subject: [PATCH 11/15] [core] Reuse the shared manifest-read pool for format table listing The dedicated createCachedThreadPool(1000) sets corePoolSize == maximumPoolSize == 1000, so it spawns a new thread for every submitted task until the pool holds 1000 threads rather than reusing idle workers, even though per-scan concurrency is bounded by the semaphore. Listing ~1000 partitions on the coordinator could thus create ~1000 threads and risk 'unable to create native thread'/OOM. Delegate to ManifestReadThreadPool.randomlyExecuteSequentialReturn, the shared IO pool Paimon already uses for scans (sized to availableProcessors, grown only when a larger parallelism is explicitly requested and capped by the format-table.scan.list-parallelism permit). --- .../CatalogFormatTableSplitEnumerator.java | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java index 0a1ef66577e7..1e4d7725d00c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java @@ -29,10 +29,9 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.source.Split; +import org.apache.paimon.utils.ManifestReadThreadPool; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; -import org.apache.paimon.utils.SemaphoredDelegatingExecutor; -import org.apache.paimon.utils.ThreadPoolUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,8 +48,6 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ThreadPoolExecutor; import java.util.function.Function; /** A {@link FormatTableSplitEnumerator} whose partitions are managed by the catalog. */ @@ -59,11 +56,6 @@ final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator private static final Logger LOG = LoggerFactory.getLogger(CatalogFormatTableSplitEnumerator.class); - private static final int LIST_POOL_MAX_THREADS = 1000; - private static final ThreadPoolExecutor LIST_POOL = - ThreadPoolUtils.createCachedThreadPool( - LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); - private final FormatTablePartitionManager partitionManager; CatalogFormatTableSplitEnumerator( @@ -101,11 +93,7 @@ List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) { } }; int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); - ExecutorService executor = - parallelism >= LIST_POOL_MAX_THREADS - ? LIST_POOL - : new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); - ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) + ManifestReadThreadPool.randomlyExecuteSequentialReturn(lister, partitions, parallelism) .forEachRemaining(splits::add); return splits; } From 103010bdfc835dbf4d91d4c6bd77f043b0f592bb Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 04:27:36 -0700 Subject: [PATCH 12/15] [core] Bound the format table listing pool and preserve the caller's filesystem context Address review of the parallel listing: - Size the dedicated pool at 128 (was 1000) and clamp format-table.scan.list-parallelism to [1, 128], so the pool reuses workers and can no longer grow to ~1000 threads or be pushed there by a large config value. Reverts the previous fix that had mutated the shared ManifestReadThreadPool. - Establish the filesystem on the caller thread (fileIO.exists on the table location) before dispatching, so the listing workers reuse a filesystem created under the caller's security context instead of lazily creating and caching it under a shared worker's context. This matches the FileStore scan path, which reads the snapshot on the caller thread before its parallel manifest reads. --- .../CatalogFormatTableSplitEnumerator.java | 25 ++++++++++++++++--- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java index 1e4d7725d00c..fdb720bcff64 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java @@ -29,9 +29,10 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.source.Split; -import org.apache.paimon.utils.ManifestReadThreadPool; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.SemaphoredDelegatingExecutor; +import org.apache.paimon.utils.ThreadPoolUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,6 +49,8 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; import java.util.function.Function; /** A {@link FormatTableSplitEnumerator} whose partitions are managed by the catalog. */ @@ -56,6 +59,11 @@ final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator private static final Logger LOG = LoggerFactory.getLogger(CatalogFormatTableSplitEnumerator.class); + private static final int LIST_POOL_MAX_THREADS = 128; + private static final ThreadPoolExecutor LIST_POOL = + ThreadPoolUtils.createCachedThreadPool( + LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); + private final FormatTablePartitionManager partitionManager; CatalogFormatTableSplitEnumerator( @@ -67,7 +75,8 @@ final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator } @Override - List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) { + List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) + throws IOException { List, Path>> partitions = findPartitions(partitionFilter); List splits = new ArrayList<>(); @@ -76,6 +85,10 @@ List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) { } FileIO fileIO = table.fileIO(); + // Establish the filesystem on the caller thread so the listing workers reuse a filesystem + // created under the caller's security context, matching how a FileStore scan reads the + // snapshot on the caller thread before its parallel manifest reads. + fileIO.exists(new Path(table.location())); Function, Path>, List> lister = pair -> { BinaryRow partitionRow = toPartitionRow(pair.getKey()); @@ -92,8 +105,12 @@ List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) { "Failed to list files for partition " + pair.getValue(), e); } }; - int parallelism = Math.max(1, coreOptions.formatTableScanListParallelism()); - ManifestReadThreadPool.randomlyExecuteSequentialReturn(lister, partitions, parallelism) + int parallelism = + Math.min( + LIST_POOL_MAX_THREADS, + Math.max(1, coreOptions.formatTableScanListParallelism())); + ExecutorService executor = new SemaphoredDelegatingExecutor(LIST_POOL, parallelism, false); + ThreadPoolUtils.randomlyExecuteSequentialReturn(executor, lister, partitions) .forEachRemaining(splits::add); return splits; } From 45fd951a91fdc9a1f3152884fd42b318c7f3ca3f Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 05:26:54 -0700 Subject: [PATCH 13/15] [test] Guard that format table listing establishes the filesystem on the caller thread --- .../CatalogFormatTableSplitEnumerator.java | 5 +- .../CatalogManagedPartitionScanTest.java | 82 +++++++++++++++++++ 2 files changed, 84 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java index fdb720bcff64..1cc31619fc98 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java @@ -85,9 +85,8 @@ List enumeratePartitions(@Nullable PartitionPredicate partitionFilter) } FileIO fileIO = table.fileIO(); - // Establish the filesystem on the caller thread so the listing workers reuse a filesystem - // created under the caller's security context, matching how a FileStore scan reads the - // snapshot on the caller thread before its parallel manifest reads. + // Establish the filesystem on the caller thread so listing workers reuse it under the + // caller's security context instead of creating it lazily under a shared worker. fileIO.exists(new Path(table.location())); Function, Path>, List> lister = pair -> { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java index 0d86bbb1f178..be7d5e4584da 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java @@ -56,6 +56,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; @@ -608,6 +609,34 @@ void testListParallelismDoesNotChangeFilesystemDiscoveredScanning() throws Excep assertThat(fileIO.maxConcurrentListings()).isOne(); } + @Test + void testCatalogListingEstablishesFilesystemOnCallerThread() throws Exception { + CallerBoundLocalFileIO fileIO = new CallerBoundLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + List partitions = new ArrayList<>(); + for (int month = 10; month <= 13; month++) { + partitions.add(partition("2025", Integer.toString(month))); + writeDataFile(fileIO, tablePath, "year=2025/month=" + month); + } + + fileIO.resetBinding(); + CallerBoundLocalFileIO.CALLER_USER.set("caller-user"); + try { + new FormatTableScan( + stringPartitionTable( + fileIO, tablePath, recordingCatalog(partitions), 4), + null, + null) + .plan() + .splits(); + } finally { + CallerBoundLocalFileIO.CALLER_USER.remove(); + } + + assertThat(fileIO.boundUser()).isEqualTo("caller-user"); + assertThat(fileIO.boundOnListingWorker()).isFalse(); + } + private FormatTable stringPartitionTable( LocalFileIO fileIO, Path tablePath, @@ -706,4 +735,57 @@ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { return super.listFiles(path, recursive); } } + + /** Binds to the thread and thread-local user that first touches the filesystem. */ + private static class CallerBoundLocalFileIO extends LocalFileIO { + + static final ThreadLocal CALLER_USER = new ThreadLocal<>(); + + private final AtomicReference boundThread = new AtomicReference<>(); + private final AtomicReference boundUser = new AtomicReference<>(); + + private void bindOnce() { + if (boundThread.compareAndSet(null, Thread.currentThread())) { + boundUser.set(CALLER_USER.get()); + } + } + + void resetBinding() { + boundThread.set(null); + boundUser.set(null); + } + + String boundUser() { + return boundUser.get(); + } + + boolean boundOnListingWorker() { + Thread thread = boundThread.get(); + return thread != null && thread.getName().contains("FORMAT-TABLE-LIST"); + } + + @Override + public boolean exists(Path path) throws IOException { + bindOnce(); + return super.exists(path); + } + + @Override + public FileStatus getFileStatus(Path path) throws IOException { + bindOnce(); + return super.getFileStatus(path); + } + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + bindOnce(); + return super.listStatus(path); + } + + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + bindOnce(); + return super.listFiles(path, recursive); + } + } } From 7fa68f36d3347ef590ae711232e349062a93cd75 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 05:49:03 -0700 Subject: [PATCH 14/15] [core] Restore Trino-style listing pool: max 1000, 64 per scan, reusing workers 128 was an arbitrary cap; revert to the intended model - per-scan concurrency 64 (format-table.scan.list-parallelism) over a shared pool capped at 1000 (Trino's hive.split-loader-concurrency / hive.max-split-iterator-threads). Build the pool as a bounded cached pool (core 0, SynchronousQueue) so it reuses idle workers and grows only on demand, instead of createCachedThreadPool whose corePoolSize == max spawns a new thread per task up to the cap. CallerRunsPolicy applies back pressure once the cap is hit. --- .../CatalogFormatTableSplitEnumerator.java | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java index 1cc31619fc98..638a5d67b58c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java @@ -33,6 +33,7 @@ import org.apache.paimon.utils.PartitionPathUtils; import org.apache.paimon.utils.SemaphoredDelegatingExecutor; import org.apache.paimon.utils.ThreadPoolUtils; +import org.apache.paimon.utils.ThreadUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,7 +51,9 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutorService; +import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.function.Function; /** A {@link FormatTableSplitEnumerator} whose partitions are managed by the catalog. */ @@ -59,10 +62,19 @@ final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator private static final Logger LOG = LoggerFactory.getLogger(CatalogFormatTableSplitEnumerator.class); - private static final int LIST_POOL_MAX_THREADS = 128; + private static final int LIST_POOL_MAX_THREADS = 1000; + + // Cached pool bounded at 1000 threads that reuses idle workers; CallerRunsPolicy lists on the + // caller for back pressure once the cap is hit. private static final ThreadPoolExecutor LIST_POOL = - ThreadPoolUtils.createCachedThreadPool( - LIST_POOL_MAX_THREADS, "FORMAT-TABLE-LIST-THREAD-POOL"); + new ThreadPoolExecutor( + 0, + LIST_POOL_MAX_THREADS, + 1, + TimeUnit.MINUTES, + new SynchronousQueue<>(), + ThreadUtils.newDaemonThreadFactory("FORMAT-TABLE-LIST-THREAD-POOL"), + new ThreadPoolExecutor.CallerRunsPolicy()); private final FormatTablePartitionManager partitionManager; From d20335c6cde211d0e326c645e9e99445627c06b0 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 25 Jul 2026 06:07:12 -0700 Subject: [PATCH 15/15] [core] Name the enumerator SplitEnumerator with Catalog/FileSystem impls per review Rename FormatTableSplitEnumerator -> SplitEnumerator, CatalogFormatTableSplitEnumerator -> CatalogSplitEnumerator, and FileSystemFormatTableSplitEnumerator -> FileSystemSplitEnumerator. --- ...umerator.java => CatalogSplitEnumerator.java} | 9 ++++----- ...rator.java => FileSystemSplitEnumerator.java} | 9 ++++----- .../paimon/table/format/FormatTableScan.java | 5 ++--- ...SplitEnumerator.java => SplitEnumerator.java} | 16 ++++++++-------- 4 files changed, 18 insertions(+), 21 deletions(-) rename paimon-core/src/main/java/org/apache/paimon/table/format/{CatalogFormatTableSplitEnumerator.java => CatalogSplitEnumerator.java} (97%) rename paimon-core/src/main/java/org/apache/paimon/table/format/{FileSystemFormatTableSplitEnumerator.java => FileSystemSplitEnumerator.java} (92%) rename paimon-core/src/main/java/org/apache/paimon/table/format/{FormatTableSplitEnumerator.java => SplitEnumerator.java} (90%) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java similarity index 97% rename from paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java rename to paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java index 638a5d67b58c..eaa43e3d697c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java @@ -56,11 +56,10 @@ import java.util.concurrent.TimeUnit; import java.util.function.Function; -/** A {@link FormatTableSplitEnumerator} whose partitions are managed by the catalog. */ -final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator { +/** A {@link SplitEnumerator} whose partitions are managed by the catalog. */ +final class CatalogSplitEnumerator extends SplitEnumerator { - private static final Logger LOG = - LoggerFactory.getLogger(CatalogFormatTableSplitEnumerator.class); + private static final Logger LOG = LoggerFactory.getLogger(CatalogSplitEnumerator.class); private static final int LIST_POOL_MAX_THREADS = 1000; @@ -78,7 +77,7 @@ final class CatalogFormatTableSplitEnumerator extends FormatTableSplitEnumerator private final FormatTablePartitionManager partitionManager; - CatalogFormatTableSplitEnumerator( + CatalogSplitEnumerator( FormatTable table, CoreOptions coreOptions, FormatTablePartitionManager partitionManager) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java similarity index 92% rename from paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java rename to paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java index 775bc8d76cbf..7373e35d4d4c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemFormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java @@ -44,13 +44,12 @@ import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; -/** A {@link FormatTableSplitEnumerator} whose partitions are discovered from the filesystem. */ -final class FileSystemFormatTableSplitEnumerator extends FormatTableSplitEnumerator { +/** A {@link SplitEnumerator} whose partitions are discovered from the filesystem. */ +final class FileSystemSplitEnumerator extends SplitEnumerator { - private static final Logger LOG = - LoggerFactory.getLogger(FileSystemFormatTableSplitEnumerator.class); + private static final Logger LOG = LoggerFactory.getLogger(FileSystemSplitEnumerator.class); - FileSystemFormatTableSplitEnumerator(FormatTable table, CoreOptions coreOptions) { + FileSystemSplitEnumerator(FormatTable table, CoreOptions coreOptions) { super(table, coreOptions); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index e9be09843d33..72fdff618291 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -63,7 +63,7 @@ public class FormatTableScan implements InnerTableScan { final FormatTable table; final CoreOptions coreOptions; @Nullable private PartitionPredicate partitionFilter; - private final FormatTableSplitEnumerator splitEnumerator; + private final SplitEnumerator splitEnumerator; @Nullable private final Integer limit; public FormatTableScan( @@ -74,8 +74,7 @@ public FormatTableScan( this.coreOptions = new CoreOptions(table.options()); this.partitionFilter = partitionFilter; this.limit = limit; - this.splitEnumerator = - FormatTableSplitEnumerator.create(table, coreOptions, table.partitionManager()); + this.splitEnumerator = SplitEnumerator.create(table, coreOptions, table.partitionManager()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java similarity index 90% rename from paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java rename to paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java index 79d1431b688f..4bfea41a4459 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSplitEnumerator.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java @@ -53,11 +53,11 @@ /** * Enumerates {@link FormatDataSplit}s for a {@link FormatTable}. * - *

Partitions come either from the catalog ({@link CatalogFormatTableSplitEnumerator}) or from - * the filesystem ({@link FileSystemFormatTableSplitEnumerator}); the source is chosen once by - * {@link #create} so the enumeration paths stay free of {@code partitionManager != null} branching. + *

Partitions come either from the catalog ({@link CatalogSplitEnumerator}) or from the + * filesystem ({@link FileSystemSplitEnumerator}); the source is chosen once by {@link #create} so + * the enumeration paths stay free of {@code partitionManager != null} branching. */ -abstract class FormatTableSplitEnumerator { +abstract class SplitEnumerator { protected final FormatTable table; protected final CoreOptions coreOptions; @@ -65,7 +65,7 @@ abstract class FormatTableSplitEnumerator { protected final long openFileCost; protected final FormatTable.Format format; - FormatTableSplitEnumerator(FormatTable table, CoreOptions coreOptions) { + SplitEnumerator(FormatTable table, CoreOptions coreOptions) { this.table = table; this.coreOptions = coreOptions; this.targetSplitSize = coreOptions.splitTargetSize(); @@ -73,14 +73,14 @@ abstract class FormatTableSplitEnumerator { this.format = table.format(); } - static FormatTableSplitEnumerator create( + static SplitEnumerator create( FormatTable table, CoreOptions coreOptions, @Nullable FormatTablePartitionManager partitionManager) { if (partitionManager != null) { - return new CatalogFormatTableSplitEnumerator(table, coreOptions, partitionManager); + return new CatalogSplitEnumerator(table, coreOptions, partitionManager); } - return new FileSystemFormatTableSplitEnumerator(table, coreOptions); + return new FileSystemSplitEnumerator(table, coreOptions); } final List enumerate(@Nullable PartitionPredicate partitionFilter) throws IOException {