Skip to content

[core] Support partition predicate pushdown for PartitionsTable #7628

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sundapeng wants to merge 4 commits into
base: master
Choose a base branch
from
Open

[core] Support partition predicate pushdown for PartitionsTable #7628

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1116648
[core] Support partition predicate pushdown for PartitionsTable
sundapeng Apr 10, 2026
6ab7f9c
[core] Fix null partition value mismatch in PartitionsTable filterByP…
sundapeng Apr 11, 2026
e5185ab
[core] Fix RestPartitionsTableTest by creating table through catalog API
sundapeng Apr 11, 2026
62d378e
[core] Fix default partition handling and improve predicate pushdown …
sundapeng Apr 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,8 @@ public RecordReader<InternalRow> createReader(Split split) {
snapshotReader,
bucketsSplit.partitionPredicate,
partitionKeys,
partitionType);
partitionType,
fileStoreTable.coreOptions().partitionDefaultName());
if (!hasResults) {
return new IteratorRecordReader<>(Collections.emptyIterator());
}
Expand Down
6 changes: 5 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/table/system/FileKeyRangesTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,11 @@ public Plan innerPlan() {
RowType partitionType = fileStoreTable.schema().logicalPartitionType();
boolean hasResults =
PartitionPredicateHelper.applyPartitionFilter(
snapshotReader, partitionPredicate, partitionKeys, partitionType);
snapshotReader,
partitionPredicate,
partitionKeys,
partitionType,
fileStoreTable.coreOptions().partitionDefaultName());
if (!hasResults) {
return Collections::emptyList;
}
Expand Down
6 changes: 5 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,11 @@ public Plan innerPlan() {
RowType partitionType = fileStoreTable.schema().logicalPartitionType();
boolean hasResults =
PartitionPredicateHelper.applyPartitionFilter(
snapshotReader, partitionPredicate, partitionKeys, partitionType);
snapshotReader,
partitionPredicate,
partitionKeys,
partitionType,
fileStoreTable.coreOptions().partitionDefaultName());
if (!hasResults) {
return Collections::emptyList;
}
Expand Down
142 changes: 129 additions & 13 deletions paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,12 @@
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.In;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LeafPredicateExtractor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.ReadonlyTable;
Expand All @@ -56,15 +61,19 @@
import org.apache.paimon.utils.InternalRowUtils;
import org.apache.paimon.utils.IteratorRecordReader;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.PartitionPredicateHelper;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.SerializationUtils;

import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;

import javax.annotation.Nullable;

import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
Expand Down Expand Up @@ -140,32 +149,95 @@ public Table copy(Map<String, String> dynamicOptions) {

private static class PartitionsScan extends ReadOnceTableScan {

@Nullable private LeafPredicate partitionPredicate;

@Override
public InnerTableScan withFilter(Predicate predicate) {
if (predicate == null) {
return this;
}

Map<String, LeafPredicate> leafPredicates =
predicate.visit(LeafPredicateExtractor.INSTANCE);
this.partitionPredicate = leafPredicates.get("partition");

// Handle Or(Equal, Equal...) pattern from PredicateBuilder.in() with <=20 literals
if (this.partitionPredicate == null) {
for (Predicate andChild : PredicateBuilder.splitAnd(predicate)) {
LeafPredicate inPred = convertOrEqualsToIn(andChild, "partition");
if (inPred != null) {
this.partitionPredicate = inPred;
break;
}
}
}
return this;
}

@Nullable
private static LeafPredicate convertOrEqualsToIn(Predicate predicate, String targetField) {
List<Predicate> orChildren = PredicateBuilder.splitOr(predicate);
if (orChildren.size() <= 1) {
return null;
}
List<Object> literals = new ArrayList<>();
String fieldName = null;
int fieldIndex = -1;
DataType fieldType = null;
for (Predicate child : orChildren) {
if (!(child instanceof LeafPredicate)) {
return null;
}
LeafPredicate leaf = (LeafPredicate) child;
if (!(leaf.function() instanceof Equal)) {
return null;
}
if (fieldName == null) {
fieldName = leaf.fieldName();
fieldIndex = leaf.index();
fieldType = leaf.type();
} else if (!fieldName.equals(leaf.fieldName())) {
return null;
}
literals.addAll(leaf.literals());
}
if (!targetField.equals(fieldName)) {
return null;
}
return new LeafPredicate(In.INSTANCE, fieldType, fieldIndex, fieldName, literals);
}

@Override
public Plan innerPlan() {
return () -> Collections.singletonList(new PartitionsSplit());
return () -> Collections.singletonList(new PartitionsSplit(partitionPredicate));
}
}

private static class PartitionsSplit extends SingletonSplit {

private static final long serialVersionUID = 1L;

@Nullable private final LeafPredicate partitionPredicate;

private PartitionsSplit(@Nullable LeafPredicate partitionPredicate) {
this.partitionPredicate = partitionPredicate;
}

@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
return o != null && getClass() == o.getClass();
if (o == null || getClass() != o.getClass()) {
return false;
}
PartitionsSplit that = (PartitionsSplit) o;
return Objects.equals(partitionPredicate, that.partitionPredicate);
}

@Override
public int hashCode() {
return 1;
return Objects.hash(partitionPredicate);
}

@Override
Expand All @@ -186,7 +258,7 @@ public PartitionsRead(FileStoreTable table) {

@Override
public InnerTableRead withFilter(Predicate predicate) {
// TODO
// filter pushdown is handled at the Scan layer through PartitionsSplit
return this;
}

Expand All @@ -207,7 +279,8 @@ public RecordReader<InternalRow> createReader(Split split) throws IOException {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}

List<Partition> partitions = listPartitions();
PartitionsSplit partitionsSplit = (PartitionsSplit) split;
List<Partition> partitions = listPartitions(partitionsSplit.partitionPredicate);

List<DataType> fieldTypes =
fileStoreTable.schema().logicalPartitionType().getFieldTypes();
Expand Down Expand Up @@ -320,17 +393,17 @@ private Timestamp toTimestamp(Long epochMillis) {
Instant.ofEpochMilli(epochMillis), ZoneId.systemDefault()));
}

