You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/03 08:02:18 UTC
(paimon) branch master updated: [core] Support partition and bucket predicate pushdown for BucketsTable (#7592)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 051dc75945 [core] Support partition and bucket predicate pushdown for BucketsTable (#7592)
051dc75945 is described below
commit 051dc7594564d6871f640e592bf9b961acd0e9d8
Author: Kerwin Zhang <xi...@alibaba-inc.com>
AuthorDate: Fri Apr 3 16:02:12 2026 +0800
[core] Support partition and bucket predicate pushdown for BucketsTable (#7592)
---
.../apache/paimon/table/system/BucketsTable.java | 67 ++++++++++-
.../paimon/table/system/FileKeyRangesTable.java | 90 ++-------------
.../org/apache/paimon/table/system/FilesTable.java | 90 ++-------------
.../paimon/utils/PartitionPredicateHelper.java | 126 +++++++++++++++++++++
.../paimon/table/system/BucketsTableTest.java | 88 ++++++++++++++
5 files changed, 292 insertions(+), 169 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
index 18aaee6f86..e5c419f515 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
@@ -27,6 +27,8 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LeafPredicateExtractor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
@@ -38,6 +40,7 @@ import org.apache.paimon.table.source.ReadOnceTableScan;
import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
@@ -48,6 +51,8 @@ import org.apache.paimon.utils.SerializationUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+import javax.annotation.Nullable;
+
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -57,9 +62,11 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.OptionalLong;
import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
+import static org.apache.paimon.utils.PartitionPredicateHelper.applyPartitionFilter;
/** A {@link Table} for showing buckets info. */
public class BucketsTable implements ReadonlyTable {
@@ -121,14 +128,27 @@ public class BucketsTable implements ReadonlyTable {
private static class BucketsScan extends ReadOnceTableScan {
+ @Nullable private LeafPredicate partitionPredicate;
+ @Nullable private LeafPredicate bucketPredicate;
+
@Override
public InnerTableScan withFilter(Predicate predicate) {
+ if (predicate == null) {
+ return this;
+ }
+
+ Map<String, LeafPredicate> leafPredicates =
+ predicate.visit(LeafPredicateExtractor.INSTANCE);
+ this.partitionPredicate = leafPredicates.get("partition");
+ this.bucketPredicate = leafPredicates.get("bucket");
return this;
}
@Override
public Plan innerPlan() {
- return () -> Collections.singletonList(new BucketsSplit());
+ return () ->
+ Collections.singletonList(
+ new BucketsSplit(partitionPredicate, bucketPredicate));
}
}
@@ -136,17 +156,32 @@ public class BucketsTable implements ReadonlyTable {
private static final long serialVersionUID = 1L;
+ @Nullable private final LeafPredicate partitionPredicate;
+ @Nullable private final LeafPredicate bucketPredicate;
+
+ private BucketsSplit(
+ @Nullable LeafPredicate partitionPredicate,
+ @Nullable LeafPredicate bucketPredicate) {
+ this.partitionPredicate = partitionPredicate;
+ this.bucketPredicate = bucketPredicate;
+ }
+
@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;
+ }
+ BucketsSplit that = (BucketsSplit) o;
+ return Objects.equals(partitionPredicate, that.partitionPredicate)
+ && Objects.equals(bucketPredicate, that.bucketPredicate);
}
@Override
public int hashCode() {
- return 1;
+ return Objects.hash(partitionPredicate, bucketPredicate);
}
@Override
@@ -167,7 +202,7 @@ public class BucketsTable implements ReadonlyTable {
@Override
public InnerTableRead withFilter(Predicate predicate) {
- // TODO
+ // filter pushdown is handled at the Scan layer through BucketsSplit
return this;
}
@@ -188,7 +223,29 @@ public class BucketsTable implements ReadonlyTable {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
- List<BucketEntry> buckets = fileStoreTable.newSnapshotReader().bucketEntries();
+ BucketsSplit bucketsSplit = (BucketsSplit) split;
+ SnapshotReader snapshotReader = fileStoreTable.newSnapshotReader();
+
+ // Apply partition filter to SnapshotReader
+ List<String> partitionKeys = fileStoreTable.partitionKeys();
+ RowType partitionType = fileStoreTable.schema().logicalPartitionType();
+ boolean hasResults =
+ applyPartitionFilter(
+ snapshotReader,
+ bucketsSplit.partitionPredicate,
+ partitionKeys,
+ partitionType);
+ if (!hasResults) {
+ return new IteratorRecordReader<>(Collections.emptyIterator());
+ }
+
+ // Apply bucket filter to SnapshotReader
+ if (bucketsSplit.bucketPredicate != null) {
+ LeafPredicate bp = bucketsSplit.bucketPredicate;
+ snapshotReader.withBucketFilter(bucket -> bp.test(GenericRow.of(null, bucket)));
+ }
+
+ List<BucketEntry> buckets = snapshotReader.bucketEntries();
@SuppressWarnings("unchecked")
CastExecutor<InternalRow, BinaryString> partitionCastExecutor =
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FileKeyRangesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FileKeyRangesTable.java
index feeedbc7b5..095dcda74a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FileKeyRangesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FileKeyRangesTable.java
@@ -27,13 +27,9 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
-import org.apache.paimon.predicate.Equal;
-import org.apache.paimon.predicate.In;
-import org.apache.paimon.predicate.LeafBinaryFunction;
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.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -52,10 +48,10 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.PartitionPredicateHelper;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import org.apache.paimon.utils.SerializationUtils;
-import org.apache.paimon.utils.TypeUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
@@ -66,7 +62,6 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
@@ -171,62 +166,13 @@ public class FileKeyRangesTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
SnapshotReader snapshotReader = fileStoreTable.newSnapshotReader();
- if (partitionPredicate != null) {
- List<String> partitionKeys = fileStoreTable.partitionKeys();
- RowType partitionType = fileStoreTable.schema().logicalPartitionType();
- if (partitionPredicate.function() instanceof Equal) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(
- partitionPredicate.literals().get(0).toString(), partitionKeys);
- if (partSpec == null) {
- return Collections::emptyList;
- }
- snapshotReader.withPartitionFilter(partSpec);
- } else if (partitionPredicate.function() instanceof In) {
- List<Predicate> orPredicates = new ArrayList<>();
- PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
- for (Object literal : partitionPredicate.literals()) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(literal.toString(), partitionKeys);
- if (partSpec == null) {
- continue;
- }
- List<Predicate> andPredicates = new ArrayList<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- Object value =
- TypeUtils.castFromString(
- partSpec.get(partitionKeys.get(i)),
- partitionType.getTypeAt(i));
- andPredicates.add(partBuilder.equal(i, value));
- }
- orPredicates.add(PredicateBuilder.and(andPredicates));
- }
- if (!orPredicates.isEmpty()) {
- snapshotReader.withPartitionFilter(PredicateBuilder.or(orPredicates));
- }
- } else if (partitionPredicate.function() instanceof LeafBinaryFunction) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(
- partitionPredicate.literals().get(0).toString(), partitionKeys);
- if (partSpec != null) {
- PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
- List<Predicate> predicates = new ArrayList<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- Object value =
- TypeUtils.castFromString(
- partSpec.get(partitionKeys.get(i)),
- partitionType.getTypeAt(i));
- predicates.add(
- new LeafPredicate(
- partitionPredicate.function(),
- partitionType.getTypeAt(i),
- i,
- partitionKeys.get(i),
- Collections.singletonList(value)));
- }
- snapshotReader.withPartitionFilter(PredicateBuilder.and(predicates));
- }
- }
+ List<String> partitionKeys = fileStoreTable.partitionKeys();
+ RowType partitionType = fileStoreTable.schema().logicalPartitionType();
+ boolean hasResults =
+ PartitionPredicateHelper.applyPartitionFilter(
+ snapshotReader, partitionPredicate, partitionKeys, partitionType);
+ if (!hasResults) {
+ return Collections::emptyList;
}
return () ->
@@ -234,26 +180,6 @@ public class FileKeyRangesTable implements ReadonlyTable {
.map(p -> new FilesTable.FilesSplit(p, bucketPredicate, levelPredicate))
.collect(Collectors.toList());
}
-
- @Nullable
- private LinkedHashMap<String, String> parsePartitionSpec(
- String partitionStr, List<String> partitionKeys) {
- if (partitionStr.startsWith("{")) {
- partitionStr = partitionStr.substring(1);
- }
- if (partitionStr.endsWith("}")) {
- partitionStr = partitionStr.substring(0, partitionStr.length() - 1);
- }
- String[] partFields = partitionStr.split(", ");
- if (partitionKeys.size() != partFields.length) {
- return null;
- }
- LinkedHashMap<String, String> partSpec = new LinkedHashMap<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- partSpec.put(partitionKeys.get(i), partFields[i]);
- }
- return partSpec;
- }
}
private static class FileKeyRangesRead implements InnerTableRead {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
index e0a89d92bf..0c7eadb7ae 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FilesTable.java
@@ -32,13 +32,9 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.FileSource;
-import org.apache.paimon.predicate.Equal;
-import org.apache.paimon.predicate.In;
-import org.apache.paimon.predicate.LeafBinaryFunction;
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.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -64,10 +60,10 @@ import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InternalRowUtils;
import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.PartitionPredicateHelper;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import org.apache.paimon.utils.SerializationUtils;
-import org.apache.paimon.utils.TypeUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
@@ -78,7 +74,6 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -197,62 +192,13 @@ public class FilesTable implements ReadonlyTable {
@Override
public Plan innerPlan() {
SnapshotReader snapshotReader = fileStoreTable.newSnapshotReader();
- if (partitionPredicate != null) {
- List<String> partitionKeys = fileStoreTable.partitionKeys();
- RowType partitionType = fileStoreTable.schema().logicalPartitionType();
- if (partitionPredicate.function() instanceof Equal) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(
- partitionPredicate.literals().get(0).toString(), partitionKeys);
- if (partSpec == null) {
- return Collections::emptyList;
- }
- snapshotReader.withPartitionFilter(partSpec);
- } else if (partitionPredicate.function() instanceof In) {
- List<Predicate> orPredicates = new ArrayList<>();
- PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
- for (Object literal : partitionPredicate.literals()) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(literal.toString(), partitionKeys);
- if (partSpec == null) {
- continue;
- }
- List<Predicate> andPredicates = new ArrayList<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- Object value =
- TypeUtils.castFromString(
- partSpec.get(partitionKeys.get(i)),
- partitionType.getTypeAt(i));
- andPredicates.add(partBuilder.equal(i, value));
- }
- orPredicates.add(PredicateBuilder.and(andPredicates));
- }
- if (!orPredicates.isEmpty()) {
- snapshotReader.withPartitionFilter(PredicateBuilder.or(orPredicates));
- }
- } else if (partitionPredicate.function() instanceof LeafBinaryFunction) {
- LinkedHashMap<String, String> partSpec =
- parsePartitionSpec(
- partitionPredicate.literals().get(0).toString(), partitionKeys);
- if (partSpec != null) {
- PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
- List<Predicate> predicates = new ArrayList<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- Object value =
- TypeUtils.castFromString(
- partSpec.get(partitionKeys.get(i)),
- partitionType.getTypeAt(i));
- predicates.add(
- new LeafPredicate(
- partitionPredicate.function(),
- partitionType.getTypeAt(i),
- i,
- partitionKeys.get(i),
- Collections.singletonList(value)));
- }
- snapshotReader.withPartitionFilter(PredicateBuilder.and(predicates));
- }
- }
+ List<String> partitionKeys = fileStoreTable.partitionKeys();
+ RowType partitionType = fileStoreTable.schema().logicalPartitionType();
+ boolean hasResults =
+ PartitionPredicateHelper.applyPartitionFilter(
+ snapshotReader, partitionPredicate, partitionKeys, partitionType);
+ if (!hasResults) {
+ return Collections::emptyList;
}
return () ->
@@ -260,26 +206,6 @@ public class FilesTable implements ReadonlyTable {
.map(p -> new FilesSplit(p, bucketPredicate, levelPredicate))
.collect(Collectors.toList());
}
-
- @Nullable
- private LinkedHashMap<String, String> parsePartitionSpec(
- String partitionStr, List<String> partitionKeys) {
- if (partitionStr.startsWith("{")) {
- partitionStr = partitionStr.substring(1);
- }
- if (partitionStr.endsWith("}")) {
- partitionStr = partitionStr.substring(0, partitionStr.length() - 1);
- }
- String[] partFields = partitionStr.split(", ");
- if (partitionKeys.size() != partFields.length) {
- return null;
- }
- LinkedHashMap<String, String> partSpec = new LinkedHashMap<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- partSpec.put(partitionKeys.get(i), partFields[i]);
- }
- return partSpec;
- }
}
static class FilesSplit extends SingletonSplit {
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPredicateHelper.java b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPredicateHelper.java
new file mode 100644
index 0000000000..634d9e5d5b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPredicateHelper.java
@@ -0,0 +1,126 @@
+/*
+ * 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.utils;
+
+import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.In;
+import org.apache.paimon.predicate.LeafBinaryFunction;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+/**
+ * Helper for applying partition predicate pushdown in system tables (BucketsTable, FilesTable,
+ * FileKeyRangesTable).
+ */
+public class PartitionPredicateHelper {
+
+ public static boolean applyPartitionFilter(
+ SnapshotReader snapshotReader,
+ @Nullable LeafPredicate partitionPredicate,
+ List<String> partitionKeys,
+ RowType partitionType) {
+ if (partitionPredicate == null) {
+ return true;
+ }
+
+ if (partitionPredicate.function() instanceof Equal) {
+ LinkedHashMap<String, String> partSpec =
+ parsePartitionSpec(
+ partitionPredicate.literals().get(0).toString(), partitionKeys);
+ if (partSpec == null) {
+ return false;
+ }
+ snapshotReader.withPartitionFilter(partSpec);
+ } else if (partitionPredicate.function() instanceof In) {
+ List<Predicate> orPredicates = new ArrayList<>();
+ PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
+ for (Object literal : partitionPredicate.literals()) {
+ LinkedHashMap<String, String> partSpec =
+ parsePartitionSpec(literal.toString(), partitionKeys);
+ if (partSpec == null) {
+ continue;
+ }
+ List<Predicate> andPredicates = new ArrayList<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ Object value =
+ TypeUtils.castFromString(
+ partSpec.get(partitionKeys.get(i)), partitionType.getTypeAt(i));
+ andPredicates.add(partBuilder.equal(i, value));
+ }
+ orPredicates.add(PredicateBuilder.and(andPredicates));
+ }
+ if (!orPredicates.isEmpty()) {
+ snapshotReader.withPartitionFilter(PredicateBuilder.or(orPredicates));
+ }
+ } else if (partitionPredicate.function() instanceof LeafBinaryFunction) {
+ LinkedHashMap<String, String> partSpec =
+ parsePartitionSpec(
+ partitionPredicate.literals().get(0).toString(), partitionKeys);
+ if (partSpec != null) {
+ PredicateBuilder partBuilder = new PredicateBuilder(partitionType);
+ List<Predicate> predicates = new ArrayList<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ Object value =
+ TypeUtils.castFromString(
+ partSpec.get(partitionKeys.get(i)), partitionType.getTypeAt(i));
+ predicates.add(
+ new LeafPredicate(
+ partitionPredicate.function(),
+ partitionType.getTypeAt(i),
+ i,
+ partitionKeys.get(i),
+ Collections.singletonList(value)));
+ }
+ snapshotReader.withPartitionFilter(PredicateBuilder.and(predicates));
+ }
+ }
+
+ return true;
+ }
+
+ @Nullable
+ public static LinkedHashMap<String, String> parsePartitionSpec(
+ String partitionStr, List<String> partitionKeys) {
+ if (partitionStr.startsWith("{")) {
+ partitionStr = partitionStr.substring(1);
+ }
+ if (partitionStr.endsWith("}")) {
+ partitionStr = partitionStr.substring(0, partitionStr.length() - 1);
+ }
+ String[] partFields = partitionStr.split(", ");
+ if (partitionKeys.size() != partFields.length) {
+ return null;
+ }
+ LinkedHashMap<String, String> partSpec = new LinkedHashMap<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ partSpec.put(partitionKeys.get(i), partFields[i]);
+ }
+ return partSpec;
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
index 3e7c72871d..03431cb3dd 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
@@ -20,9 +20,14 @@ package org.apache.paimon.table.system;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
@@ -31,11 +36,17 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.Projection;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
import static org.apache.paimon.table.system.BucketsTable.BUCKETS;
import static org.assertj.core.api.Assertions.assertThat;
@@ -77,4 +88,81 @@ public class BucketsTableTest extends TableTestBase {
GenericRow.of(BinaryString.fromString("{1}"), 0, 2L, 2L),
GenericRow.of(BinaryString.fromString("{2}"), 0, 2L, 2L));
}
+
+ @Test
+ public void testBucketsTableWithPartitionFilter() throws Exception {
+ PredicateBuilder builder = new PredicateBuilder(BucketsTable.TABLE_TYPE);
+
+ Predicate filter = builder.equal(0, BinaryString.fromString("{1}"));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("{1}"), 0, 2L, 2L));
+
+ filter = builder.equal(0, BinaryString.fromString("{2}"));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("{2}"), 0, 2L, 2L));
+
+ filter =
+ builder.in(
+ 0,
+ Arrays.asList(
+ BinaryString.fromString("{1}"), BinaryString.fromString("{2}")));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("{1}"), 0, 2L, 2L),
+ GenericRow.of(BinaryString.fromString("{2}"), 0, 2L, 2L));
+
+ filter = builder.equal(0, BinaryString.fromString("{999}"));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4})).isEmpty();
+ }
+
+ @Test
+ public void testBucketsTableWithBucketFilter() throws Exception {
+ PredicateBuilder builder = new PredicateBuilder(BucketsTable.TABLE_TYPE);
+
+ Predicate filter = builder.equal(1, 0);
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("{1}"), 0, 2L, 2L),
+ GenericRow.of(BinaryString.fromString("{2}"), 0, 2L, 2L));
+
+ filter = builder.equal(1, 1);
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4})).isEmpty();
+ }
+
+ @Test
+ public void testBucketsTableWithCombinedFilter() throws Exception {
+ PredicateBuilder builder = new PredicateBuilder(BucketsTable.TABLE_TYPE);
+
+ Predicate filter =
+ PredicateBuilder.and(
+ builder.equal(0, BinaryString.fromString("{1}")), builder.equal(1, 0));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("{1}"), 0, 2L, 2L));
+
+ filter =
+ PredicateBuilder.and(
+ builder.equal(0, BinaryString.fromString("{1}")), builder.equal(1, 1));
+ assertThat(readWithFilter(bucketsTable, filter, new int[] {0, 1, 2, 4})).isEmpty();
+ }
+
+ private List<InternalRow> readWithFilter(Table table, Predicate filter, int[] projection)
+ throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(filter);
+ if (projection != null) {
+ readBuilder.withProjection(projection);
+ }
+ RecordReader<InternalRow> reader =
+ readBuilder.newRead().createReader(readBuilder.newScan().plan());
+ InternalRowSerializer serializer =
+ new InternalRowSerializer(
+ projection == null
+ ? table.rowType()
+ : Projection.of(projection).project(table.rowType()));
+ List<InternalRow> rows = new ArrayList<>();
+ reader.forEachRemaining(row -> rows.add(serializer.copy(row)));
+ return rows;
+ }
}