[Bug] A filter on _ROW_ID cannot be pushed down: AssertionError normally, NPE after a schema change

[PDGGK]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

master (77ffb4a)

Compute Engine

Reproduced at the table API level, so engine-independent.

Minimal reproduce step

An append table with row tracking on, a filter on _ROW_ID pushed into the scan:

Schema schema =
        Schema.newBuilder()
                .column("a", DataTypes.INT())
                .column("b", DataTypes.INT())
                .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
                .option(CoreOptions.BUCKET.key(), "-1")
                .build();
catalog.createTable(id, schema, false);
FileStoreTable table = (FileStoreTable) catalog.getTable(id);
write(table, GenericRow.of(1, 10), GenericRow.of(2, 20));

RowTrackingTable rt = new RowTrackingTable(table);
RowType type = rt.rowType();                       // a, b, _ROW_ID, _SEQUENCE_NUMBER
Predicate p = new PredicateBuilder(type).equal(type.getFieldIndex("_ROW_ID"), 0L);
rt.newReadBuilder().withFilter(p).newScan().plan();

What doesn't meet your expectations?

The scan fails, and it fails differently depending on whether the table has been altered:

before ALTER : java.lang.AssertionError: index (2) should < 2
after  ALTER : java.lang.NullPointerException: Find no field _ROW_ID
   (catalog.alterTable(id, SchemaChange.addColumn("c", DataTypes.INT()), false))

I expected the predicate either to be applied, or to be dropped so the engine evaluates it after the scan — the way a filter on a field that is missing from a data file is already handled.

Anything else?

The two failures have different causes and I think only the second one is obvious from reading:

After ALTER — SchemaEvolutionUtil.devolveFilters resolves the predicate's field against the table schema and hard-fails when it is absent:

// SchemaEvolutionUtil.java:162-165
DataField tableField =
        checkNotNull(
                nameToTableFields.get(fieldRef.name()),
                String.format("Find no field %s", fieldRef.name()));

nameToTableFields comes from TableSchema.fields(), which never contains the row-tracking fields — SpecialFields.rowTypeWithRowTracking appends _ROW_ID and _SEQUENCE_NUMBER at read time and explicitly rejects a schema that already declares them. Note the asymmetry three lines below: a field that is in the table schema but absent from the data file is handled gracefully (dataField == null → drop, or keep when keepNewFieldFilter). Only a field absent from the table schema throws.

This path is reached only when the file's schema id differs from the table's — SimpleStatsEvolutions.tryDevolveFilter and filterUnsafeFilter both return early when they are equal — which is why the ALTER is what changes the error.

Before ALTER — AssertionError: index (2) should < 2. _ROW_ID is index 2 in the projected row type, and something downstream is applying that index to a 2-field row. I have not traced this one to a line.

So I don't think "make devolveFilters drop the predicate" is the whole fix: it would turn the second failure back into the first, and the query would still not run. It looks like the question is whether a predicate over the row-tracking projection should be pushed into the scan at all, and if so how its indices should be mapped — which seemed like a decision for you rather than something to guess at, hence an issue rather than a PR.

One thing I could not determine: whether Flink or Spark actually push a _ROW_ID predicate down, or evaluate it after the scan. If they never push it, this is only reachable through the table API and the priority is correspondingly lower. There are no existing tests filtering on _ROW_ID, so I could not settle it from the repo.

Happy to send a patch once you say which direction you want.

Are you willing to submit a PR?

• I'm willing to submit a PR!