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 78d169dad6 [format] Drop filters on unprojected columns before Parquet
conversion (#9858)
78d169dad6 is described below
commit 78d169dad6c6ddc60889954d3bc6322625663d9f
Author: zhigang <[email protected]>
AuthorDate: Thu Sep 17 16:24:37 2026 +0800
[format] Drop filters on unprojected columns before Parquet conversion
(#9858)
---
.../paimon/table/AppendOnlySimpleTableTest.java | 47 ++++++++++++++++++++
.../format/parquet/ParquetReaderFactory.java | 21 ++++++++-
.../format/parquet/ParquetReadWriteTest.java | 51 ++++++++++++++++++++++
3 files changed, 118 insertions(+), 1 deletion(-)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
index dd6f531f8c..9b8fda8e24 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
@@ -1104,6 +1104,53 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testParquetFilterOnUnprojectedColumn(boolean fileIndexEnabled)
throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("status", DataTypes.STRING())
+ .build();
+ FileStoreTable table =
+ createUnawareBucketFileStoreTable(
+ rowType,
+ options -> {
+ options.set(FILE_FORMAT, FILE_FORMAT_PARQUET);
+ options.set(WRITE_ONLY, true);
+ if (fileIndexEnabled) {
+ options.set("file-index.bitmap.columns",
"status");
+ }
+ });
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(0, BinaryString.fromString("A")));
+ write.write(GenericRow.of(1, BinaryString.fromString("B")));
+ write.write(GenericRow.of(2, null));
+ write.write(GenericRow.of(3, BinaryString.fromString("A")));
+ commit.commit(write.prepareCommit());
+ }
+
+ ReadBuilder readBuilder =
+ table.newReadBuilder()
+ .withFilter(
+ new PredicateBuilder(rowType)
+ .equal(1,
BinaryString.fromString("A")))
+ .withReadType(rowType.project(new int[] {0}));
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan().splits())) {
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(1);
+ ids.add(row.getInt(0));
+ });
+ }
+ // ReadBuilder filtering is inclusive: all matching rows must survive
projection.
+ assertThat(ids).contains(0, 3);
+ }
+
@Test
public void testTopNResultFilterParquetRowRanges() throws Exception {
RowType rowType =
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
index ab7d277a62..e66edb6f20 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
@@ -34,6 +34,8 @@ import
org.apache.paimon.format.shredding.ShreddingReadPlanFactory;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.ReadBatchSizer;
import org.apache.paimon.types.ArrayType;
@@ -69,9 +71,12 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.IntFunction;
@@ -113,7 +118,21 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
this.readType = readType;
this.batchSize = batchSize;
this.caseSensitive =
conf.getOptional(CatalogOptions.CASE_SENSITIVE).orElse(true);
- this.predicates = predicates;
+ this.predicates = predicates == null ? null : new ArrayList<>();
+ if (predicates != null) {
+ Set<String> projectedFields =
+ caseSensitive ? new HashSet<>() : new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ projectedFields.addAll(readType.getFieldNames());
+ for (Predicate predicate : predicates) {
+ // Parquet treats unprojected filter columns as null. Keep
only conjuncts
+ // covered by the projection; an OR with an unprojected field
must be dropped.
+ for (Predicate conjunct :
PredicateBuilder.splitAnd(predicate)) {
+ if
(projectedFields.containsAll(PredicateVisitor.collectFieldNames(conjunct))) {
+ this.predicates.add(conjunct);
+ }
+ }
+ }
+ }
}
@VisibleForTesting
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
index fb03ac3778..a9774a8b51 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
@@ -36,6 +36,7 @@ import
org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.ReadBatchSizer;
import org.apache.paimon.reader.RecordReader;
@@ -81,6 +82,7 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
@@ -494,6 +496,55 @@ public class ParquetReadWriteTest {
innerTestTypes(folder, values, rowGroupSize);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testReadWithUnprojectedFilter(boolean conjunction) throws IOException
{
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("status", DataTypes.STRING())
+ .build();
+ Path path =
+ createTempParquetFileByPaimon(
+ folder,
+ Arrays.asList(
+ GenericRow.of(0, BinaryString.fromString("A")),
+ GenericRow.of(1,
BinaryString.fromString("B"))),
+ 1024,
+ rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Predicate status = builder.equal(1, BinaryString.fromString("A"));
+ Predicate id = builder.equal(0, -1);
+ Predicate predicate =
+ conjunction ? PredicateBuilder.and(status, id) :
PredicateBuilder.or(status, id);
+
+ // Exercise page pruning without row-group filters masking its
behavior.
+ Options options = new Options();
+ options.setString("parquet.filter.stats.enabled", "false");
+ options.setString("parquet.filter.dictionary.enabled", "false");
+ options.setString("parquet.filter.bloom.enabled", "false");
+ ParquetReaderFactory factory =
+ new ParquetReaderFactory(
+ options,
+ rowType.project(new int[] {0}),
+ 1024,
+ Collections.singletonList(predicate));
+ LocalFileIO io = LocalFileIO.create();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ factory.createReader(
+ new FormatReaderContext(io, path,
io.getFileSize(path), null, null))) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ if (conjunction) {
+ // AND must retain the projected id predicate, which excludes
every row.
+ assertThat(ids).isEmpty();
+ } else {
+ // OR must retain the status match, even though its column is not
projected.
+ assertThat(ids).contains(0);
+ }
+ }
+
@ParameterizedTest
@MethodSource("parameters")
void testProjection(int rowGroupSize) throws IOException {
