[core] Do not push filters on row tracking fields into append file stats and readers

paimon-api/src/main/java/org/apache/paimon/table/SpecialFields.java

@@ -102,6 +102,9 @@ public class SpecialFields {
                             ROW_ID.name())
                     .collect(Collectors.toSet());
 
+    public static final Set<String> ROW_TRACKING_FIELD_NAMES =
+            Stream.of(ROW_ID.name(), SEQUENCE_NUMBER.name()).collect(Collectors.toSet());
+
     public static boolean isSystemField(int fieldId) {
         return fieldId >= SYSTEM_FIELD_ID_START;
     }

paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java

@@ -651,6 +651,14 @@ public static List<Predicate> excludePredicateWithFields(
                 .collect(Collectors.toList());
     }
 
+    /** Excludes the predicates that reference any of the fields. */
+    public static List<Predicate> excludePredicateReferencingFields(
+            List<Predicate> predicates, Set<String> fields) {
+        return predicates.stream()
+                .filter(f -> Collections.disjoint(PredicateVisitor.collectFieldNames(f), fields))
+                .collect(Collectors.toList());
+    }
+
     @Nullable
     public static Predicate partition(
             Map<String, String> map, RowType rowType, String defaultPartValue) {

paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java

@@ -71,7 +71,9 @@
 import java.util.Map;
 import java.util.function.Function;
 
+import static org.apache.paimon.predicate.PredicateBuilder.excludePredicateReferencingFields;
 import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
+import static org.apache.paimon.table.SpecialFields.ROW_TRACKING_FIELD_NAMES;
 
 /** A {@link SplitRead} to read raw file directly from {@link DataSplit}. */
 public class RawFileSplitRead implements SplitRead<InternalRow> {
@@ -94,6 +96,8 @@ public class RawFileSplitRead implements SplitRead<InternalRow> {
 
     private RowType readRowType;
     @Nullable private List<Predicate> filters;
+    // true if some filters are not pushed down, so TopN and limit must wait for them
+    private boolean hasResidualFilters;
     @Nullable private TopN topN;
     @Nullable private Long limit;
     @Nullable private ReadBatchSizer readBatchSizer;
@@ -146,6 +150,13 @@ public SplitRead<InternalRow> withReadType(RowType readRowType) {
     public RawFileSplitRead withFilter(Predicate predicate) {
         if (predicate != null) {
             this.filters = splitAnd(predicate);
+            if (rowTrackingEnabled) {
+                // Row ids and sequence numbers may only be in the file meta, not in the file.
+                List<Predicate> pushed =
+                        excludePredicateReferencingFields(filters, ROW_TRACKING_FIELD_NAMES);
+                this.hasResidualFilters = pushed.size() < filters.size();
+                this.filters = pushed;
+            }
         }
         return this;
     }
@@ -216,7 +227,9 @@ public RecordReader<InternalRow> createReader(
 
         RowType outputRowType = readRowType;
         Builder formatReaderMappingBuilder =
-                createFormatReaderMappingBuilder(outputRowType, topN, intLimit(limit));
+                hasResidualFilters
+                        ? createFormatReaderMappingBuilder(outputRowType, null, null)
+                        : createFormatReaderMappingBuilder(outputRowType, topN, intLimit(limit));
 
         for (DataFileMeta file : files) {
             suppliers.add(
@@ -232,7 +245,7 @@ public RecordReader<InternalRow> createReader(
 
         RecordReader<InternalRow> reader = ConcatRecordReader.create(suppliers);
         // Apply the final limit after deletion vectors when no later predicate can drop rows.
-        if (topN == null && (filters == null || filters.isEmpty())) {
+        if (topN == null && !hasResidualFilters && (filters == null || filters.isEmpty())) {
             return LimitRecordReader.limit(reader, limit);
         }
         return reader;

paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java

@@ -53,6 +53,11 @@
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 
+import static org.apache.paimon.predicate.PredicateBuilder.andNullable;
+import static org.apache.paimon.predicate.PredicateBuilder.excludePredicateReferencingFields;
+import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
+import static org.apache.paimon.table.SpecialFields.ROW_TRACKING_FIELD_NAMES;
+
 /** {@link FileStoreTable} for append table. */
 public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
 
@@ -109,7 +114,16 @@ public boolean supportStreamingReadOverwrite() {
 
     @Override
     protected BiConsumer<FileStoreScan, Predicate> nonPartitionFilterConsumer() {
-        return (scan, predicate) -> ((AppendOnlyFileStoreScan) scan).withFilter(predicate);
+        if (!coreOptions().rowTrackingEnabled() || coreOptions().dataEvolutionEnabled()) {
+            return (scan, predicate) -> ((AppendOnlyFileStoreScan) scan).withFilter(predicate);
+        }
+        // File stats have no row ids and sequence numbers, leave filters on them to the reader.
+        return (scan, predicate) ->
+                ((AppendOnlyFileStoreScan) scan)
+                        .withFilter(
+                                andNullable(
+                                        excludePredicateReferencingFields(
+                                                splitAnd(predicate), ROW_TRACKING_FIELD_NAMES)));
     }
 
     @Override

paimon-core/src/test/java/org/apache/paimon/table/system/RowTrackingTableTest.java

@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.SubstringTransform;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.paimon.data.DataFormatTestUtil.toStringNoRowKind;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RowTrackingTable}. */
+public class RowTrackingTableTest extends TableTestBase {
+
+    @Test
+    public void testFilterOnRowTrackingFields() throws Exception {
+        Identifier id = identifier("T");
+        catalog.createTable(
+                id,
+                Schema.newBuilder()
+                        .column("a", DataTypes.INT())
+                        .column("b", DataTypes.INT())
+                        .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+                        .option(CoreOptions.BUCKET.key(), "-1")
+                        .build(),
+                false);
+        write(catalog.getTable(id), GenericRow.of(1, 10), GenericRow.of(2, 20));
+        write(catalog.getTable(id), GenericRow.of(3, 30));
+
+        for (Integer limit : Arrays.asList(null, 1)) {
+            assertThat(read(id, "_ROW_ID", 2L, limit)).containsExactly("3, 30, 2, 2");
+            assertThat(read(id, "_SEQUENCE_NUMBER", 2L, limit)).containsExactly("3, 30, 2, 2");
+        }
+
+        // after a schema change, the files are read with an older schema
+        catalog.alterTable(id, SchemaChange.addColumn("c", DataTypes.INT()), false);
+        assertThat(read(id, "_ROW_ID", 2L, null)).containsExactly("3, 30, NULL, 2, 2");
+    }
+
+    @Test
+    public void testFilterMixingRowTrackingAndDataFields() throws Exception {
+        Identifier id = identifier("T");
+        catalog.createTable(
+                id,
+                Schema.newBuilder()
+                        .column("s", DataTypes.STRING())
+                        .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+                        .option(CoreOptions.BUCKET.key(), "-1")
+                        .build(),
+                false);
+        BinaryString abc = BinaryString.fromString("abc");
+        write(catalog.getTable(id), GenericRow.of(abc), GenericRow.of(abc), GenericRow.of(abc));
+        catalog.alterTable(id, SchemaChange.addColumn("c", DataTypes.INT()), false);
+
+        // SUBSTRING(s, _ROW_ID, 1) = 'b' only matches row id 2
+        RowTrackingTable table = new RowTrackingTable((FileStoreTable) catalog.getTable(id));
+        RowType type = table.rowType();
+        Predicate filter =
+                new PredicateBuilder(type)
+                        .equal(
+                                new SubstringTransform(
+                                        Arrays.asList(
+                                                new FieldRef(0, "s", DataTypes.STRING()),
+                                                new FieldRef(
+                                                        type.getFieldIndex("_ROW_ID"),
+                                                        "_ROW_ID",
+                                                        DataTypes.BIGINT()),
+                                                1)),
+                                BinaryString.fromString("b"));
+        for (Integer limit : Arrays.asList(null, 1)) {
+            assertThat(read(table, filter, limit)).containsExactly("abc, NULL, 2, 1");
+        }
+    }
+
+    private List<String> read(Identifier id, String field, long value, @Nullable Integer limit)
+            throws Exception {
+        RowTrackingTable table = new RowTrackingTable((FileStoreTable) catalog.getTable(id));
+        Predicate filter =
+                new PredicateBuilder(table.rowType())
+                        .equal(table.rowType().getFieldIndex(field), value);
+        return read(table, filter, limit);
+    }
+
+    private static List<String> read(
+            RowTrackingTable table, Predicate filter, @Nullable Integer limit) throws Exception {
+        ReadBuilder readBuilder = table.newReadBuilder().withFilter(filter);
+        if (limit != null) {
+            readBuilder.withLimit(limit);
+        }
+        List<String> result = new ArrayList<>();
+        try (RecordReader<InternalRow> reader =
+                readBuilder.newRead().executeFilter().createReader(readBuilder.newScan().plan())) {
+            reader.forEachRemaining(row -> result.add(toStringNoRowKind(row, table.rowType())));
+        }
+        return result;
+    }
+}