mxm commented on code in PR #18525:
URL: https://github.com/apache/iceberg/pull/18525#discussion_r4346512196
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java:
##########
@@ -94,21 +94,9 @@ public void serialize(DynamicRecordInternal toSerialize,
DataOutputView dataOutp
}
dataOutputView.writeInt(toSerialize.writerKey());
- final RowDataSerializer rowDataSerializer;
- if (writeSchemaAndSpec) {
- rowDataSerializer =
- serializerCache.serializer(
- toSerialize.tableName(), toSerialize.schema(),
toSerialize.spec());
- } else {
- // Check that the schema id can be resolved. Not strictly necessary for
serialization.
- Tuple3<RowDataSerializer, Schema, PartitionSpec> serializer =
- serializerCache.serializerWithSchemaAndSpec(
- toSerialize.tableName(),
- toSerialize.schema().schemaId(),
- toSerialize.spec().specId());
- rowDataSerializer = serializer.f0;
- }
-
+ RowDataSerializer rowDataSerializer =
+ serializerCache.serializer(
+ toSerialize.tableName(), toSerialize.schema(), toSerialize.spec());
Review Comment:
>When i check the TableSerializerCache implementation it is happening even
today.
In the current
[TableSerializerCache](https://github.com/apache/iceberg/blob/main/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java#L93)
implementation, when the schema evolves and encounters a new Schema ID, we do
reload the table inside TableSerializerCache. This leads to new Serializer
instances even for the SchemaId's seen before.
True, although this happens only on schema evolution and not for every call.
A mere refresh of the TableMetadataCache without schema evolution does not
trigger this scenario because we resolve a pre-existing schema by id inside the
serializer cache.
>Wdyt of separating out the path for Serialization, where we have valid
SchemaId and Schema by maintaining the cache keyed by SchemaId and Serializer
created from Schema passed from caller ?
That's a reasonable refactor. IMHO the easiest fix is to cache the
serializers by schema id in the immediate update path where we can trust the
schema id. In the non-immediate update path, we can continue to cache by schema
reference like the current code does. We could achieve this with the existing
code if we converted the cache type to `Map<Object, RowDataSerializer>`. That
way we can pass in schema id or schema. Arguably, it's nicer to separate this
into two implementations, one with schema id, one with Schema reference.
--
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]
