kuni-92 opened a new issue, #18499:
URL: https://github.com/apache/iceberg/issues/18499
**Apache Iceberg version:** 1.12.0 (reproduced). The code involved appears
unchanged on `main` (`flink/v2.3`), so the same behaviour is expected there.
**Query engine:** Flink (`DynamicIcebergSink`, DataStream API, upsert mode)
**Please describe the bug 🐞**
### Summary
`DynamicIcebergSink.TableMetadataCache.schema()` iterates **all historical
schemas** of a table by schema id ascending and returns the first one that
`CompareSchemasVisitor` marks `SAME` with the input. The visitor compares by
field **name** (via `PartnerIdByNameAccessors`) and does not consider the
input's field ids, so when a dropped column is reused — the typical case is
`DROP COLUMN c` → `ADD COLUMN c <same type>` — the matcher picks the historical
schema that still holds the retired field id. The writer then writes the
Parquet file against the retired id, and the reader with the current schema
cannot locate the field and returns `null` (or `initial-default` on tables with
that feature).
The same name-first matching also masks changes that would otherwise trigger
the sink's auto schema evolution:
- `DROP COLUMN c` followed by inputs that happen to match an older schema
that still contained `c` → `dropUnusedColumns(true)` never fires, the column
stays in the current schema.
- `DROP COLUMN c` → `ADD COLUMN c` with the sink's auto schema evolution
path → `SCHEMA_UPDATE_NEEDED` is never returned, so the sink never issues
`addColumn`, and the column is not re-added to the catalog.
### Root cause
Line numbers at tag `apache-iceberg-1.12.0`, `flink/v2.2`.
1. `TableMetadataCache#schema(TableIdentifier, Schema, boolean)` iterates
`cached.tableSchemas.entrySet()` (= `table.schemas()`, every historical schema
sorted by schema id ascending) and returns the first entry whose
`CompareSchemasVisitor.visit(input, tableSchema, ...)` is `SAME`
(`TableMetadataCache.java:140-184`).
2. `CompareSchemasVisitor` resolves the partner on the table side with
`PartnerIdByNameAccessors`, so matching is name-based; the `field()` method
only checks name/type/optionality and does not require the input's field id to
match the table's (`CompareSchemasVisitor.java`).
3. After `DROP COLUMN c` → `ADD COLUMN c` the current schema (schema id N)
has `c` with a new field id, but schema id N-1 (or any earlier schema that
contained `c` with the retired field id) is still in `table.schemas()`. The
historical schema is iterated first and returned as `SAME`.
4. `ResolvedSchemaInfo` is built from that historical schema, so the writer
serializes the Parquet file using the retired field id. Readers consulting the
current schema look up the new field id, do not find it in the file, and return
`null`.
The per-input-schema cache (`cached.inputSchemas`) also returns the stale
`ResolvedSchemaInfo` directly on subsequent writes of the same input shape,
bypassing the TTL check (`TableMetadataCache.java:141-146`), so a short
`cacheRefreshMs` does not help.
### Reproduction
Minimal DataStream job against an Iceberg table (REST catalog, upsert mode,
equality field `id`):
```java
// schema v0: (id INT PK, val STRING)
// schema v1: v0 + (c STRING) -- addColumn("c", STRING)
// schema v2: deleteColumn("c")
// schema v3: addColumn("c", STRING) -- same name, same type
// before each DML, an external process evolves the catalog as above
sink.collect(dynamicRecord(id=1, val="x", c="bar")); // under schema v3
```
**Expected**: Reader returns `c = "bar"` for `id = 1`.
**Actual**: The Parquet file is written with the retired field id (= `c`
from schema v1). The reader with schema v3 looks up the new field id, does not
find it in the file, and returns `null`.
Confirmed via manifest stats:
```
# Parquet column stats keyed by field_id
value_counts: [{key:1, 1}, {key:2, 1}, {key:3, 1}] # key=3 = retired id
for c
lower_bounds: [{key:1, 1}, {key:2, "x"}, {key:3, "bar"}]
# metadata.json current-schema-id (schema v3)
id=1 name=id required=True
id=2 name=val required=False
id=4 name=c required=False
```
Reader asks for `c = field id 4`, Parquet only has field id 3 → `null`.
### Impact
- Silent data loss on upsert tables when a column with a reused name is
written: the record carries the correct value but readers always see `null` (or
the column's default if one is set).
- `dropUnusedColumns(true)` is effectively inert on tables that have ever
contained a column with the same name as a current column, which breaks the
documented auto-drop behaviour.
- Any flow that mutates the Iceberg catalog outside the sink (external
schema sync via `UpdateSchema`, Debezium-driven DDL replication, etc.) can hit
this because `TableMetadataCache` does not consider external changes to
historical schemas as invalidating its cache.
### Proposed fix
Two options, either or both:
1. In `TableMetadataCache.schema()`, prefer the current schema when multiple
historical schemas match. Simplest form: iterate `cached.tableSchemas` with the
current schema id first (or iterate the entrySet in descending order), so that
the current schema's field ids win over a historical schema that matches by
name.
2. In `CompareSchemasVisitor.field()`, downgrade a name-match whose input
field id does not match the resolved `tableSchemaId` from `SAME` to
`DATA_CONVERSION_NEEDED` (or a dedicated `FIELD_ID_MISMATCH`), so that an input
carrying an up-to-date field id is only resolved against the current schema.
Either direction lets external catalog mutations be reflected the next time
`cacheRefreshMs` elapses, and makes `dropUnusedColumns(true)` fire on same-name
DROP + new-shape inputs.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [x] I cannot contribute a fix for this bug at this time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
