apache / apache/iceberg

Flink: Dynamic Sink — CompareSchemasVisitor never returns SAME for schemas containing empty structs, causing NPE via TableMetadataCache.NOT_FOUND

Open
#17,650 1 comment 0 reactions 0 assignees View on GitHub
bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.