[mosaic] Resolve row-group filter fields by name

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<Predicate> resolvePredicates(
+            @Nullable List<Predicate> predicates, RowType dataSchemaRowType) {
+        if (predicates == null || predicates.isEmpty()) {
+            return predicates;
+        }
+        List<Predicate> 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;
     }

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<String, String> options = new HashMap<>();
+        options.put(MosaicFileFormat.STATS_COLUMNS.key(), statsColumns);
+        FormatTable table = buildFormatTable(rowType, Collections.singletonList("dt"), options);
+        List<InternalRow> 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<String> partitionKeys, Map<String, String> options) {
         String location = new Path(tempPath.toUri()).toString();
@@ -218,7 +269,15 @@ private static void writeAll(FormatTable table, List<InternalRow> rows) throws E
     }
 
     private static List<InternalRow> readAll(FormatTable table, RowType rowType) throws Exception {
+        return readAll(table, rowType, null);
+    }
+
+    private static List<InternalRow> readAll(
+            FormatTable table, RowType rowType, @Nullable Predicate filter) throws Exception {
         ReadBuilder readBuilder = table.newReadBuilder();
+        if (filter != null) {
+            readBuilder.withFilter(filter);
+        }
         List<Split> splits = readBuilder.newScan().plan().splits();
         assertThat(splits).isNotEmpty();