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 238dd3658c [format] Drop nested filters on unprojected Parquet fields
before conversion (#10379)
238dd3658c is described below
commit 238dd3658cd86f6d1ee830745a0dc67eed104189
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Oct 5 21:41:57 2026 +0900
[format] Drop nested filters on unprojected Parquet fields before
conversion (#10379)
---
.../paimon/table/AppendOnlySimpleTableTest.java | 53 ++++++++++++++++++
.../format/parquet/ParquetReaderFactory.java | 45 +++++++++++++++-
.../format/parquet/ParquetReadWriteTest.java | 63 ++++++++++++++++++++++
3 files changed, 160 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 e6352b8d91..ce4bd35f3e 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
@@ -50,6 +50,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.NestedFieldTransform;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.TopN;
@@ -1209,6 +1210,58 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
assertThat(ids).contains(0, 3);
}
+ @Test
+ public void testParquetFilterOnUnprojectedNestedField() throws Exception {
+ RowType nestedType =
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "a", DataTypes.INT()),
+ DataTypes.FIELD(3, "b", DataTypes.INT()));
+ FileStoreTable table =
+ createUnawareBucketFileStoreTable(
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("s", nestedType)
+ .build(),
+ options -> {
+ options.set(FILE_FORMAT, FILE_FORMAT_PARQUET);
+ options.set(WRITE_ONLY, true);
+ });
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(0, GenericRow.of(10, 7)));
+ write.write(GenericRow.of(1, GenericRow.of(20, 8)));
+ commit.commit(write.prepareCommit());
+ }
+
+ RowType rowType = table.rowType();
+ DataField sField = rowType.getField("s");
+ RowType sType = (RowType) sField.type();
+ // Prune s down to s.a, while filtering on s.b.
+ RowType readType =
+ new RowType(
+ Arrays.asList(
+ rowType.getField("id"),
+ sField.newType(
+ new RowType(
+
Collections.singletonList(sType.getField("a"))))));
+ Predicate filter =
+ new PredicateBuilder(rowType)
+ .equal(
+ new NestedFieldTransform(
+ new FieldRef(1, "s", sType),
+ Collections.singletonList("b")),
+ 7);
+ ReadBuilder readBuilder =
table.newReadBuilder().withFilter(filter).withReadType(readType);
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan().splits())) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ // ReadBuilder filtering is inclusive: the matching row must survive
nested pruning.
+ assertThat(ids).contains(0);
+ }
+
@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 557002831d..5fcee88b50 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,9 @@ import
org.apache.paimon.format.shredding.ShreddingReadPlanFactories;
import org.apache.paimon.format.shredding.ShreddingReadPlanFactory;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.NestedFieldTransform;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.PredicateVisitor;
@@ -130,7 +133,8 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
// 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))) {
+ if
(projectedFields.containsAll(PredicateVisitor.collectFieldNames(conjunct))
+ && nestedPathsProjected(conjunct)) {
this.predicates.add(conjunct);
}
}
@@ -138,6 +142,45 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
}
}
+ /**
+ * Parquet also reads an unrequested nested column as all-null, so a
nested field filter is only
+ * safe when its whole path survives the (possibly pruned) read type.
+ */
+ private boolean nestedPathsProjected(Predicate predicate) {
+ if (predicate instanceof CompoundPredicate) {
+ for (Predicate child : ((CompoundPredicate) predicate).children())
{
+ if (!nestedPathsProjected(child)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (!(predicate instanceof LeafPredicate)
+ || !(((LeafPredicate) predicate).transform() instanceof
NestedFieldTransform)) {
+ return true;
+ }
+ NestedFieldTransform nested =
+ (NestedFieldTransform) ((LeafPredicate) predicate).transform();
+ DataType current = findProjectedField(readType,
nested.fieldRef().name());
+ for (String component : nested.path()) {
+ current =
+ current instanceof RowType
+ ? findProjectedField((RowType) current, component)
+ : null;
+ }
+ return current != null;
+ }
+
+ @Nullable
+ private DataType findProjectedField(RowType rowType, String name) {
+ for (DataField field : rowType.getFields()) {
+ if (caseSensitive ? field.name().equals(name) :
field.name().equalsIgnoreCase(name)) {
+ return field.type();
+ }
+ }
+ return null;
+ }
+
@VisibleForTesting
Map<MessageType, RequestedSchema> requestedSchemaCache() {
return requestedSchemaCache;
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 a9774a8b51..6f4ec92970 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,8 @@ 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.FieldRef;
+import org.apache.paimon.predicate.NestedFieldTransform;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.ReadBatchSizer;
@@ -545,6 +547,67 @@ public class ParquetReadWriteTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testReadWithUnprojectedNestedFilter(boolean conjunction) throws
IOException {
+ RowType nestedType =
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "a", DataTypes.INT()),
+ DataTypes.FIELD(3, "b", DataTypes.INT()));
+ RowType rowType =
+ RowType.builder().field("id", DataTypes.INT()).field("s",
nestedType).build();
+ Path path =
+ createTempParquetFileByPaimon(
+ folder,
+ Arrays.asList(
+ GenericRow.of(0, GenericRow.of(10, 7)),
+ GenericRow.of(1, GenericRow.of(20, 8))),
+ 1024,
+ rowType);
+ // Read only s.a, so a filter on s.b references a pruned nested field.
+ RowType readType =
+ new RowType(
+ Arrays.asList(
+ rowType.getField("id"),
+ rowType.getField("s")
+ .newType(
+ new RowType(
+
Collections.singletonList(
+
nestedType.getField("a"))))));
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ FieldRef s = new FieldRef(1, "s", nestedType);
+ NestedFieldTransform a = new NestedFieldTransform(s,
Collections.singletonList("a"));
+ NestedFieldTransform b = new NestedFieldTransform(s,
Collections.singletonList("b"));
+ Predicate pruned = builder.equal(b, 7);
+
+ if (conjunction) {
+ // AND must drop the pruned s.b conjunct but keep pushing the
projected s.a one.
+ Predicate matching = PredicateBuilder.and(pruned, builder.equal(a,
10));
+ Predicate excluding = PredicateBuilder.and(pruned,
builder.equal(a, -1));
+ assertThat(readIds(path, readType, matching)).contains(0);
+ assertThat(readIds(path, readType, excluding)).isEmpty();
+ } else {
+ // OR touching the pruned s.b must be dropped as a whole.
+ Predicate predicate = PredicateBuilder.or(pruned, builder.equal(0,
-1));
+ assertThat(readIds(path, readType, predicate)).contains(0);
+ }
+ }
+
+ private List<Integer> readIds(Path path, RowType readType, Predicate
predicate)
+ throws IOException {
+ ParquetReaderFactory factory =
+ new ParquetReaderFactory(
+ new Options(), readType, 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)));
+ }
+ return ids;
+ }
+
@ParameterizedTest
@MethodSource("parameters")
void testProjection(int rowGroupSize) throws IOException {
