This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit ef9423f555bea3553d772830901c615484eebbdb 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 {
