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 c2057c32d5 [core][spark] Push down object table filters (#9383)
c2057c32d5 is described below
commit c2057c32d545cbf86feadb9116ef8205517d0a22
Author: Wenchao Wu <[email protected]>
AuthorDate: Wed Aug 26 10:55:39 2026 +0800
[core][spark] Push down object table filters (#9383)
---
.../paimon/table/object/ObjectTableImpl.java | 6 +-
.../paimon/table/object/ObjectTableTest.java | 96 ++++++++++++++++++++++
.../apache/paimon/spark/read/ObjectTableScan.scala | 9 +-
.../paimon/spark/table/PaimonObjectTableTest.scala | 55 +++++++++----
4 files changed, 148 insertions(+), 18 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTableImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTableImpl.java
index 83f3c53536..f8548b3167 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTableImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTableImpl.java
@@ -209,10 +209,11 @@ public class ObjectTableImpl implements ReadonlyTable,
ObjectTable {
private static class ObjectRead implements InnerTableRead {
private @Nullable RowType readType;
+ private @Nullable Predicate predicate;
@Override
public InnerTableRead withFilter(Predicate predicate) {
- // TODO
+ this.predicate = predicate;
return this;
}
@@ -261,6 +262,9 @@ public class ObjectTableImpl implements ReadonlyTable,
ObjectTable {
}
}
};
+ if (predicate != null) {
+ iterator = Iterators.filter(iterator, predicate::test);
+ }
if (readType != null) {
iterator =
Iterators.transform(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/object/ObjectTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/object/ObjectTableTest.java
new file mode 100644
index 0000000000..f9d5e33b17
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/object/ObjectTableTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.object;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.InternalRow;
+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.table.source.ReadBuilder;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ObjectTableImpl}. */
+public class ObjectTableTest {
+
+ @TempDir java.nio.file.Path tempPath;
+ private ObjectTable table;
+
+ @BeforeEach
+ public void beforeEach() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path objectPath = new Path(tempPath.toUri());
+ fileIO.writeFile(new Path(objectPath, "drop.txt"), "drop", false);
+ fileIO.writeFile(new Path(objectPath, "dir/keep.txt"), "keep", false);
+
+ table =
+ ObjectTable.builder()
+ .identifier(Identifier.create("default", "objects"))
+ .fileIO(fileIO)
+ .location(objectPath.toString())
+ .build();
+ }
+
+ @Test
+ public void testReadWithoutFilter() throws Exception {
+ assertThat(readPaths(table.newReadBuilder()))
+ .containsExactlyInAnyOrder("drop.txt", "dir/keep.txt");
+ }
+
+ @Test
+ public void testReadWithFilterAndProjection() throws Exception {
+ Predicate predicate =
+ new PredicateBuilder(ObjectTable.SCHEMA)
+ .equal(1, BinaryString.fromString("keep.txt"));
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withFilter(predicate).withProjection(new int[] {0});
+
+ assertThat(readPaths(readBuilder)).containsExactly("dir/keep.txt");
+ }
+
+ @Test
+ public void testReadWithNoMatches() throws Exception {
+ Predicate predicate =
+ new PredicateBuilder(ObjectTable.SCHEMA)
+ .equal(1, BinaryString.fromString("missing.txt"));
+
+
assertThat(readPaths(table.newReadBuilder().withFilter(predicate))).isEmpty();
+ }
+
+ private List<String> readPaths(ReadBuilder readBuilder) throws Exception {
+ List<String> paths = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row ->
paths.add(row.getString(0).toString()));
+ }
+ return paths;
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/ObjectTableScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/ObjectTableScan.scala
index 4542b4d2a4..ff2f3ca8c7 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/ObjectTableScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/ObjectTableScan.scala
@@ -31,14 +31,17 @@ import scala.collection.JavaConverters._
class ObjectTableScanBuilder(val table: ObjectTable) extends
PaimonBaseScanBuilder {
override def build(): ObjectTableScan =
- ObjectTableScan(table, requiredSchema)
+ ObjectTableScan(table, requiredSchema, pushedDataFilters)
}
/** Scan implementation for [[ObjectTable]] */
-case class ObjectTableScan(table: ObjectTable, requiredSchema: StructType)
extends BaseScan {
+case class ObjectTableScan(
+ table: ObjectTable,
+ requiredSchema: StructType,
+ pushedDataFilters: Seq[Predicate])
+ extends BaseScan {
override val pushedPartitionFilters: Seq[PartitionPredicate] = Nil
- override val pushedDataFilters: Seq[Predicate] = Nil
override val pushedLimit: Option[Int] = None
protected def getInputSplits: Array[Split] = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonObjectTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonObjectTableTest.scala
index 3c3ecee016..be6d6b9815 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonObjectTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonObjectTableTest.scala
@@ -21,6 +21,7 @@ package org.apache.paimon.spark.table
import org.apache.paimon.catalog.Identifier
import org.apache.paimon.fs.Path
import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
+import org.apache.paimon.spark.read.ObjectTableScan
import org.apache.paimon.table.`object`.ObjectTable
import org.apache.spark.sql.Row
@@ -35,7 +36,45 @@ class PaimonObjectTableTest extends
PaimonSparkTestWithRestCatalogBase {
}
test("ObjectTable: read file metadata") {
- val tableName = "object_table_test"
+ withObjectTable("object_table_metadata_test") {
+ tableName =>
+ checkAnswer(
+ sql(s"SELECT path, name FROM $tableName ORDER BY path"),
+ Seq(
+ Row("file1.txt", "file1.txt"),
+ Row("file2.txt", "file2.txt"),
+ Row("subdir/file3.txt", "file3.txt"))
+ )
+
+ // Verify schema has expected columns
+ checkAnswer(
+ sql(s"SELECT COUNT(*) FROM $tableName"),
+ Seq(Row(3))
+ )
+ }
+ }
+
+ test("ObjectTable: push down filters") {
+ withObjectTable("object_table_filter_pushdown_test") {
+ tableName =>
+ val filteredQuery =
+ s"SELECT path FROM $tableName WHERE name = 'file2.txt' AND length >
0"
+ checkAnswer(sql(filteredQuery), Seq(Row("file2.txt")))
+
assert(getScan(filteredQuery).asInstanceOf[ObjectTableScan].pushedDataFilters.size
== 2)
+ }
+ }
+
+ test("ObjectTable: retain unsupported filters for post-scan evaluation") {
+ withObjectTable("object_table_post_scan_filter_test") {
+ tableName =>
+ val filteredQuery =
+ s"SELECT path FROM $tableName WHERE reverse(name) = 'txt.2elif'"
+ checkAnswer(sql(filteredQuery), Seq(Row("file2.txt")))
+
assert(getScan(filteredQuery).asInstanceOf[ObjectTableScan].pushedDataFilters.isEmpty)
+ }
+ }
+
+ private def withObjectTable(tableName: String)(testCode: String => Unit):
Unit = {
withTable(tableName) {
sql(
s"CREATE TABLE $tableName TBLPROPERTIES (" +
@@ -54,19 +93,7 @@ class PaimonObjectTableTest extends
PaimonSparkTestWithRestCatalogBase {
fileIO.mkdirs(new Path(basePath, "subdir"))
fileIO.writeFile(new Path(basePath, "subdir/file3.txt"), "content3",
false)
- checkAnswer(
- sql(s"SELECT path, name FROM $tableName ORDER BY path"),
- Seq(
- Row("file1.txt", "file1.txt"),
- Row("file2.txt", "file2.txt"),
- Row("subdir/file3.txt", "file3.txt"))
- )
-
- // Verify schema has expected columns
- checkAnswer(
- sql(s"SELECT COUNT(*) FROM $tableName"),
- Seq(Row(3))
- )
+ testCode(tableName)
}
}
}
