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 d0f993e8e7 [mosaic] Resolve row-group filter fields by name (#10729)
d0f993e8e7 is described below
commit d0f993e8e781000fb9318932106942b3bed84e7e
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Oct 26 13:20:41 2026 +0900
[mosaic] Resolve row-group filter fields by name (#10729)
---
.../paimon/format/mosaic/MosaicRecordsReader.java | 25 ++++++++-
.../format/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 f0e1018b90..9b1075262f 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.fs.Path;
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 class MosaicRecordsReader implements
FileRecordReader<InternalRow> {
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 @@ public class MosaicRecordsReader implements
FileRecordReader<InternalRow> {
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;
}
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 db88d765d8..8384d66a48 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.InternalRow;
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.BeforeAll;
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 @@ class FormatTableMosaicReadTest {
.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 @@ class FormatTableMosaicReadTest {
}
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();
