predicates = PredicateBuilder.splitAnd(predicate);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
new file mode 100644
index 000000000000..4bfea41a4459
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
@@ -0,0 +1,168 @@
+/*
+ * 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.manifest.PartitionEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+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;
+
+/**
+ * Enumerates {@link FormatDataSplit}s for a {@link FormatTable}.
+ *
+ * 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 SplitEnumerator {
+
+ protected final FormatTable table;
+ protected final CoreOptions coreOptions;
+ protected final long targetSplitSize;
+ protected final long openFileCost;
+ protected final FormatTable.Format format;
+
+ SplitEnumerator(FormatTable table, CoreOptions coreOptions) {
+ this.table = table;
+ this.coreOptions = coreOptions;
+ this.targetSplitSize = coreOptions.splitTargetSize();
+ this.openFileCost = coreOptions.splitOpenFileCost();
+ this.format = table.format();
+ }
+
+ static SplitEnumerator create(
+ FormatTable table,
+ CoreOptions coreOptions,
+ @Nullable FormatTablePartitionManager partitionManager) {
+ if (partitionManager != null) {
+ return new CatalogSplitEnumerator(table, coreOptions, partitionManager);
+ }
+ return new FileSystemSplitEnumerator(table, coreOptions);
+ }
+
+ final List enumerate(@Nullable PartitionPredicate partitionFilter) throws IOException {
+ if (table.partitionKeys().isEmpty()) {
+ return createSplits(table.fileIO(), new Path(table.location()), null);
+ }
+ 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 =
+ convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName());
+ return new InternalRowSerializer(partitionType).toBinaryRow(row);
+ }
+
+ 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;
+ }
+ }
+}
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..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
@@ -53,9 +53,14 @@
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.concurrent.atomic.AtomicReference;
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 +511,281 @@ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
return super.listFiles(path, recursive);
}
}
+
+ @Test
+ void testCatalogPartitionListingRunsInParallelAndPreservesOrder() throws Exception {
+ ParallelTrackingLocalFileIO fileIO = new ParallelTrackingLocalFileIO();
+ 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 parallel =
+ plannedFiles(
+ new FormatTableScan(
+ stringPartitionTable(
+ fileIO, tablePath, recordingCatalog(partitions), 8),
+ null,
+ null)
+ .plan()
+ .splits());
+
+ assertThat(parallel).containsExactlyElementsOf(expected);
+ assertThat(fileIO.maxConcurrentListings()).isGreaterThan(1);
+ }
+
+ @Test
+ void testParallelListingSkipsMissingCatalogRegisteredPartition() 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");
+ fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing"));
+
+ List files =
+ plannedFiles(
+ new FormatTableScan(
+ stringPartitionTable(
+ fileIO, tablePath, recordingCatalog(partitions), 4),
+ null,
+ null)
+ .plan()
+ .splits());
+
+ assertThat(files).containsExactlyInAnyOrder(oct, dec);
+ }
+
+ @Test
+ 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");
+ fileIO.failListFilesContaining("month=11", new IOException("boom"));
+
+ FormatTable table =
+ stringPartitionTable(fileIO, tablePath, recordingCatalog(partitions), 4);
+ assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IOException.class)
+ .hasRootCauseMessage("boom");
+ }
+
+ @Test
+ 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");
+ fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing"));
+
+ FormatTable table = createStringPartitionTable(fileIO, tablePath, null);
+ assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
+ .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();
+ }
+
+ @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,
+ @Nullable FormatTablePartitionManager partitionManager,
+ 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(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<>();
+
+ 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);
+ }
+ }
+
+ /** 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);
+ }
+ }
}