This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 6383a3df85 Flink: Backport: resolving the record serializer without
loading the table (#18525) (#18781)
6383a3df85 is described below
commit 6383a3df85f3ad245b8aa921b35b68436ea6919e
Author: Swapna Marru <[email protected]>
AuthorDate: Wed Nov 4 02:52:07 2026 -0800
Flink: Backport: resolving the record serializer without loading the table
(#18525) (#18781)
---
...namicRecordSerializerDeserializerBenchmark.java | 7 +-
.../flink/sink/dynamic/DynamicIcebergSink.java | 12 +-
.../dynamic/DynamicRecordInternalSerializer.java | 26 ++---
.../sink/dynamic/DynamicRecordInternalType.java | 13 ++-
.../flink/sink/dynamic/DynamicRecordProcessor.java | 6 +-
.../flink/sink/dynamic/TableSerializerCache.java | 95 ++++++++++------
.../DynamicRecordInternalSerializerTestBase.java | 2 +-
.../TestDynamicRecordInternalSerializer.java | 52 ++++++++-
.../sink/dynamic/TestDynamicRecordProcessor.java | 2 +-
.../sink/dynamic/TestTableSerializerCache.java | 122 ++++++++++++++++++---
...namicRecordSerializerDeserializerBenchmark.java | 7 +-
.../flink/sink/dynamic/DynamicIcebergSink.java | 12 +-
.../dynamic/DynamicRecordInternalSerializer.java | 26 ++---
.../sink/dynamic/DynamicRecordInternalType.java | 10 +-
.../flink/sink/dynamic/DynamicRecordProcessor.java | 6 +-
.../flink/sink/dynamic/TableSerializerCache.java | 95 ++++++++++------
.../DynamicRecordInternalSerializerTestBase.java | 2 +-
.../TestDynamicRecordInternalSerializer.java | 52 ++++++++-
.../sink/dynamic/TestDynamicRecordProcessor.java | 2 +-
.../sink/dynamic/TestTableSerializerCache.java | 122 ++++++++++++++++++---
20 files changed, 528 insertions(+), 143 deletions(-)
diff --git
a/flink/v1.20/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
b/flink/v1.20/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
index 6e5d04499e..3fe18a6a0d 100644
---
a/flink/v1.20/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
+++
b/flink/v1.20/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
@@ -110,7 +110,12 @@ public class DynamicRecordSerializerDeserializerBenchmark {
"hadoop",
new Configuration(),
ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
warehouse.getPath()));
- this.type = new DynamicRecordInternalType(catalogLoader, true, 100);
+ this.type =
+ new DynamicRecordInternalType(
+ catalogLoader,
+ true,
+ 100,
+
FlinkDynamicSinkOptions.INPUT_SCHEMAS_PER_TABLE_CACHE_MAX_SIZE.defaultValue());
}
@Benchmark
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index eab3d81c44..42c52894ad 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -522,9 +522,17 @@ public class DynamicIcebergSink
Configuration flinkConfig = fromReadableConfig();
DynamicRecordInternalType type =
- new DynamicRecordInternalType(catalogLoader, false,
flinkDynamicSinkConf.cacheMaxSize());
+ new DynamicRecordInternalType(
+ catalogLoader,
+ false,
+ flinkDynamicSinkConf.cacheMaxSize(),
+ flinkDynamicSinkConf.inputSchemasPerTableCacheMaxSize());
DynamicRecordInternalType sideOutputType =
- new DynamicRecordInternalType(catalogLoader, true,
flinkDynamicSinkConf.cacheMaxSize());
+ new DynamicRecordInternalType(
+ catalogLoader,
+ true,
+ flinkDynamicSinkConf.cacheMaxSize(),
+ flinkDynamicSinkConf.inputSchemasPerTableCacheMaxSize());
SingleOutputStreamOperator<DynamicRecordInternal> converted =
input
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
index e9e7daef46..b7d777946d 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
@@ -65,7 +65,10 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public TypeSerializer<DynamicRecordInternal> duplicate() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(serializerCache.catalogLoader(),
serializerCache.maximumSize()),
+ new TableSerializerCache(
+ serializerCache.catalogLoader(),
+ serializerCache.maximumSize(),
+ serializerCache.inputSchemasPerTableCacheMaximumSize()),
writeSchemaAndSpec,
writeLongUTF);
}
@@ -96,17 +99,10 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
dataOutputView.writeInt(toSerialize.writerKey());
final RowDataSerializer rowDataSerializer;
if (writeSchemaAndSpec) {
- rowDataSerializer =
- serializerCache.serializer(
- toSerialize.tableName(), toSerialize.schema(),
toSerialize.spec());
+ rowDataSerializer = serializerCache.serializer(toSerialize.tableName(),
toSerialize.schema());
} 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 =
+ serializerCache.serializerById(toSerialize.tableName(),
toSerialize.schema());
}
rowDataSerializer.serialize(toSerialize.rowData(), dataOutputView);
@@ -133,7 +129,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
}
spec = PartitionSpecParser.fromJson(schema, dataInputView.readUTF());
- rowDataSerializer = serializerCache.serializer(tableName, schema, spec);
+ rowDataSerializer = serializerCache.serializer(tableName, schema);
} else {
Integer schemaId = dataInputView.readInt();
Integer specId = dataInputView.readInt();
@@ -184,7 +180,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
spec = PartitionSpecParser.fromJson(schema, dataInputView.readUTF());
reuse.setSchema(schema);
reuse.setSpec(spec);
- rowDataSerializer = serializerCache.serializer(tableName, schema, spec);
+ rowDataSerializer = serializerCache.serializer(tableName, schema);
} else {
Integer schemaId = dataInputView.readInt();
Integer specId = dataInputView.readInt();
@@ -216,7 +212,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from) {
RowDataSerializer rowDataSerializer =
- serializerCache.serializer(from.tableName(), from.schema(),
from.spec());
+ serializerCache.serializer(from.tableName(), from.schema());
return new DynamicRecordInternal(
from.tableName(),
from.branch(),
@@ -231,7 +227,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from,
DynamicRecordInternal reuse) {
RowDataSerializer rowDataSerializer =
- serializerCache.serializer(from.tableName(), from.schema(),
from.spec());
+ serializerCache.serializer(from.tableName(), from.schema());
reuse.setTableName(from.tableName());
reuse.setBranch(from.branch());
reuse.setSchema(from.schema());
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
index 6be081aadf..b23db385b1 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
@@ -31,12 +31,17 @@ class DynamicRecordInternalType extends
TypeInformation<DynamicRecordInternal> {
private final CatalogLoader catalogLoader;
private final boolean writeSchemaAndSpec;
private final int cacheSize;
+ private final int inputSchemasPerTableCacheMaximumSize;
DynamicRecordInternalType(
- CatalogLoader catalogLoader, boolean writeSchemaAndSpec, int cacheSize) {
+ CatalogLoader catalogLoader,
+ boolean writeSchemaAndSpec,
+ int cacheSize,
+ int inputSchemasPerTableCacheMaximumSize) {
this.catalogLoader = catalogLoader;
this.writeSchemaAndSpec = writeSchemaAndSpec;
this.cacheSize = cacheSize;
+ this.inputSchemasPerTableCacheMaximumSize =
inputSchemasPerTableCacheMaximumSize;
}
@Override
@@ -72,13 +77,15 @@ class DynamicRecordInternalType extends
TypeInformation<DynamicRecordInternal> {
@Override
public TypeSerializer<DynamicRecordInternal>
createSerializer(SerializerConfig serializerConfig) {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(catalogLoader, cacheSize),
writeSchemaAndSpec);
+ new TableSerializerCache(catalogLoader, cacheSize,
inputSchemasPerTableCacheMaximumSize),
+ writeSchemaAndSpec);
}
@Override
public TypeSerializer<DynamicRecordInternal>
createSerializer(ExecutionConfig executionConfig) {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(catalogLoader, cacheSize),
writeSchemaAndSpec);
+ new TableSerializerCache(catalogLoader, cacheSize,
inputSchemasPerTableCacheMaximumSize),
+ writeSchemaAndSpec);
}
@Override
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
index ba2d816ae9..91f5d52b02 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
@@ -104,12 +104,14 @@ class DynamicRecordProcessor<T> extends
ProcessFunction<T, DynamicRecordInternal
this.forwardStream =
new OutputTag<>(
DYNAMIC_FORWARD_STREAM,
- new DynamicRecordInternalType(catalogLoader, true,
cacheMaximumSize)) {};
+ new DynamicRecordInternalType(
+ catalogLoader, true, cacheMaximumSize,
inputSchemasPerTableCacheMaximumSize)) {};
if (!immediateUpdate) {
updateStream =
new OutputTag<>(
DYNAMIC_TABLE_UPDATE_STREAM,
- new DynamicRecordInternalType(catalogLoader, true,
cacheMaximumSize)) {};
+ new DynamicRecordInternalType(
+ catalogLoader, true, cacheMaximumSize,
inputSchemasPerTableCacheMaximumSize)) {};
}
this.dynamicRecordWithConfig =
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
index 13b4fa56c9..e7b37c391a 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
@@ -20,7 +20,6 @@ package org.apache.iceberg.flink.sink.dynamic;
import java.io.Serializable;
import java.util.Map;
-import javax.annotation.Nullable;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -51,43 +50,41 @@ class TableSerializerCache implements Serializable {
private final CatalogLoader catalogLoader;
private final int maximumSize;
+ private final int inputSchemasPerTableCacheMaximumSize;
// Intentionally not closed; the catalog is reused for the serializer's
lifetime.
private transient Catalog catalog;
private transient Map<String, SerializerInfo> serializers;
- TableSerializerCache(CatalogLoader catalogLoader, int maximumSize) {
+ TableSerializerCache(
+ CatalogLoader catalogLoader, int maximumSize, int
inputSchemasPerTableCacheMaximumSize) {
this.catalogLoader = catalogLoader;
this.maximumSize = maximumSize;
+ this.inputSchemasPerTableCacheMaximumSize =
inputSchemasPerTableCacheMaximumSize;
}
- RowDataSerializer serializer(String tableName, Schema schema, PartitionSpec
spec) {
- return serializer(tableName, schema, spec, null, null).f0;
+ RowDataSerializer serializer(String tableName, Schema schema) {
+ return serializerInfo(tableName).serializerBySchema(schema);
}
- Tuple3<RowDataSerializer, Schema, PartitionSpec> serializerWithSchemaAndSpec(
- String tableName, Integer schemaId, Integer specId) {
- return serializer(tableName, null, null, schemaId, specId);
+ /**
+ * Resolves the serializer when the caller already holds the schema and spec
along with their ids.
+ * The serializer is looked up by schema id, so it is not rebuilt when the
{@link Schema} instance
+ * changes.
+ *
+ * <p>The id-keyed serializers are shared with {@link
#serializerWithSchemaAndSpec}, so the schema
+ * must come from the table. Do not use this for user-supplied schemas,
whose ids are not reliable
+ * and could collide with the table's schemas.
+ */
+ RowDataSerializer serializerById(String tableName, Schema schema) {
+ return serializerInfo(tableName).serializerBySchemaId(schema);
}
- private Tuple3<RowDataSerializer, Schema, PartitionSpec> serializer(
- String tableName,
- @Nullable Schema unknownSchema,
- @Nullable PartitionSpec unknownSpec,
- @Nullable Integer schemaId,
- @Nullable Integer specId) {
- Preconditions.checkState(
- (unknownSchema == null && unknownSpec == null) ^ (schemaId == null &&
specId == null),
- "Either the full schema/spec or their ids must be provided.");
-
- if (serializers == null) {
- // We need to initialize the cache at the first time
- this.serializers = new LRUCache<>(maximumSize);
- }
-
- SerializerInfo info = serializers.computeIfAbsent(tableName,
SerializerInfo::new);
- Schema schema = unknownSchema != null ? unknownSchema :
info.schemas.get(schemaId);
- PartitionSpec spec = unknownSpec != null ? unknownSpec :
info.specs.get(specId);
+ Tuple3<RowDataSerializer, Schema, PartitionSpec> serializerWithSchemaAndSpec(
+ String tableName, Integer schemaId, Integer specId) {
+ SerializerInfo info = serializerInfo(tableName);
+ Schema schema = info.schemas.get(schemaId);
+ PartitionSpec spec = info.specs.get(specId);
if (schema == null || spec == null) {
info.update();
@@ -95,11 +92,28 @@ class TableSerializerCache implements Serializable {
spec = info.specs.get(specId);
}
- RowDataSerializer serializer =
- info.serializers.computeIfAbsent(
- schema, s -> new RowDataSerializer(FlinkSchemaUtil.convert(s)));
+ Preconditions.checkArgument(
+ schema != null,
+ "SchemaId:%s not found in table %s. Expected SchemaId:%s to exist.",
+ schemaId,
+ tableName,
+ schemaId);
+ Preconditions.checkArgument(
+ spec != null,
+ "SpecId:%s not found in table %s. Expected SpecId:%s to exist.",
+ specId,
+ tableName,
+ specId);
+
+ return Tuple3.of(info.serializerBySchemaId(schema), schema, spec);
+ }
+
+ private SerializerInfo serializerInfo(String tableName) {
+ if (serializers == null) {
+ this.serializers = new LRUCache<>(maximumSize);
+ }
- return Tuple3.of(serializer, schema, spec);
+ return serializers.computeIfAbsent(tableName, SerializerInfo::new);
}
CatalogLoader catalogLoader() {
@@ -110,19 +124,38 @@ class TableSerializerCache implements Serializable {
return maximumSize;
}
+ int inputSchemasPerTableCacheMaximumSize() {
+ return inputSchemasPerTableCacheMaximumSize;
+ }
+
private class SerializerInfo {
private final String tableName;
- private final Map<Schema, RowDataSerializer> serializers;
+ private final Map<Schema, RowDataSerializer> serializersBySchema;
+ private final Map<Integer, RowDataSerializer> serializersBySchemaId;
private Map<Integer, Schema> schemas;
private Map<Integer, PartitionSpec> specs;
SerializerInfo(String tableName) {
this.tableName = tableName;
- this.serializers = Maps.newHashMapWithExpectedSize(2);
+ this.serializersBySchema = new
LRUCache<>(inputSchemasPerTableCacheMaximumSize);
+ this.serializersBySchemaId = Maps.newHashMapWithExpectedSize(2);
this.schemas = Maps.newHashMapWithExpectedSize(1);
this.specs = Maps.newHashMapWithExpectedSize(0);
}
+ private RowDataSerializer serializerBySchema(Schema schema) {
+ return serializersBySchema.computeIfAbsent(schema,
this::createSerializer);
+ }
+
+ private RowDataSerializer serializerBySchemaId(Schema schema) {
+ return serializersBySchemaId.computeIfAbsent(
+ schema.schemaId(), id -> createSerializer(schema));
+ }
+
+ private RowDataSerializer createSerializer(Schema schema) {
+ return new RowDataSerializer(FlinkSchemaUtil.convert(schema));
+ }
+
private void update() {
if (catalog == null) {
catalog = catalogLoader.loadCatalog();
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
index 07096b891a..07a2ceca99 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
@@ -65,7 +65,7 @@ abstract class DynamicRecordInternalSerializerTestBase
@Override
protected TypeSerializer<DynamicRecordInternal> createSerializer() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1),
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1, 10),
writeFullSchemaAndSpec,
writeLongUTF);
}
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
index 5667583599..d24076c382 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
@@ -20,15 +20,18 @@ package org.apache.iceberg.flink.sink.dynamic;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import java.io.IOException;
import java.util.Collections;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.types.RowKind;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -107,6 +110,53 @@ public class TestDynamicRecordInternalSerializer {
.isTrue();
}
+ @Test
+ void testSerializeWithSchemaIdForMissingTable() {
+ DynamicRecordInternalSerializer serializer =
+ new DynamicRecordInternalSerializer(
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1,
10), false);
+ GenericRowData row = GenericRowData.of(123L,
StringData.fromString("test"), 1.23f);
+ DynamicRecordInternal record =
+ new DynamicRecordInternal(
+ "missingTable",
+ DynamicRecordInternalSerializerTestBase.BRANCH,
+ DynamicRecordInternalSerializerTestBase.SCHEMA,
+ row,
+ DynamicRecordInternalSerializerTestBase.SPEC,
+ 42,
+ false,
+ Collections.emptySet());
+ DataOutputSerializer output = new DataOutputSerializer(128);
+
+ assertThatCode(() -> serializer.serialize(record,
output)).doesNotThrowAnyException();
+ assertThat(output.length()).isGreaterThan(0);
+ }
+
+ @Test
+ void testCacheSizesFromType() {
+ DynamicRecordInternalType type =
+ new DynamicRecordInternalType(CATALOG_EXTENSION.catalogLoader(), true,
5, 3);
+ TypeSerializer<DynamicRecordInternal> internalTypeSerializer =
+ type.createSerializer(new SerializerConfigImpl());
+
assertThat(internalTypeSerializer).isInstanceOf(DynamicRecordInternalSerializer.class);
+ DynamicRecordInternalSerializer dynamicRecordInternalSerializer =
+ (DynamicRecordInternalSerializer) internalTypeSerializer;
+
+
assertThat(dynamicRecordInternalSerializer.getSerializerCache().maximumSize()).isEqualTo(5);
+ assertThat(
+ dynamicRecordInternalSerializer
+ .getSerializerCache()
+ .inputSchemasPerTableCacheMaximumSize())
+ .isEqualTo(3);
+
+ DynamicRecordInternalSerializer duplicate =
+ (DynamicRecordInternalSerializer)
dynamicRecordInternalSerializer.duplicate();
+ assertThat(duplicate.getSerializerCache())
+ .isNotSameAs(dynamicRecordInternalSerializer.getSerializerCache());
+ assertThat(duplicate.getSerializerCache().maximumSize()).isEqualTo(5);
+
assertThat(duplicate.getSerializerCache().inputSchemasPerTableCacheMaximumSize()).isEqualTo(3);
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void copyPreservesValues(boolean reuse) {
@@ -167,7 +217,7 @@ public class TestDynamicRecordInternalSerializer {
private DynamicRecordInternalSerializer createSerializer() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1), true);
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1, 10),
true);
}
private static class OldTypeSerializerSnapshot
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
index 48b94c14c7..6f50e62f2c 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
@@ -179,7 +179,7 @@ class TestDynamicRecordProcessor {
return new OutputTag<>(
DynamicRecordProcessor.DYNAMIC_TABLE_UPDATE_STREAM,
new DynamicRecordInternalType(
- CATALOG_EXTENSION.catalogLoader(), true, CACHE_REFRESH_MS)) {};
+ CATALOG_EXTENSION.catalogLoader(), true, CACHE_REFRESH_MS, 10)) {};
}
private static DynamicRecord recordWithDistributionMode(
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
index bbfb39618f..156ec0def0 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
@@ -24,6 +24,7 @@ import static
org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.apache.iceberg.types.Types.StringType;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.function.Supplier;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -52,16 +53,13 @@ class TestTableSerializerCache {
optional(2, "data", StringType.get()),
optional(3, "double", DoubleType.get()));
- TableSerializerCache cache = new
TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10);
+ TableSerializerCache cache = new
TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10, 10);
@Test
void testFullSchemaCaching() {
- Supplier<RowDataSerializer> creator1a =
- () -> cache.serializer("table", schema1,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator1b =
- () -> cache.serializer("table", schema2,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator2 =
- () -> cache.serializer("table2", schema2,
PartitionSpec.unpartitioned());
+ Supplier<RowDataSerializer> creator1a = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator1b = () -> cache.serializer("table",
schema2);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table2",
schema2);
RowDataSerializer serializer1a = creator1a.get();
RowDataSerializer serializer1b = creator1b.get();
@@ -73,10 +71,32 @@ class TestTableSerializerCache {
assertThat(serializer2).isSameAs(creator2.get());
}
+ @Test
+ void testSerializerById() {
+ LoadCountingCatalogLoader catalogLoader =
+ new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader());
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
+
+ Schema schema1Copy = new Schema(schema1.schemaId(), required(1, "id",
LongType.get()));
+ Supplier<RowDataSerializer> creator1a = () ->
cache.serializerById("table", schema1);
+ Supplier<RowDataSerializer> creator1b = () ->
cache.serializerById("table", schema1Copy);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializerById("table",
schema2);
+ Supplier<RowDataSerializer> creator3 = () ->
cache.serializerById("table2", schema1);
+
+ RowDataSerializer serializer1a = creator1a.get();
+ RowDataSerializer serializer1b = creator1b.get();
+ RowDataSerializer serializer2 = creator2.get();
+ RowDataSerializer serializer3 = creator3.get();
+
+ assertThat(serializer1a).isNotSameAs(serializer2).isNotSameAs(serializer3);
+ assertThat(serializer1b).isSameAs(serializer1a);
+ assertThat(catalogLoader.loadCount()).isEqualTo(0);
+ }
+
@Test
void testCachingWithSchemaLookup() {
CatalogLoader catalogLoader = CATALOG_EXTENSION.catalogLoader();
- cache = new TableSerializerCache(catalogLoader, 10);
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
Catalog catalog = catalogLoader.loadCatalog();
Table table = catalog.createTable(TableIdentifier.of("table"), schema1);
@@ -98,15 +118,87 @@ class TestTableSerializerCache {
assertThat(serializerWithSchemaAndSpec.f2).isSameAs(serializerWithSchemaAndSpec2.f2);
}
+ @Test
+ void testSchemaLookupWithSchemaEvolution() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int specId = table.spec().specId();
+ int schemaId1 = table.schema().schemaId();
+ Supplier<RowDataSerializer> creator1a =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId1, specId).f0;
+
+ table.updateSchema().addColumn("data", StringType.get()).commit();
+
+ int schemaId2 = table.schema().schemaId();
+ Supplier<RowDataSerializer> creator2 =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId2, specId).f0;
+ Supplier<RowDataSerializer> creator1b =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId1, specId).f0;
+
+ assertThat(creator2.get()).isNotSameAs(creator1a.get());
+ assertThat(creator1b.get()).isSameAs(creator1a.get());
+ }
+
+ @Test
+ void testSchemaSerializerEviction() {
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10, 1);
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table",
schema2);
+
+ RowDataSerializer serializer1 = creator1.get();
+ assertThat(creator1.get()).isSameAs(serializer1);
+
+ // Only one schema fits, so caching schema2 evicts the serializer for
schema1
+ RowDataSerializer serializer2 = creator2.get();
+ assertThat(creator2.get()).isSameAs(serializer2);
+ assertThat(creator1.get()).isNotSameAs(serializer1);
+ }
+
+ @Test
+ void testSerializerByIdSharedWithSchemaLookup() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int schemaId = table.schema().schemaId();
+ int specId = table.spec().specId();
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializerById("table",
table.schema());
+ Supplier<RowDataSerializer> creator2 =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId, specId).f0;
+
+ RowDataSerializer serializer1 = creator1.get();
+
+ assertThat(creator2.get()).isSameAs(serializer1);
+ assertThat(creator1.get()).isSameAs(serializer1);
+ }
+
+ @Test
+ void testSchemaLookupWithUnknownSchemaId() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int specId = table.spec().specId();
+
+ assertThatThrownBy(() -> cache.serializerWithSchemaAndSpec("table", 999,
specId))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("SchemaId:999 not found in table table. Expected
SchemaId:999 to exist.");
+ }
+
+ @Test
+ void testSchemaLookupWithUnknownSpecId() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int schemaId = table.schema().schemaId();
+
+ assertThatThrownBy(() -> cache.serializerWithSchemaAndSpec("table",
schemaId, 999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("SpecId:999 not found in table table. Expected SpecId:999
to exist.");
+ }
+
@Test
void testCacheEviction() {
- cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 0);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 0, 10);
assertThat(cache.maximumSize()).isEqualTo(0);
- Supplier<RowDataSerializer> creator1 =
- () -> cache.serializer("table", schema1,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator2 =
- () -> cache.serializer("table2", schema2,
PartitionSpec.unpartitioned());
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table2",
schema2);
RowDataSerializer serializer1 = creator1.get();
RowDataSerializer serializer2 = creator2.get();
@@ -118,7 +210,7 @@ class TestTableSerializerCache {
@Test
void testCacheSize() {
- cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000,
10);
assertThat(cache.maximumSize()).isEqualTo(1000);
}
@@ -129,7 +221,7 @@ class TestTableSerializerCache {
LoadCountingCatalogLoader catalogLoader =
new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader());
- cache = new TableSerializerCache(catalogLoader, 10);
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
// schema/spec ids are unknown, so both lookups miss the cache and need
the catalog
cache.serializerWithSchemaAndSpec(
diff --git
a/flink/v2.2/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
b/flink/v2.2/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
index 6e5d04499e..3fe18a6a0d 100644
---
a/flink/v2.2/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
+++
b/flink/v2.2/flink/src/jmh/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordSerializerDeserializerBenchmark.java
@@ -110,7 +110,12 @@ public class DynamicRecordSerializerDeserializerBenchmark {
"hadoop",
new Configuration(),
ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
warehouse.getPath()));
- this.type = new DynamicRecordInternalType(catalogLoader, true, 100);
+ this.type =
+ new DynamicRecordInternalType(
+ catalogLoader,
+ true,
+ 100,
+
FlinkDynamicSinkOptions.INPUT_SCHEMAS_PER_TABLE_CACHE_MAX_SIZE.defaultValue());
}
@Benchmark
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index c8f9fece23..3d478ecd6f 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -520,9 +520,17 @@ public class DynamicIcebergSink
Configuration flinkConfig = fromReadableConfig();
DynamicRecordInternalType type =
- new DynamicRecordInternalType(catalogLoader, false,
flinkDynamicSinkConf.cacheMaxSize());
+ new DynamicRecordInternalType(
+ catalogLoader,
+ false,
+ flinkDynamicSinkConf.cacheMaxSize(),
+ flinkDynamicSinkConf.inputSchemasPerTableCacheMaxSize());
DynamicRecordInternalType sideOutputType =
- new DynamicRecordInternalType(catalogLoader, true,
flinkDynamicSinkConf.cacheMaxSize());
+ new DynamicRecordInternalType(
+ catalogLoader,
+ true,
+ flinkDynamicSinkConf.cacheMaxSize(),
+ flinkDynamicSinkConf.inputSchemasPerTableCacheMaxSize());
SingleOutputStreamOperator<DynamicRecordInternal> converted =
input
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
index e9e7daef46..b7d777946d 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializer.java
@@ -65,7 +65,10 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public TypeSerializer<DynamicRecordInternal> duplicate() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(serializerCache.catalogLoader(),
serializerCache.maximumSize()),
+ new TableSerializerCache(
+ serializerCache.catalogLoader(),
+ serializerCache.maximumSize(),
+ serializerCache.inputSchemasPerTableCacheMaximumSize()),
writeSchemaAndSpec,
writeLongUTF);
}
@@ -96,17 +99,10 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
dataOutputView.writeInt(toSerialize.writerKey());
final RowDataSerializer rowDataSerializer;
if (writeSchemaAndSpec) {
- rowDataSerializer =
- serializerCache.serializer(
- toSerialize.tableName(), toSerialize.schema(),
toSerialize.spec());
+ rowDataSerializer = serializerCache.serializer(toSerialize.tableName(),
toSerialize.schema());
} 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 =
+ serializerCache.serializerById(toSerialize.tableName(),
toSerialize.schema());
}
rowDataSerializer.serialize(toSerialize.rowData(), dataOutputView);
@@ -133,7 +129,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
}
spec = PartitionSpecParser.fromJson(schema, dataInputView.readUTF());
- rowDataSerializer = serializerCache.serializer(tableName, schema, spec);
+ rowDataSerializer = serializerCache.serializer(tableName, schema);
} else {
Integer schemaId = dataInputView.readInt();
Integer specId = dataInputView.readInt();
@@ -184,7 +180,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
spec = PartitionSpecParser.fromJson(schema, dataInputView.readUTF());
reuse.setSchema(schema);
reuse.setSpec(spec);
- rowDataSerializer = serializerCache.serializer(tableName, schema, spec);
+ rowDataSerializer = serializerCache.serializer(tableName, schema);
} else {
Integer schemaId = dataInputView.readInt();
Integer specId = dataInputView.readInt();
@@ -216,7 +212,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from) {
RowDataSerializer rowDataSerializer =
- serializerCache.serializer(from.tableName(), from.schema(),
from.spec());
+ serializerCache.serializer(from.tableName(), from.schema());
return new DynamicRecordInternal(
from.tableName(),
from.branch(),
@@ -231,7 +227,7 @@ class DynamicRecordInternalSerializer extends
TypeSerializer<DynamicRecordIntern
@Override
public DynamicRecordInternal copy(DynamicRecordInternal from,
DynamicRecordInternal reuse) {
RowDataSerializer rowDataSerializer =
- serializerCache.serializer(from.tableName(), from.schema(),
from.spec());
+ serializerCache.serializer(from.tableName(), from.schema());
reuse.setTableName(from.tableName());
reuse.setBranch(from.branch());
reuse.setSchema(from.schema());
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
index c18c8f670d..ef5a270f98 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalType.java
@@ -30,12 +30,17 @@ class DynamicRecordInternalType extends
TypeInformation<DynamicRecordInternal> {
private final CatalogLoader catalogLoader;
private final boolean writeSchemaAndSpec;
private final int cacheSize;
+ private final int inputSchemasPerTableCacheMaximumSize;
DynamicRecordInternalType(
- CatalogLoader catalogLoader, boolean writeSchemaAndSpec, int cacheSize) {
+ CatalogLoader catalogLoader,
+ boolean writeSchemaAndSpec,
+ int cacheSize,
+ int inputSchemasPerTableCacheMaximumSize) {
this.catalogLoader = catalogLoader;
this.writeSchemaAndSpec = writeSchemaAndSpec;
this.cacheSize = cacheSize;
+ this.inputSchemasPerTableCacheMaximumSize =
inputSchemasPerTableCacheMaximumSize;
}
@Override
@@ -71,7 +76,8 @@ class DynamicRecordInternalType extends
TypeInformation<DynamicRecordInternal> {
@Override
public TypeSerializer<DynamicRecordInternal>
createSerializer(SerializerConfig serializerConfig) {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(catalogLoader, cacheSize),
writeSchemaAndSpec);
+ new TableSerializerCache(catalogLoader, cacheSize,
inputSchemasPerTableCacheMaximumSize),
+ writeSchemaAndSpec);
}
@Override
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
index ba2d816ae9..91f5d52b02 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
@@ -104,12 +104,14 @@ class DynamicRecordProcessor<T> extends
ProcessFunction<T, DynamicRecordInternal
this.forwardStream =
new OutputTag<>(
DYNAMIC_FORWARD_STREAM,
- new DynamicRecordInternalType(catalogLoader, true,
cacheMaximumSize)) {};
+ new DynamicRecordInternalType(
+ catalogLoader, true, cacheMaximumSize,
inputSchemasPerTableCacheMaximumSize)) {};
if (!immediateUpdate) {
updateStream =
new OutputTag<>(
DYNAMIC_TABLE_UPDATE_STREAM,
- new DynamicRecordInternalType(catalogLoader, true,
cacheMaximumSize)) {};
+ new DynamicRecordInternalType(
+ catalogLoader, true, cacheMaximumSize,
inputSchemasPerTableCacheMaximumSize)) {};
}
this.dynamicRecordWithConfig =
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
index 13b4fa56c9..e7b37c391a 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
@@ -20,7 +20,6 @@ package org.apache.iceberg.flink.sink.dynamic;
import java.io.Serializable;
import java.util.Map;
-import javax.annotation.Nullable;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -51,43 +50,41 @@ class TableSerializerCache implements Serializable {
private final CatalogLoader catalogLoader;
private final int maximumSize;
+ private final int inputSchemasPerTableCacheMaximumSize;
// Intentionally not closed; the catalog is reused for the serializer's
lifetime.
private transient Catalog catalog;
private transient Map<String, SerializerInfo> serializers;
- TableSerializerCache(CatalogLoader catalogLoader, int maximumSize) {
+ TableSerializerCache(
+ CatalogLoader catalogLoader, int maximumSize, int
inputSchemasPerTableCacheMaximumSize) {
this.catalogLoader = catalogLoader;
this.maximumSize = maximumSize;
+ this.inputSchemasPerTableCacheMaximumSize =
inputSchemasPerTableCacheMaximumSize;
}
- RowDataSerializer serializer(String tableName, Schema schema, PartitionSpec
spec) {
- return serializer(tableName, schema, spec, null, null).f0;
+ RowDataSerializer serializer(String tableName, Schema schema) {
+ return serializerInfo(tableName).serializerBySchema(schema);
}
- Tuple3<RowDataSerializer, Schema, PartitionSpec> serializerWithSchemaAndSpec(
- String tableName, Integer schemaId, Integer specId) {
- return serializer(tableName, null, null, schemaId, specId);
+ /**
+ * Resolves the serializer when the caller already holds the schema and spec
along with their ids.
+ * The serializer is looked up by schema id, so it is not rebuilt when the
{@link Schema} instance
+ * changes.
+ *
+ * <p>The id-keyed serializers are shared with {@link
#serializerWithSchemaAndSpec}, so the schema
+ * must come from the table. Do not use this for user-supplied schemas,
whose ids are not reliable
+ * and could collide with the table's schemas.
+ */
+ RowDataSerializer serializerById(String tableName, Schema schema) {
+ return serializerInfo(tableName).serializerBySchemaId(schema);
}
- private Tuple3<RowDataSerializer, Schema, PartitionSpec> serializer(
- String tableName,
- @Nullable Schema unknownSchema,
- @Nullable PartitionSpec unknownSpec,
- @Nullable Integer schemaId,
- @Nullable Integer specId) {
- Preconditions.checkState(
- (unknownSchema == null && unknownSpec == null) ^ (schemaId == null &&
specId == null),
- "Either the full schema/spec or their ids must be provided.");
-
- if (serializers == null) {
- // We need to initialize the cache at the first time
- this.serializers = new LRUCache<>(maximumSize);
- }
-
- SerializerInfo info = serializers.computeIfAbsent(tableName,
SerializerInfo::new);
- Schema schema = unknownSchema != null ? unknownSchema :
info.schemas.get(schemaId);
- PartitionSpec spec = unknownSpec != null ? unknownSpec :
info.specs.get(specId);
+ Tuple3<RowDataSerializer, Schema, PartitionSpec> serializerWithSchemaAndSpec(
+ String tableName, Integer schemaId, Integer specId) {
+ SerializerInfo info = serializerInfo(tableName);
+ Schema schema = info.schemas.get(schemaId);
+ PartitionSpec spec = info.specs.get(specId);
if (schema == null || spec == null) {
info.update();
@@ -95,11 +92,28 @@ class TableSerializerCache implements Serializable {
spec = info.specs.get(specId);
}
- RowDataSerializer serializer =
- info.serializers.computeIfAbsent(
- schema, s -> new RowDataSerializer(FlinkSchemaUtil.convert(s)));
+ Preconditions.checkArgument(
+ schema != null,
+ "SchemaId:%s not found in table %s. Expected SchemaId:%s to exist.",
+ schemaId,
+ tableName,
+ schemaId);
+ Preconditions.checkArgument(
+ spec != null,
+ "SpecId:%s not found in table %s. Expected SpecId:%s to exist.",
+ specId,
+ tableName,
+ specId);
+
+ return Tuple3.of(info.serializerBySchemaId(schema), schema, spec);
+ }
+
+ private SerializerInfo serializerInfo(String tableName) {
+ if (serializers == null) {
+ this.serializers = new LRUCache<>(maximumSize);
+ }
- return Tuple3.of(serializer, schema, spec);
+ return serializers.computeIfAbsent(tableName, SerializerInfo::new);
}
CatalogLoader catalogLoader() {
@@ -110,19 +124,38 @@ class TableSerializerCache implements Serializable {
return maximumSize;
}
+ int inputSchemasPerTableCacheMaximumSize() {
+ return inputSchemasPerTableCacheMaximumSize;
+ }
+
private class SerializerInfo {
private final String tableName;
- private final Map<Schema, RowDataSerializer> serializers;
+ private final Map<Schema, RowDataSerializer> serializersBySchema;
+ private final Map<Integer, RowDataSerializer> serializersBySchemaId;
private Map<Integer, Schema> schemas;
private Map<Integer, PartitionSpec> specs;
SerializerInfo(String tableName) {
this.tableName = tableName;
- this.serializers = Maps.newHashMapWithExpectedSize(2);
+ this.serializersBySchema = new
LRUCache<>(inputSchemasPerTableCacheMaximumSize);
+ this.serializersBySchemaId = Maps.newHashMapWithExpectedSize(2);
this.schemas = Maps.newHashMapWithExpectedSize(1);
this.specs = Maps.newHashMapWithExpectedSize(0);
}
+ private RowDataSerializer serializerBySchema(Schema schema) {
+ return serializersBySchema.computeIfAbsent(schema,
this::createSerializer);
+ }
+
+ private RowDataSerializer serializerBySchemaId(Schema schema) {
+ return serializersBySchemaId.computeIfAbsent(
+ schema.schemaId(), id -> createSerializer(schema));
+ }
+
+ private RowDataSerializer createSerializer(Schema schema) {
+ return new RowDataSerializer(FlinkSchemaUtil.convert(schema));
+ }
+
private void update() {
if (catalog == null) {
catalog = catalogLoader.loadCatalog();
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
index 07096b891a..07a2ceca99 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordInternalSerializerTestBase.java
@@ -65,7 +65,7 @@ abstract class DynamicRecordInternalSerializerTestBase
@Override
protected TypeSerializer<DynamicRecordInternal> createSerializer() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1),
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1, 10),
writeFullSchemaAndSpec,
writeLongUTF);
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
index 5667583599..d24076c382 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordInternalSerializer.java
@@ -20,15 +20,18 @@ package org.apache.iceberg.flink.sink.dynamic;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import java.io.IOException;
import java.util.Collections;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.types.RowKind;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -107,6 +110,53 @@ public class TestDynamicRecordInternalSerializer {
.isTrue();
}
+ @Test
+ void testSerializeWithSchemaIdForMissingTable() {
+ DynamicRecordInternalSerializer serializer =
+ new DynamicRecordInternalSerializer(
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1,
10), false);
+ GenericRowData row = GenericRowData.of(123L,
StringData.fromString("test"), 1.23f);
+ DynamicRecordInternal record =
+ new DynamicRecordInternal(
+ "missingTable",
+ DynamicRecordInternalSerializerTestBase.BRANCH,
+ DynamicRecordInternalSerializerTestBase.SCHEMA,
+ row,
+ DynamicRecordInternalSerializerTestBase.SPEC,
+ 42,
+ false,
+ Collections.emptySet());
+ DataOutputSerializer output = new DataOutputSerializer(128);
+
+ assertThatCode(() -> serializer.serialize(record,
output)).doesNotThrowAnyException();
+ assertThat(output.length()).isGreaterThan(0);
+ }
+
+ @Test
+ void testCacheSizesFromType() {
+ DynamicRecordInternalType type =
+ new DynamicRecordInternalType(CATALOG_EXTENSION.catalogLoader(), true,
5, 3);
+ TypeSerializer<DynamicRecordInternal> internalTypeSerializer =
+ type.createSerializer(new SerializerConfigImpl());
+
assertThat(internalTypeSerializer).isInstanceOf(DynamicRecordInternalSerializer.class);
+ DynamicRecordInternalSerializer dynamicRecordInternalSerializer =
+ (DynamicRecordInternalSerializer) internalTypeSerializer;
+
+
assertThat(dynamicRecordInternalSerializer.getSerializerCache().maximumSize()).isEqualTo(5);
+ assertThat(
+ dynamicRecordInternalSerializer
+ .getSerializerCache()
+ .inputSchemasPerTableCacheMaximumSize())
+ .isEqualTo(3);
+
+ DynamicRecordInternalSerializer duplicate =
+ (DynamicRecordInternalSerializer)
dynamicRecordInternalSerializer.duplicate();
+ assertThat(duplicate.getSerializerCache())
+ .isNotSameAs(dynamicRecordInternalSerializer.getSerializerCache());
+ assertThat(duplicate.getSerializerCache().maximumSize()).isEqualTo(5);
+
assertThat(duplicate.getSerializerCache().inputSchemasPerTableCacheMaximumSize()).isEqualTo(3);
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void copyPreservesValues(boolean reuse) {
@@ -167,7 +217,7 @@ public class TestDynamicRecordInternalSerializer {
private DynamicRecordInternalSerializer createSerializer() {
return new DynamicRecordInternalSerializer(
- new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1), true);
+ new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1, 10),
true);
}
private static class OldTypeSerializerSnapshot
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
index 48b94c14c7..6f50e62f2c 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
@@ -179,7 +179,7 @@ class TestDynamicRecordProcessor {
return new OutputTag<>(
DynamicRecordProcessor.DYNAMIC_TABLE_UPDATE_STREAM,
new DynamicRecordInternalType(
- CATALOG_EXTENSION.catalogLoader(), true, CACHE_REFRESH_MS)) {};
+ CATALOG_EXTENSION.catalogLoader(), true, CACHE_REFRESH_MS, 10)) {};
}
private static DynamicRecord recordWithDistributionMode(
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
index bbfb39618f..156ec0def0 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
@@ -24,6 +24,7 @@ import static
org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.apache.iceberg.types.Types.StringType;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.function.Supplier;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -52,16 +53,13 @@ class TestTableSerializerCache {
optional(2, "data", StringType.get()),
optional(3, "double", DoubleType.get()));
- TableSerializerCache cache = new
TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10);
+ TableSerializerCache cache = new
TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10, 10);
@Test
void testFullSchemaCaching() {
- Supplier<RowDataSerializer> creator1a =
- () -> cache.serializer("table", schema1,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator1b =
- () -> cache.serializer("table", schema2,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator2 =
- () -> cache.serializer("table2", schema2,
PartitionSpec.unpartitioned());
+ Supplier<RowDataSerializer> creator1a = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator1b = () -> cache.serializer("table",
schema2);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table2",
schema2);
RowDataSerializer serializer1a = creator1a.get();
RowDataSerializer serializer1b = creator1b.get();
@@ -73,10 +71,32 @@ class TestTableSerializerCache {
assertThat(serializer2).isSameAs(creator2.get());
}
+ @Test
+ void testSerializerById() {
+ LoadCountingCatalogLoader catalogLoader =
+ new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader());
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
+
+ Schema schema1Copy = new Schema(schema1.schemaId(), required(1, "id",
LongType.get()));
+ Supplier<RowDataSerializer> creator1a = () ->
cache.serializerById("table", schema1);
+ Supplier<RowDataSerializer> creator1b = () ->
cache.serializerById("table", schema1Copy);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializerById("table",
schema2);
+ Supplier<RowDataSerializer> creator3 = () ->
cache.serializerById("table2", schema1);
+
+ RowDataSerializer serializer1a = creator1a.get();
+ RowDataSerializer serializer1b = creator1b.get();
+ RowDataSerializer serializer2 = creator2.get();
+ RowDataSerializer serializer3 = creator3.get();
+
+ assertThat(serializer1a).isNotSameAs(serializer2).isNotSameAs(serializer3);
+ assertThat(serializer1b).isSameAs(serializer1a);
+ assertThat(catalogLoader.loadCount()).isEqualTo(0);
+ }
+
@Test
void testCachingWithSchemaLookup() {
CatalogLoader catalogLoader = CATALOG_EXTENSION.catalogLoader();
- cache = new TableSerializerCache(catalogLoader, 10);
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
Catalog catalog = catalogLoader.loadCatalog();
Table table = catalog.createTable(TableIdentifier.of("table"), schema1);
@@ -98,15 +118,87 @@ class TestTableSerializerCache {
assertThat(serializerWithSchemaAndSpec.f2).isSameAs(serializerWithSchemaAndSpec2.f2);
}
+ @Test
+ void testSchemaLookupWithSchemaEvolution() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int specId = table.spec().specId();
+ int schemaId1 = table.schema().schemaId();
+ Supplier<RowDataSerializer> creator1a =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId1, specId).f0;
+
+ table.updateSchema().addColumn("data", StringType.get()).commit();
+
+ int schemaId2 = table.schema().schemaId();
+ Supplier<RowDataSerializer> creator2 =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId2, specId).f0;
+ Supplier<RowDataSerializer> creator1b =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId1, specId).f0;
+
+ assertThat(creator2.get()).isNotSameAs(creator1a.get());
+ assertThat(creator1b.get()).isSameAs(creator1a.get());
+ }
+
+ @Test
+ void testSchemaSerializerEviction() {
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10, 1);
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table",
schema2);
+
+ RowDataSerializer serializer1 = creator1.get();
+ assertThat(creator1.get()).isSameAs(serializer1);
+
+ // Only one schema fits, so caching schema2 evicts the serializer for
schema1
+ RowDataSerializer serializer2 = creator2.get();
+ assertThat(creator2.get()).isSameAs(serializer2);
+ assertThat(creator1.get()).isNotSameAs(serializer1);
+ }
+
+ @Test
+ void testSerializerByIdSharedWithSchemaLookup() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int schemaId = table.schema().schemaId();
+ int specId = table.spec().specId();
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializerById("table",
table.schema());
+ Supplier<RowDataSerializer> creator2 =
+ () -> cache.serializerWithSchemaAndSpec("table", schemaId, specId).f0;
+
+ RowDataSerializer serializer1 = creator1.get();
+
+ assertThat(creator2.get()).isSameAs(serializer1);
+ assertThat(creator1.get()).isSameAs(serializer1);
+ }
+
+ @Test
+ void testSchemaLookupWithUnknownSchemaId() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int specId = table.spec().specId();
+
+ assertThatThrownBy(() -> cache.serializerWithSchemaAndSpec("table", 999,
specId))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("SchemaId:999 not found in table table. Expected
SchemaId:999 to exist.");
+ }
+
+ @Test
+ void testSchemaLookupWithUnknownSpecId() {
+ Table table =
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 10,
10);
+ int schemaId = table.schema().schemaId();
+
+ assertThatThrownBy(() -> cache.serializerWithSchemaAndSpec("table",
schemaId, 999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("SpecId:999 not found in table table. Expected SpecId:999
to exist.");
+ }
+
@Test
void testCacheEviction() {
- cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 0);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 0, 10);
assertThat(cache.maximumSize()).isEqualTo(0);
- Supplier<RowDataSerializer> creator1 =
- () -> cache.serializer("table", schema1,
PartitionSpec.unpartitioned());
- Supplier<RowDataSerializer> creator2 =
- () -> cache.serializer("table2", schema2,
PartitionSpec.unpartitioned());
+ Supplier<RowDataSerializer> creator1 = () -> cache.serializer("table",
schema1);
+ Supplier<RowDataSerializer> creator2 = () -> cache.serializer("table2",
schema2);
RowDataSerializer serializer1 = creator1.get();
RowDataSerializer serializer2 = creator2.get();
@@ -118,7 +210,7 @@ class TestTableSerializerCache {
@Test
void testCacheSize() {
- cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
+ cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000,
10);
assertThat(cache.maximumSize()).isEqualTo(1000);
}
@@ -129,7 +221,7 @@ class TestTableSerializerCache {
LoadCountingCatalogLoader catalogLoader =
new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader());
- cache = new TableSerializerCache(catalogLoader, 10);
+ cache = new TableSerializerCache(catalogLoader, 10, 10);
// schema/spec ids are unknown, so both lookups miss the cache and need
the catalog
cache.serializerWithSchemaAndSpec(
