[Bug] A filter on _ROW_ID cannot be pushed down: AssertionError normally, NPE after a schema change
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] 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:
```java
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:
```java
// 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?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Reproduce the table-API case in the issue with and without the schema change. Read SchemaEvolutionUtil.java around lines 162-165, then trace SimpleStatsEvolutions.tryDevolveFilter, filterUnsafeFilter, and RowTrackingTable's projected fields. Done means the _ROW_ID filter no longer fails in either case, with tests covering both paths and the intended pushdown behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, databases
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100