Flink: Dynamic Sink — CompareSchemasVisitor never returns SAME for schemas containing empty structs, causing NPE via TableMetadataCache.NOT_FOUND
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Apache Iceberg version
1.10.1
### Query engine
Flink
### Please describe the bug 🐞
CompareSchemasVisitor.struct() computes its result as:
Result result = fields.stream().reduce(Result::merge).orElse(Result.SCHEMA_UPDATE_NEEDED);
For a struct with zero fields, the reduce is empty, so orElse(SCHEMA_UPDATE_NEEDED) always fires. Consequently any schema containing an empty struct (e.g. derived from an Avro union whose branch is a zero-field "marker" record — common in registry schemas) can never compare SAME with any table schema — including a table created from that exact schema.
In the Dynamic Sink this is fatal: TableUpdater.findOrCreateSchema sees SCHEMA_UPDATE_NEEDED, applies EvolveSchemaVisitor (a no-op here), commits, re-compares — still SCHEMA_UPDATE_NEEDED — and TableMetadataCache.schema() caches and returns NOT_FOUND, whose resolvedTableSchema() is null. The null propagates to DynamicRecordProcessor.emit, crashing with an unrelated-looking NPE:
Caused by: java.lang.NullPointerException
at org.apache.iceberg.flink.sink.dynamic.DynamicSinkUtil.getEqualityFieldIds(DynamicSinkUtil.java:34)
at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.emit(DynamicRecordProcessor.java:173)
at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.collect(DynamicRecordProcessor.java:118)
The job then crash-loops on every record for that table.
Minimal reproduction
Schema schema = new Schema(
Types.NestedField.required(1, "id", Types.StringType.get()),
Types.NestedField.optional(2, "marker", Types.StructType.of())); // empty struct
InMemoryCatalog catalog = new InMemoryCatalog();
catalog.initialize("mem", Map.of());
catalog.createNamespace(Namespace.of("db"));
Table table = catalog.createTable(TableIdentifier.of("db", "t"), schema);
CompareSchemasVisitor.Result r = CompareSchemasVisitor.visit(schema, table.schema(), true);
// r == SCHEMA_UPDATE_NEEDED — expected SAME (identical schemas)
Expected behavior
An empty struct compared against an empty struct should be SAME. Additionally, TableUpdater.findOrCreateSchema returning a ResolvedSchemaInfo with a null schema after a "successful" evolution deserves a descriptive exception rather than surfacing as an NPE three operators later.
### 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
- [ ] I cannot contribute a fix for this bug at this time
Contributor guide
Research direction
Start with CompareSchemasVisitor.struct() and run the minimal reproduction using an empty StructType to confirm that identical schemas return SCHEMA_UPDATE_NEEDED. Follow the result through TableUpdater.findOrCreateSchema and TableMetadataCache.schema(), then verify that empty structs compare as SAME and that a failed resolved schema reports a descriptive exception instead of reaching DynamicRecordProcessor.emit as an NPE.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- databases
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100