kinolaev commented on code in PR #18507:
URL: https://github.com/apache/iceberg/pull/18507#discussion_r4293634227
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java:
##########
@@ -87,13 +92,32 @@ public DynamicRecordInternal map(DynamicRecordInternal
data) throws Exception {
data.spec(),
tableCreator);
TableMetadataCache.ResolvedSchemaInfo compareInfo = newData.f0;
+ Schema tableSchema = compareInfo.resolvedTableSchema();
- data.setSchema(compareInfo.resolvedTableSchema());
+ data.setEqualityFieldIds(remapFieldIds(data.equalityFields(),
data.schema(), tableSchema));
Review Comment:
As far as I understand, it shouldn't become a hot path: records should be
routed to the updater only on table creation and for roughly one cache refresh
interval after a schema change. The remap itself is a couple of map lookups per
equality field, which is small next to what the operator already does for each
record. Also, since equality fields are set per record and the cache is keyed
by input schema, caching the mapping would need an additional key. So I'd keep
it out of the cache.
However, with `immediateUpdate` disabled, records are currently still routed
to the updater after a schema change, even once the cache refresh interval has
passed. `TableMetadataCache` stores `NOT_FOUND` for a new input schema and
returns it on later lookups without checking the refresh interval, so it never
sees the updated table schema. Here is a test in `TestTableMetadataCache` that
confirms it:
```java
@Test
void refreshesUnresolvedSchemaAfterRefreshInterval() {
Catalog catalog = CATALOG_EXTENSION.catalog();
TableIdentifier tableIdentifier =
TableIdentifier.parse("default.myTable");
Table table = catalog.createTable(tableIdentifier, SCHEMA);
table.newAppend().commit();
AtomicLong nowMillis = new AtomicLong(Instant.now().toEpochMilli());
Clock clock =
new Clock() {
@Override
public ZoneId getZone() {
return ZoneId.systemDefault();
}
@Override
public Clock withZone(ZoneId zone) {
return this;
}
@Override
public Instant instant() {
return Instant.ofEpochMilli(nowMillis.get());
}
};
long refreshMs = 100L;
TableMetadataCache cache =
new TableMetadataCache(catalog, 10, refreshMs, 10, CASE_SENSITIVE,
PRESERVE_COLUMNS, clock);
cache.update(tableIdentifier, table);
assertThat(cache.schema(tableIdentifier, SCHEMA2).compareResult())
.isEqualTo(CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED);
table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
nowMillis.addAndGet(refreshMs + 1);
assertThat(cache.schema(tableIdentifier, SCHEMA2).compareResult())
.isEqualTo(CompareSchemasVisitor.Result.SAME);
}
```
It fails with:
```
TestTableMetadataCache > refreshesUnresolvedSchemaAfterRefreshInterval()
FAILED
expected: SAME
but was: SCHEMA_UPDATE_NEEDED
```
I think it's a bug that should be fixed in a separate PR. WDYT, @mxm?
--
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]
