This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9636197fdb [core] Support partition pruning for AndPartitionPredicate
in format table scan (#8353)
9636197fdb is described below
commit 9636197fdb8a2077f58dbff3abae137a82c785b4
Author: Zouxxyy <[email protected]>
AuthorDate: Thu Jun 25 20:45:26 2026 +0800
[core] Support partition pruning for AndPartitionPredicate in format table
scan (#8353)
When a format table scan receives an `AndPartitionPredicate` (e.g. from
combining a static
pushed-down partition filter with a runtime/DPP filter), the scan-path
prefix optimization
and per-directory pruning were skipped because only
`DefaultPartitionPredicate` was handled.
This caused a full listing of all partition directories from the table
root — expensive on
object stores like OSS/S3.
This PR adds `extractPartitionPredicate()` which recursively unwraps
`AndPartitionPredicate`
into a single `Predicate` for pruning. Children that cannot be expressed
as a `Predicate`
(e.g. `MultiplePartitionPredicate`) are skipped; correctness is still
guaranteed by the
in-memory `partitionFilter.test()` in `plan()`.
Also replaces `exists()` + `getFileStatus()` with a single
`getFileStatus()` catching
`FileNotFoundException` to save one RPC per directory on object stores.
---
.../paimon/partition/PartitionPredicate.java | 4 +
.../paimon/table/format/FormatTableScan.java | 61 +++++++++--
.../apache/paimon/utils/PartitionPathUtils.java | 42 ++++----
.../paimon/table/format/FormatTableScanTest.java | 112 ++++++++++++++++++---
.../paimon/spark/sql/FormatTableTestBase.scala | 40 ++++++++
5 files changed, 213 insertions(+), 46 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
index 2820b52e9b..3a594aa409 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
@@ -325,6 +325,10 @@ public interface PartitionPredicate extends Serializable {
this.predicates = Collections.unmodifiableList(new
ArrayList<>(predicates));
}
+ public List<PartitionPredicate> predicates() {
+ return predicates;
+ }
+
@Override
public boolean test(BinaryRow partition) {
for (PartitionPredicate predicate : predicates) {
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 0b63e71318..9dca62508f 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
@@ -33,6 +33,7 @@ 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.partition.PartitionPredicate.AndPartitionPredicate;
import
org.apache.paimon.partition.PartitionPredicate.DefaultPartitionPredicate;
import
org.apache.paimon.partition.PartitionPredicate.MultiplePartitionPredicate;
import org.apache.paimon.predicate.Equal;
@@ -54,6 +55,9 @@ import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PartitionPathUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.IOException;
@@ -76,6 +80,8 @@ 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 final FormatTable table;
private final CoreOptions coreOptions;
@Nullable private PartitionPredicate partitionFilter;
@@ -174,6 +180,10 @@ public class FormatTableScan implements InnerTableScan {
}
List<Pair<LinkedHashMap<String, String>, Path>> findPartitions() {
+ LOG.debug(
+ "Find partitions for format table {}, partition filter: {}",
+ table.name(),
+ partitionFilter);
boolean onlyValueInPath =
coreOptions.formatTablePartitionOnlyValueInPath();
if (partitionFilter instanceof MultiplePartitionPredicate) {
// generate partitions directly
@@ -190,17 +200,22 @@ public class FormatTableScan implements InnerTableScan {
// This will prune partition directories early during traversal,
// which is especially important for cloud storage like OSS/S3
Map<String, Predicate> partitionPredicates = new HashMap<>();
- if (partitionFilter instanceof DefaultPartitionPredicate) {
- Predicate predicate = ((DefaultPartitionPredicate)
partitionFilter).predicate();
+ Optional<Predicate> predicate =
extractPartitionPredicate(partitionFilter);
+ LOG.debug(
+ "Extracted predicate for format table {} partition
pruning: {}",
+ table.name(),
+ predicate.orElse(null));
+ if (predicate.isPresent()) {
partitionPredicates =
-
PredicateUtils.splitPartitionPredicate(table.partitionType(), predicate);
+ PredicateUtils.splitPartitionPredicate(
+ table.partitionType(), predicate.get());
}
Pair<Path, Integer> scanPathAndLevel =
computeScanPathAndLevel(
new Path(table.location()),
table.partitionKeys(),
- partitionFilter,
+ predicate,
table.partitionType(),
onlyValueInPath);
return searchPartSpecAndPaths(
@@ -241,22 +256,48 @@ public class FormatTableScan implements InnerTableScan {
return result;
}
+ /**
+ * Extracts the underlying {@link Predicate} used for partition-directory
pruning from a {@link
+ * PartitionPredicate}. Unlike data-table scans, which prune purely via
{@link
+ * PartitionPredicate#test} on partitions read from the manifest, a format
table has no manifest
+ * and must derive a {@link Predicate} to compute the scan-path prefix and
per-directory filters
+ * while listing. {@link AndPartitionPredicate} is unwrapped recursively.
Returns empty when the
+ * predicate cannot be expressed as a single {@link Predicate} (e.g. {@link
+ * MultiplePartitionPredicate}), in which case the caller falls back to
listing without pruning.
+ */
+ static Optional<Predicate> extractPartitionPredicate(
+ @Nullable PartitionPredicate partitionFilter) {
+ if (partitionFilter instanceof DefaultPartitionPredicate) {
+ return Optional.of(((DefaultPartitionPredicate)
partitionFilter).predicate());
+ } else if (partitionFilter instanceof AndPartitionPredicate) {
+ List<Predicate> predicates = new ArrayList<>();
+ for (PartitionPredicate child :
+ ((AndPartitionPredicate) partitionFilter).predicates()) {
+ Optional<Predicate> 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().
+ }
+ return predicates.isEmpty()
+ ? Optional.empty()
+ : Optional.of(PredicateBuilder.and(predicates));
+ }
+ return Optional.empty();
+ }
+
protected static Pair<Path, Integer> computeScanPathAndLevel(
Path tableLocation,
List<String> partitionKeys,
- PartitionPredicate partitionFilter,
+ Optional<Predicate> predicate,
RowType partitionType,
boolean onlyValueInPath) {
Path scanPath = tableLocation;
int level = partitionKeys.size();
if (!partitionKeys.isEmpty()) {
- // Try to optimize for equality partition filters
- if (partitionFilter instanceof DefaultPartitionPredicate) {
+ if (predicate.isPresent()) {
Map<String, String> equalityPrefix =
extractLeadingEqualityPartitionSpecWhenOnlyAnd(
- partitionKeys,
- ((DefaultPartitionPredicate)
partitionFilter).predicate(),
- partitionType);
+ partitionKeys, predicate.get(), partitionType);
if (!equalityPrefix.isEmpty()) {
// Use optimized scan for specific partition path
String partitionPath =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
index c20abf7f34..fc2fe23e05 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
@@ -28,6 +28,7 @@ import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.BitSet;
@@ -345,28 +346,25 @@ public class PartitionPathUtils {
ArrayList<FileStatus> result = new ArrayList<>();
try {
- if (fileIO.exists(path)) {
- // ignore hidden file
- FileStatus fileStatus = fileIO.getFileStatus(path);
- // Calculate the starting offset when we begin from a prefix
path
- // For example, if partitionKeys = [ds, hr] and expectLevel =
1 (only hr remaining),
- // then levelOffset = 2 - 1 = 1, so we access partitionKeys[1]
for level 0
- int levelOffset = partitionKeys.size() - expectLevel;
- listStatusRecursively(
- fileIO,
- fileStatus,
- 0,
- expectLevel,
- result,
- partitionKeys,
- onlyValueInPath,
- partitionFilter,
- partitionType,
- defaultPartValue,
- levelOffset);
- } else {
- return new FileStatus[0];
- }
+ FileStatus fileStatus = fileIO.getFileStatus(path);
+ // Calculate the starting offset when we begin from a prefix path
+ // For example, if partitionKeys = [ds, hr] and expectLevel = 1
(only hr remaining),
+ // then levelOffset = 2 - 1 = 1, so we access partitionKeys[1] for
level 0
+ int levelOffset = partitionKeys.size() - expectLevel;
+ listStatusRecursively(
+ fileIO,
+ fileStatus,
+ 0,
+ expectLevel,
+ result,
+ partitionKeys,
+ onlyValueInPath,
+ partitionFilter,
+ partitionType,
+ defaultPartValue,
+ levelOffset);
+ } catch (FileNotFoundException e) {
+ return new FileStatus[0];
} catch (IOException e) {
throw new RuntimeException("Failed to list files in " + path, e);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
index 936b3d6bd1..0584577a1a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
@@ -50,6 +50,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
@@ -123,7 +124,11 @@ public class FormatTableScanTest {
Pair<Path, Integer> result =
FormatTableScan.computeScanPathAndLevel(
- defaultTableLocation, partitionKeys, partitionFilter,
partitionType, false);
+ defaultTableLocation,
+ partitionKeys,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
+ partitionType,
+ false);
assertThat(result.getLeft()).isEqualTo(defaultTableLocation);
assertThat(result.getRight()).isEqualTo(0);
@@ -150,7 +155,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
defaultTableLocation,
partitionType.getFieldNames(),
- null,
+ Optional.empty(),
partitionType,
false);
@@ -169,7 +174,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -210,7 +215,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
String partitionPath = enablePartitionValueOnly ? "2023/12" :
"year=2023/month=12";
@@ -253,7 +258,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
datePartitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
datePartitionType,
enablePartitionValueOnly);
String partitionPath = enablePartitionValueOnly ? "2026-05-01" :
"dt=2026-05-01";
@@ -275,7 +280,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -317,7 +322,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -352,7 +357,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -392,7 +397,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -415,7 +420,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -448,7 +453,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -482,7 +487,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -516,7 +521,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -549,7 +554,7 @@ public class FormatTableScanTest {
FormatTableScan.computeScanPathAndLevel(
tableLocation,
partitionKeys,
- partitionFilter,
+
FormatTableScan.extractPartitionPredicate(partitionFilter),
partitionType,
enablePartitionValueOnly);
@@ -934,4 +939,83 @@ public class FormatTableScanTest {
.collect(java.util.stream.Collectors.toList());
assertEquals(Arrays.asList("4", "5", "6", "7", "8", "9"), month);
}
+
+ @TestTemplate
+ void testFindPartitionsWithAndPartitionPredicate() throws IOException {
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO setupFileIO = LocalFileIO.create();
+
+ // Create partition directories for years 2022 to 2026, 12 months each
(60 partitions)
+ for (int year = 2022; year <= 2026; year++) {
+ String partPath = enablePartitionValueOnly ? String.valueOf(year)
: "year=" + year;
+ for (int month = 1; month <= 12; month++) {
+ String monthPart =
+ enablePartitionValueOnly
+ ? partPath + "/" + month
+ : partPath + "/month=" + month;
+ setupFileIO.mkdirs(new Path(tableLocation, monthPart));
+ }
+ }
+
+ AtomicInteger listCount = new AtomicInteger(0);
+ LocalFileIO localFileIO =
+ new LocalFileIO() {
+ @Override
+ public FileStatus[] listStatus(Path path) throws
IOException {
+ listCount.getAndIncrement();
+ return super.listStatus(path);
+ }
+ };
+
+ RowType rowType =
+ RowType.builder()
+ .field("year", DataTypes.INT())
+ .field("month", DataTypes.INT())
+ .field("a", DataTypes.INT())
+ .build();
+
+ FormatTable formatTable =
+ FormatTable.builder()
+ .fileIO(localFileIO)
+ .identifier(Identifier.create("test_db", "test_table"))
+ .rowType(rowType)
+ .partitionKeys(Arrays.asList("year", "month"))
+ .location(tableLocation.toString())
+ .format(FormatTable.Format.CSV)
+ .options(
+ Collections.singletonMap(
+
FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+
String.valueOf(enablePartitionValueOnly)))
+ .build();
+
+ // Simulate a static pushdown (year = 2024) combined with a
runtime/DPP partition filter
+ // (month > 3 AND month < 10): each is a DefaultPartitionPredicate and
the scan combines
+ // them via PartitionPredicate.and, exactly like BaseScan / runtime
filtering does. The
+ // combined predicate must still prune partition directories;
otherwise it falls back to a
+ // full table-root listing.
+ PredicateBuilder builder = new
PredicateBuilder(formatTable.partitionType());
+ PartitionPredicate yearFilter =
+ PartitionPredicate.fromPredicate(
+ formatTable.partitionType(), builder.equal(0, 2024));
+ PartitionPredicate monthFilter =
+ PartitionPredicate.fromPredicate(
+ formatTable.partitionType(),
+ PredicateBuilder.and(builder.greaterThan(1, 3),
builder.lessThan(1, 10)));
+ PartitionPredicate combined =
+ PartitionPredicate.and(Arrays.asList(yearFilter, monthFilter));
+
+ FormatTableScan scan = new FormatTableScan(formatTable, combined,
null);
+ List<Pair<LinkedHashMap<String, String>, Path>> result =
scan.findPartitions();
+
+ // Should prune to year=2024 and list its months only once (months
4-9).
+ assertEquals(6, result.size());
+ assertEquals(1, listCount.get());
+ List<String> months =
+ result.stream()
+ .map(pair -> pair.getKey().get("month"))
+ .sorted()
+ .distinct()
+ .collect(java.util.stream.Collectors.toList());
+ assertEquals(Arrays.asList("4", "5", "6", "7", "8", "9"), months);
+ }
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
index f32447cf43..4d1d49a799 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
@@ -440,6 +440,46 @@ abstract class FormatTableTestBase extends
PaimonHiveTestBase with AdaptiveSpark
}
}
+ test("Paimon format table: runtime filter combined with pushed-down
partition filter") {
+ withTable("dwd_fact", "dim_date") {
+ sql("""
+ |CREATE TABLE dwd_fact (id INT, amount DOUBLE, dt STRING, hour
STRING)
+ |USING PARQUET
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |PARTITIONED BY (dt, hour)
+ |""".stripMargin)
+ sql("""
+ |CREATE TABLE dim_date (dt STRING, name STRING)
+ |USING PARQUET
+ |TBLPROPERTIES ('format-table.implementation'='paimon')
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO dwd_fact VALUES
+ |(1, 10.0, '20260622', '00'),
+ |(2, 20.0, '20260622', '01'),
+ |(3, 30.0, '20260621', '00'),
+ |(4, 40.0, '20260620', '23')
+ |""".stripMargin)
+ sql("INSERT INTO dim_date VALUES ('20260622', 'today')")
+
+ val df = sql("""
+ |SELECT f.id, f.dt, f.hour, d.name
+ |FROM dwd_fact f
+ |JOIN dim_date d ON f.dt = d.dt
+ |WHERE f.dt >= '20260620'
+ |ORDER BY f.id
+ |""".stripMargin)
+
+ checkAnswer(df, Seq(Row(1, "20260622", "00", "today"), Row(2,
"20260622", "01", "today")))
+
+ // Static pushdown (dt >= '20260620') alone keeps all 4 partitions; the
runtime filter on dt
+ // prunes the scan down to the two partitions of dt='20260622'.
+ val filteredSplits =
collectFilteredInputSplits(df.queryExecution.executedPlan, "dwd_fact")
+ assert(filteredSplits.size == 2)
+ }
+ }
+
def collectFilteredInputSplits(plan: SparkPlan, tableName: String):
Seq[Split] = {
flatMap(plan) {
case s: BatchScanExec =>