private List<Partition> listPartitions() {
private List<Partition> listPartitions(@Nullable LeafPredicate partitionPredicate) {
CatalogLoader catalogLoader = fileStoreTable.catalogEnvironment().catalogLoader();
if (TimeTravelUtil.hasTimeTravelOptions(new Options(fileStoreTable.options()))
|| catalogLoader == null) {
return listPartitionEntries();
return listPartitionEntries(partitionPredicate);
}

try (Catalog catalog = catalogLoader.load()) {
Identifier baseIdentifier = fileStoreTable.catalogEnvironment().identifier();
if (baseIdentifier == null) {
return listPartitionEntries();
return listPartitionEntries(partitionPredicate);
}
String branch = fileStoreTable.coreOptions().branch();
Identifier identifier;
Expand All @@ -343,15 +416,58 @@ private List<Partition> listPartitions() {
} else {
identifier = baseIdentifier;
}
return catalog.listPartitions(identifier);
List<Partition> partitions = catalog.listPartitions(identifier);
return filterByPredicate(partitions, partitionPredicate);
} catch (Exception e) {
return listPartitionEntries();
return listPartitionEntries(partitionPredicate);
}
}

private List<Partition> filterByPredicate(
List<Partition> partitions, @Nullable LeafPredicate partitionPredicate) {
if (partitionPredicate == null) {
return partitions;
}
List<String> partitionKeys = fileStoreTable.partitionKeys();
String defaultPartitionName = fileStoreTable.coreOptions().partitionDefaultName();
return partitions.stream()
.filter(
p -> {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < partitionKeys.size(); i++) {
if (i > 0) {
sb.append("/");
}
String value = p.spec().get(partitionKeys.get(i));
sb.append(partitionKeys.get(i))
.append("=")
.append(value == null ? defaultPartitionName : value);
}
return partitionPredicate.test(
GenericRow.of(BinaryString.fromString(sb.toString())));
})
.collect(Collectors.toList());
}

private List<Partition> listPartitionEntries() {
List<PartitionEntry> partitionEntries =
fileStoreTable.newScan().withLevelFilter(level -> true).listPartitionEntries();
private List<Partition> listPartitionEntries(@Nullable LeafPredicate partitionPredicate) {
InnerTableScan scan = fileStoreTable.newScan().withLevelFilter(level -> true);

if (partitionPredicate != null) {
List<String> partitionKeys = fileStoreTable.partitionKeys();
RowType partitionType = fileStoreTable.schema().logicalPartitionType();
String defaultPartitionName = fileStoreTable.coreOptions().partitionDefaultName();
Predicate partPred =
PartitionPredicateHelper.buildPartitionPredicate(
partitionPredicate,
partitionKeys,
partitionType,
defaultPartitionName);
if (partPred != null) {
scan.withPartitionFilter(partPred);
}
}

List<PartitionEntry> partitionEntries = scan.listPartitionEntries();
RowType partitionType = fileStoreTable.schema().logicalPartitionType();
String defaultPartitionName = fileStoreTable.coreOptions().partitionDefaultName();
String[] partitionColumns = fileStoreTable.partitionKeys().toArray(new String[0]);
Expand Down
Loading