{"record":{"id":"b92567720428d3b8","repo":"apache/iceberg","slug":"performance-degraded-as-records-with-different-sch-b92567","errorCode":null,"errorMessage":"Performance degraded as records with different schema is generated for the same table. Likely the DynamicRecord.schema is not reused. Reuse the same instance if the record schema is the same to improve performance","messagePattern":"Performance degraded as records with different schema is generated for the same table\\. Likely the DynamicRecord\\.schema is not reused\\. Reuse the same instance if the record schema is the same to improve performance","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java","lineNumber":280,"sourceCode":"    private CacheItem(\n        long createdTimestampMillis,\n        boolean tableExists,\n        Set<String> branches,\n        Map<Integer, Schema> tableSchemas,\n        Map<Integer, PartitionSpec> specs,\n        int inputSchemaCacheMaximumSize) {\n      this.createdTimestampMillis = createdTimestampMillis;\n      this.tableExists = tableExists;\n      this.branches = branches;\n      this.tableSchemas = tableSchemas;\n      this.specs = specs;\n      this.inputSchemas =\n          new LRUCache<>(inputSchemaCacheMaximumSize, CacheItem::inputSchemaEvictionListener);\n    }\n\n    private static void inputSchemaEvictionListener(\n        Map.Entry<Schema, ResolvedSchemaInfo> evictedEntry) {\n      LOG.warn(\n          \"Performance degraded as records with different schema is generated for the same table. \"\n              + \"Likely the DynamicRecord.schema is not reused. \"\n              + \"Reuse the same instance if the record schema is the same to improve performance\");\n    }\n\n    @VisibleForTesting\n    Map<Schema, ResolvedSchemaInfo> inputSchemas() {\n      return inputSchemas;\n    }\n  }\n\n  static class ResolvedSchemaInfo {\n    private final Schema resolvedTableSchema;\n    private final CompareSchemasVisitor.Result compareResult;\n    private final DataConverter recordConverter;\n\n    ResolvedSchemaInfo(\n        Schema tableSchema,","sourceCodeStart":262,"sourceCodeEnd":298,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java#L262-L298","documentation":"TableMetadataCache keeps an LRU of input Schemas mapped to resolved schema info. Each entry registers inputSchemaEvictionListener, which fires when a Schema is evicted from the cache. Frequent evictions mean DynamicRecords arriving with structurally identical schemas are carrying distinct Schema instances, defeating the cache and degrading sink performance. This is a performance warning, not a failure.","triggerScenarios":"The inputSchema LRUCache (capacity inputSchemaCacheMaximumSize) evicts entries because upstream operators deserialize DynamicRecords independently, creating new Schema objects per record instead of reusing one instance for equal schemas.","commonSituations":"High-churn dynamic sink pipelines where records pass through serialization boundaries (e.g. network shuffle) without schema canonicalization; many distinct tables/schemas exceeding the cache size so legit churn looks like reuse failure.","solutions":["Ensure DynamicRecord producers reuse a single Schema instance per distinct schema (intern/canonicalize schemas before emitting records).","Increase the input schema cache maximum size (cache maximum size config) if many genuinely distinct schemas are in use.","Check upstream operators for unnecessary serialization/deserialization of DynamicRecords that re-create Schema objects."],"exampleFix":"// before\nfor (Row row : input) {\n  out.collect(new DynamicRecord(table, new Schema(...), row)); // new Schema every record\n}\n\n// after\nprivate static final Schema SCHEMA = new Schema(...);\nfor (Row row : input) {\n  out.collect(new DynamicRecord(table, SCHEMA, row)); // reuse instance\n}","handlingStrategy":"type-guard","validationCode":"// Before emitting DynamicRecords, canonicalize schemas:\nSchema canonical = schemaCache.computeIfAbsent(candidateSchema, s -> s);\n// then construct records only with `canonical`","typeGuard":"private boolean isCanonical(Schema emitted, java.util.concurrent.ConcurrentHashMap<Schema, Schema> cache) {\n  return cache.getOrDefault(emitted, emitted) == emitted;\n}","tryCatchPattern":null,"preventionTips":["Intern Schema instances at the pipeline source (one instance per distinct schema).","Avoid creating Schema objects inside per-record loops or map functions.","Size the input schema cache for the actual number of distinct live schemas."],"tags":["flink","dynamic-sink","cache","performance","schema-reuse"],"backgroundTag":"cache-eviction-thrashing","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}