{"record":{"id":"5b4bb801dec8ecb5","repo":"apache/iceberg","slug":"fail-to-deserialize-aggregated-statistics","errorCode":null,"errorMessage":"Fail to deserialize aggregated statistics","messagePattern":"Fail to deserialize aggregated statistics","errorType":"exception","errorClass":"UncheckedIOException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java","lineNumber":97,"sourceCode":"      CompletedStatistics completedStatistics = statisticsSerializer.deserialize(input);\n      if (!completedStatistics.isValid()) {\n        throw new RuntimeException(\"Fail to deserialize aggregated statistics,change to v1\");\n      }\n\n      return completedStatistics;\n    } catch (Exception e) {\n      try {\n        // If we restore from a lower version, the new version of SortKeySerializer cannot correctly\n        // parse the checkpointData, so we need to first switch the version to v1. Once the state\n        // data is successfully parsed, we need to switch the serialization version to the latest\n        // version to parse the subsequent data passed from the TM.\n        statisticsSerializer.changeSortKeySerializerVersion(1);\n        DataInputDeserializer input = new DataInputDeserializer(bytes);\n        CompletedStatistics deserialize = statisticsSerializer.deserialize(input);\n        statisticsSerializer.changeSortKeySerializerVersionLatest();\n        return deserialize;\n      } catch (IOException ioException) {\n        throw new UncheckedIOException(\"Fail to deserialize aggregated statistics\", ioException);\n      }\n    }\n  }\n\n  static byte[] serializeGlobalStatistics(\n      GlobalStatistics globalStatistics, TypeSerializer<GlobalStatistics> statisticsSerializer) {\n    try {\n      DataOutputSerializer out = new DataOutputSerializer(1024);\n      statisticsSerializer.serialize(globalStatistics, out);\n      return out.getCopyOfBuffer();\n    } catch (IOException e) {\n      throw new UncheckedIOException(\"Fail to serialize aggregated statistics\", e);\n    }\n  }\n\n  static GlobalStatistics deserializeGlobalStatistics(\n      byte[] bytes, TypeSerializer<GlobalStatistics> statisticsSerializer) {\n    try {","sourceCodeStart":79,"sourceCodeEnd":115,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java#L79-L115","documentation":"StatisticsUtil.deserializeCompletedStatistics wraps IOException from Flink's TypeSerializer.deserialize when turning serialized bytes back into a CompletedStatistics object for the upsert/adaptive shuffle path. The library throws UncheckedIOException because IOException here indicates corrupted or incompatible serialized statistics, which callers cannot meaningfully recover from.","triggerScenarios":"Calling StatisticsUtil.deserializeCompletedStatistics with bytes produced by a different sort-key serializer version, truncated/corrupted bytes, or bytes serialized with an incompatible schema of the sort key across job restarts/savepoint restores.","commonSituations":"Restoring a Flink job from a savepoint/checkpoint taken with a different table schema (sort key columns changed), upgrading Iceberg/Flink versions where the serializer version differs, or hand-editing state bytes.","solutions":["Verify the table sort key (partition/cluster-by columns) has not changed between when the statistics bytes were serialized and deserialization","Restore the job from a checkpoint/savepoint matching the current schema, or drop the stale state","Ensure the same statisticsSerializer (same serializer version) is used at both serialize and deserialize sites","Upgrade both writer and coordinator nodes to the same Iceberg version to keep serializer versions aligned"],"exampleFix":"// before: bytes from an old savepoint with different sort key\nCompletedStatistics stats = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);\n// after: guard for stale state after schema change\nif (stateBytesVersion != serializer.currentVersion()) {\n  LOG.warn(\"Discarding statistics serialized with old serializer version\");\n  return null;\n}\nCompletedStatistics stats = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);","handlingStrategy":"try-catch","validationCode":"// verify serializer version before deserializing\nPreconditions.checkState(\n    serializerVersion == expectedSortKeyVersion,\n    \"Statistics serializer version mismatch\");","typeGuard":"if (bytes == null || bytes.length == 0) { return null; }","tryCatchPattern":"try {\n  return StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);\n} catch (UncheckedIOException e) {\n  LOG.warn(\"Dropping unreadable aggregated statistics\", e);\n  return null;\n}","preventionTips":["Keep table sort-key schema stable while statistics state is live","Align Iceberg versions across job upgrades before restoring state","Never reuse statistics bytes from checkpoints older than a schema change"],"tags":["flink","serialization","io","unchecked-ioexception"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}