From 500515898680834a8666041c3a602d9f9bd3c415 Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Wed, 21 Oct 2026 23:03:28 +0900 Subject: [PATCH] [mosaic] Resolve row-group filter fields by name Format Table reads pass filters indexed against the full table row type together with a data row type that excludes partition columns. MosaicRecordsReader evaluated row-group stats by field index, so a partition column declared before the filtered column made the filter read another column's stats: matching row groups were skipped or the read failed with ArrayIndexOutOfBoundsException when mosaic.stats-columns was set. Remap filters by name against the file row type, as Parquet and ORC already do. Generated-by: Claude Code --- .../format/mosaic/MosaicRecordsReader.java | 25 +++++++- .../mosaic/FormatTableMosaicReadTest.java | 59 +++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java index f0e1018b9073..9b1075262f26 100644 --- a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java +++ b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java @@ -31,6 +31,7 @@ import org.apache.paimon.mosaic.ColumnStatistics; import org.apache.paimon.mosaic.MosaicReader; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateRemapper; import org.apache.paimon.reader.FileRecordIterator; import org.apache.paimon.reader.FileRecordReader; import org.apache.paimon.table.SpecialFields; @@ -214,7 +215,7 @@ public MosaicRecordsReader( this.inputFileAdapter = inputFileAdapter; this.dataSchemaRowType = dataSchemaRowType; this.projectedFieldCount = projectedRowType.getFieldCount(); - this.predicates = predicates; + this.predicates = resolvePredicates(predicates, dataSchemaRowType); this.selection = selection; this.allocator = allocator; @@ -298,6 +299,28 @@ static long estimatedRowBytes(RowType projectedRowType) { return Math.max(1, bytes); } + /** + * Resolves predicate fields by name against the file's row type. Callers such as Format Tables + * and primary-key tables pass predicates whose indices refer to a different row type. + */ + @Nullable + private static List resolvePredicates( + @Nullable List predicates, RowType dataSchemaRowType) { + if (predicates == null || predicates.isEmpty()) { + return predicates; + } + List resolved = new ArrayList<>(predicates.size()); + for (Predicate predicate : predicates) { + try { + resolved.add(PredicateRemapper.remap(predicate, dataSchemaRowType)); + } catch (IllegalArgumentException e) { + // The field is not in this file's row type. Dropping the conjunct only relaxes + // row-group pruning. + } + } + return resolved; + } + int prefetchDepth() { return prefetchDepth; } diff --git a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java index db88d765d826..8384d66a48dc 100644 --- a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java +++ b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java @@ -25,6 +25,8 @@ import org.apache.paimon.data.serializer.InternalRowSerializer; 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.FormatTable; import org.apache.paimon.table.sink.BatchTableCommit; @@ -40,6 +42,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -186,6 +190,53 @@ void testInvalidCompressionPropagatedFromWriter() { .hasMessageContaining("Mosaic format only supports zstd"); } + @Test + void testFilterOnDataColumnAfterPartitionColumn() throws Exception { + RowType rowType = partitionFirstRowType(); + FormatTable table = writePartitionFirstTable(rowType, "a,b"); + PredicateBuilder builder = new PredicateBuilder(rowType); + + assertThat(readAll(table, rowType, builder.equal(1, 1))) + .extracting(row -> row.getInt(1)) + .contains(1); + assertThat(readAll(table, rowType, builder.equal(2, 101))) + .extracting(row -> row.getInt(2)) + .contains(101); + assertThat(readAll(table, rowType, builder.greaterOrEqual(1, 105))).isEmpty(); + } + + @Test + void testFilterOnColumnWithoutStatsAfterPartitionColumn() throws Exception { + RowType rowType = partitionFirstRowType(); + FormatTable table = writePartitionFirstTable(rowType, "b"); + PredicateBuilder builder = new PredicateBuilder(rowType); + + assertThat(readAll(table, rowType, builder.equal(1, 1))) + .extracting(row -> row.getInt(1)) + .contains(1); + } + + private static RowType partitionFirstRowType() { + return RowType.builder() + .field("dt", DataTypes.STRING()) + .field("a", DataTypes.INT()) + .field("b", DataTypes.INT()) + .build(); + } + + private FormatTable writePartitionFirstTable(RowType rowType, String statsColumns) + throws Exception { + Map options = new HashMap<>(); + options.put(MosaicFileFormat.STATS_COLUMNS.key(), statsColumns); + FormatTable table = buildFormatTable(rowType, Collections.singletonList("dt"), options); + List rows = new ArrayList<>(); + for (int i = 1; i <= 10; i++) { + rows.add(GenericRow.of(BinaryString.fromString("p"), i, 100 + i)); + } + writeAll(table, rows); + return table; + } + private FormatTable buildFormatTable( RowType rowType, List partitionKeys, Map options) { String location = new Path(tempPath.toUri()).toString(); @@ -218,7 +269,15 @@ private static void writeAll(FormatTable table, List rows) throws E } private static List readAll(FormatTable table, RowType rowType) throws Exception { + return readAll(table, rowType, null); + } + + private static List readAll( + FormatTable table, RowType rowType, @Nullable Predicate filter) throws Exception { ReadBuilder readBuilder = table.newReadBuilder(); + if (filter != null) { + readBuilder.withFilter(filter); + } List splits = readBuilder.newScan().plan().splits(); assertThat(splits).isNotEmpty();