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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 15f8c9a72 [core] Support filter pushdown in FilesTable (#2808)
15f8c9a72 is described below
commit 15f8c9a723123a8f482cce314aff657833be7919
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jan 30 11:59:54 2024 +0800
[core] Support filter pushdown in FilesTable (#2808)
---
.../table/source/AbstractInnerTableScan.java | 14 ++++
.../apache/paimon/table/source/InnerTableScan.java | 9 +++
.../paimon/table/source/InnerTableScanImpl.java | 7 --
.../apache/paimon/table/system/AuditLogTable.java | 24 ++++++
.../org/apache/paimon/table/system/FilesTable.java | 91 ++++++++++++++++++++--
.../apache/paimon/table/system/FilesTableTest.java | 38 ++++++++-
.../apache/paimon/flink/CatalogTableITCase.java | 27 +++++++
7 files changed, 197 insertions(+), 13 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
index fd605b64e..16246c113 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
@@ -48,6 +48,7 @@ import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
@@ -70,11 +71,24 @@ public abstract class AbstractInnerTableScan implements
InnerTableScan {
return this;
}
+ @Override
public AbstractInnerTableScan withBucketFilter(Filter<Integer>
bucketFilter) {
snapshotReader.withBucketFilter(bucketFilter);
return this;
}
+ @Override
+ public AbstractInnerTableScan withPartitionFilter(Map<String, String>
partitionSpec) {
+ snapshotReader.withPartitionFilter(partitionSpec);
+ return this;
+ }
+
+ @Override
+ public AbstractInnerTableScan withLevelFilter(Filter<Integer> levelFilter)
{
+ snapshotReader.withLevelFilter(levelFilter);
+ return this;
+ }
+
public AbstractInnerTableScan withMetricsRegistry(MetricRegistry
metricsRegistry) {
snapshotReader.withMetricRegistry(metricsRegistry);
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
index 234d82e52..613b2efc2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table.source;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.utils.Filter;
import java.util.Map;
@@ -36,6 +37,14 @@ public interface InnerTableScan extends TableScan {
return this;
}
+ default InnerTableScan withBucketFilter(Filter<Integer> bucketFilter) {
+ return this;
+ }
+
+ default InnerTableScan withLevelFilter(Filter<Integer> levelFilter) {
+ return this;
+ }
+
default InnerTableScan withMetricsRegistry(MetricRegistry metricRegistry) {
// do nothing, should implement this if need
return this;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java
index a30d82f8c..1e892fe1f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java
@@ -30,7 +30,6 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
/** {@link TableScan} implementation for batch planning. */
public class InnerTableScanImpl extends AbstractInnerTableScan {
@@ -60,12 +59,6 @@ public class InnerTableScanImpl extends
AbstractInnerTableScan {
return this;
}
- @Override
- public InnerTableScan withPartitionFilter(Map<String, String>
partitionSpec) {
- snapshotReader.withPartitionFilter(partitionSpec);
- return this;
- }
-
@Override
public InnerTableScan withLimit(int limit) {
this.pushDownLimit = limit;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index fead5068b..96c2621b6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -315,6 +315,30 @@ public class AuditLogTable implements DataTable,
ReadonlyTable {
return this;
}
+ @Override
+ public InnerTableScan withLimit(int limit) {
+ batchScan.withLimit(limit);
+ return this;
+ }
+
+ @Override
+ public InnerTableScan withPartitionFilter(Map<String, String>
partitionSpec) {
+ batchScan.withPartitionFilter(partitionSpec);
+ return this;
+ }
+
+ @Override
+ public InnerTableScan withBucketFilter(Filter<Integer> bucketFilter) {
+ batchScan.withBucketFilter(bucketFilter);
+ return this;
+ }
+
+ @Override
+ public InnerTableScan withLevelFilter(Filter<Integer> levelFilter) {
+ batchScan.withLevelFilter(levelFilter);
+ return this;
+ }
+
@Override
public Plan plan() {
return batchScan.plan();
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 0949de2df..899e7e4e4 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
@@ -19,13 +19,17 @@
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.LazyGenericRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.format.FieldStats;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.LeafPredicate;
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;
@@ -54,12 +58,15 @@ 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.util.ArrayList;
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;
@@ -142,19 +149,47 @@ public class FilesTable implements ReadonlyTable {
private final FileStoreTable storeTable;
+ @Nullable private LeafPredicate partitionPredicate;
+ @Nullable private LeafPredicate bucketPredicate;
+ @Nullable private LeafPredicate levelPredicate;
+
private FilesScan(FileStoreTable storeTable) {
this.storeTable = storeTable;
}
@Override
- public InnerTableScan withFilter(Predicate predicate) {
- // TODO
+ public InnerTableScan withFilter(Predicate pushdown) {
+ List<Predicate> predicates = PredicateBuilder.splitAnd(pushdown);
+ for (Predicate predicate : predicates) {
+ if (predicate instanceof LeafPredicate) {
+ LeafPredicate leaf = (LeafPredicate) predicate;
+ switch (leaf.fieldName()) {
+ case "partition":
+ this.partitionPredicate = leaf;
+ break;
+ case "bucket":
+ this.bucketPredicate = leaf;
+ break;
+ case "level":
+ this.levelPredicate = leaf;
+ break;
+ default:
+ break;
+ }
+ }
+ }
return this;
}
@Override
public Plan innerPlan() {
- return () -> Collections.singletonList(new FilesSplit(storeTable));
+ return () ->
+ Collections.singletonList(
+ new FilesSplit(
+ storeTable,
+ partitionPredicate,
+ bucketPredicate,
+ levelPredicate));
}
}
@@ -164,8 +199,19 @@ public class FilesTable implements ReadonlyTable {
private final FileStoreTable storeTable;
- private FilesSplit(FileStoreTable storeTable) {
+ @Nullable private final LeafPredicate partitionPredicate;
+ @Nullable private final LeafPredicate bucketPredicate;
+ @Nullable private final LeafPredicate levelPredicate;
+
+ private FilesSplit(
+ FileStoreTable storeTable,
+ @Nullable LeafPredicate partitionPredicate,
+ @Nullable LeafPredicate bucketPredicate,
+ @Nullable LeafPredicate levelPredicate) {
this.storeTable = storeTable;
+ this.partitionPredicate = partitionPredicate;
+ this.bucketPredicate = bucketPredicate;
+ this.levelPredicate = levelPredicate;
}
@Override
@@ -178,7 +224,42 @@ public class FilesTable implements ReadonlyTable {
}
private TableScan.Plan plan() {
- return storeTable.newScan().plan();
+ InnerTableScan scan = storeTable.newScan();
+ if (partitionPredicate != null) {
+ if (partitionPredicate.function() instanceof Equal) {
+ String partitionStr =
partitionPredicate.literals().get(0).toString();
+ if (partitionStr.startsWith("[")) {
+ partitionStr = partitionStr.substring(1);
+ }
+ if (partitionStr.endsWith("]")) {
+ partitionStr = partitionStr.substring(0,
partitionStr.length() - 1);
+ }
+ String[] partFields = partitionStr.split(", ");
+ LinkedHashMap<String, String> partSpec = new
LinkedHashMap<>();
+ List<String> partitionKeys = storeTable.partitionKeys();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ partSpec.put(partitionKeys.get(i), partFields[i]);
+ }
+ scan.withPartitionFilter(partSpec);
+ }
+ // TODO support range?
+ }
+ if (bucketPredicate != null) {
+ scan.withBucketFilter(
+ bucket -> {
+ // bucket index: 1
+ return bucketPredicate.test(GenericRow.of(null,
bucket));
+ });
+ }
+ if (levelPredicate != null) {
+ scan.withLevelFilter(
+ level -> {
+ // level index: 5
+ return levelPredicate.test(
+ GenericRow.of(null, null, null, null,
null, level));
+ });
+ }
+ return scan.plan();
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/FilesTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/FilesTableTest.java
index da8730861..068d5d33a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/system/FilesTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/FilesTableTest.java
@@ -31,6 +31,8 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
@@ -38,6 +40,7 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.SnapshotManager;
@@ -45,17 +48,18 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.FileNotFoundException;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FilesTable}. */
public class FilesTableTest extends TableTestBase {
- private static final String tableName = "MyTable";
private FileStoreTable table;
private FileStoreScan scan;
@@ -64,6 +68,7 @@ public class FilesTableTest extends TableTestBase {
@BeforeEach
public void before() throws Exception {
+ String tableName = "MyTable";
FileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, tableName));
Schema schema =
@@ -94,6 +99,37 @@ public class FilesTableTest extends TableTestBase {
write(table, GenericRow.of(2, 1, 3), GenericRow.of(2, 2, 4));
}
+ @Test
+ public void testReadWithFilter() throws Exception {
+ compact(table, row(2), 0);
+ write(table, GenericRow.of(3, 1, 1));
+ assertThat(readPartBucketLevel(null))
+ .containsExactlyInAnyOrder("[1]-0-0", "[1]-0-0", "[1]-1-0",
"[2]-0-5");
+
+ PredicateBuilder builder = new PredicateBuilder(FilesTable.TABLE_TYPE);
+ assertThat(readPartBucketLevel(builder.equal(0, "[2]")))
+ .containsExactlyInAnyOrder("[2]-0-5");
+ assertThat(readPartBucketLevel(builder.equal(1,
1))).containsExactlyInAnyOrder("[1]-1-0");
+ assertThat(readPartBucketLevel(builder.equal(5,
5))).containsExactlyInAnyOrder("[2]-0-5");
+ }
+
+ private List<String> readPartBucketLevel(Predicate predicate) throws
IOException {
+ ReadBuilder readBuilder =
filesTable.newReadBuilder().withFilter(predicate);
+ List<String> rows = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(
+ row ->
+ rows.add(
+ row.getString(0)
+ + "-"
+ + row.getInt(1)
+ + "-"
+ + row.getInt(5)));
+ return rows;
+ }
+
@Test
public void testReadFilesFromLatest() throws Exception {
List<InternalRow> expectedRow = getExceptedResult(2L);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 175a438bd..150caf7b0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -37,6 +37,7 @@ import javax.annotation.Nonnull;
import java.util.List;
import java.util.stream.Collectors;
+import static
org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYNC;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -630,6 +631,32 @@ public class CatalogTableITCase extends CatalogITCaseBase {
: "[5],[10]")));
}
+ @Test
+ public void testFilesTableWithFilter() {
+ tEnv.getConfig().set(TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_WITH_FILTER (k INT, p INT, v INT, PRIMARY KEY
(k, p) NOT ENFORCED) "
+ + "PARTITIONED BY (p) WITH ('bucket'='2')");
+ sql("INSERT INTO T_WITH_FILTER VALUES (1, 2, 3), (4, 5, 6)");
+ sql("CALL sys.compact('default.T_WITH_FILTER')");
+ sql("INSERT INTO T_WITH_FILTER VALUES (7, 8, 9)");
+
+ assertThat(sql("SELECT `partition`, bucket, level FROM
T_WITH_FILTER$files"))
+ .containsExactlyInAnyOrder(
+ Row.of("[2]", 0, 5), Row.of("[5]", 0, 5),
Row.of("[8]", 1, 0));
+
+ assertThat(
+ sql(
+ "SELECT `partition`, bucket, level FROM
T_WITH_FILTER$files WHERE `partition`='[2]'"))
+ .containsExactlyInAnyOrder(Row.of("[2]", 0, 5));
+
+ assertThat(sql("SELECT `partition`, bucket, level FROM
T_WITH_FILTER$files WHERE bucket=0"))
+ .containsExactlyInAnyOrder(Row.of("[2]", 0, 5), Row.of("[5]",
0, 5));
+
+ assertThat(sql("SELECT `partition`, bucket, level FROM
T_WITH_FILTER$files WHERE level=0"))
+ .containsExactlyInAnyOrder(Row.of("[8]", 1, 0));
+ }
+
@Nonnull
private List<String> getRowStringList(List<Row> rows) {
return rows.stream()
