{"record":{"id":"2a2989252c53a6c3","repo":"apache/iceberg","slug":"fail-to-deserialize-aggregated-statistics-change-t-2a2989","errorCode":null,"errorMessage":"Fail to deserialize aggregated statistics,change to v1","messagePattern":"Fail to deserialize aggregated statistics,change to v1","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java","lineNumber":81,"sourceCode":"  static byte[] serializeCompletedStatistics(\n      CompletedStatistics completedStatistics,\n      TypeSerializer<CompletedStatistics> statisticsSerializer) {\n    try {\n      DataOutputSerializer out = new DataOutputSerializer(1024);\n      statisticsSerializer.serialize(completedStatistics, out);\n      return out.getCopyOfBuffer();\n    } catch (IOException e) {\n      throw new UncheckedIOException(\"Fail to serialize aggregated statistics\", e);\n    }\n  }\n\n  static CompletedStatistics deserializeCompletedStatistics(\n      byte[] bytes, CompletedStatisticsSerializer statisticsSerializer) {\n    try {\n      DataInputDeserializer input = new DataInputDeserializer(bytes);\n      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    }","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java#L63-L99","documentation":"StatisticsUtil.deserializeCompletedStatistics throws a plain RuntimeException \"Fail to deserialize aggregated statistics,change to v1\" when bytes deserialized into a CompletedStatistics that reports !isValid(), signalling the payload was not written with the current (latest) sort-key serializer version. The catch block then retries the deserialization with sort key serializer version 1, so this exception is an internal control-flow signal; if it escapes, the fallback also failed.","triggerScenarios":"Deserializing completed statistics whose bytes were produced with the v1 SortKeySerializer (older Iceberg checkpoint) while the current job uses v2 — the validity check fails and the v1-fallback marker is thrown; if surrounding code doesn't catch it, state restoration aborts.","commonSituations":"Upgrading iceberg-flink-runtime while restoring from a savepoint written before the SortKeySerializer version bump; mixing old and new operator states during rescale.","solutions":["This should be transparently handled by the internal v1 fallback — if you see it escape, check that you are calling StatisticsUtil.deserializeCompletedStatistics (not custom deserialization) so the fallback path runs.","Upgrade both job stages to the same iceberg-flink-runtime version so v2 serialization is used end-to-end.","Start from a fresh checkpoint if the fallback also fails; statistics are advisory and will be rebuilt."],"exampleFix":"// before: custom code without fallback\nCompletedStatistics s = completedSerializer.deserialize(new DataInputDeserializer(bytes));\n// after: use version-aware util\nCompletedStatistics s = StatisticsUtil.deserializeCompletedStatistics(bytes, completedSerializer);","handlingStrategy":"try-catch","validationCode":"// Ensure bytes came from StatisticsUtil.serializeCompletedStatistics of a compatible version\nif (bytes == null || bytes.length == 0) { return null; }","typeGuard":null,"tryCatchPattern":"try {\n  stats = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);\n} catch (RuntimeException | UncheckedIOException e) {\n  // v1 fallback failed too; rebuild statistics from scratch\n  stats = null;\n}","preventionTips":["Always deserialize through StatisticsUtil so the v1 fallback runs","Upgrade writer and reader to the same iceberg-flink-runtime version","Start fresh checkpoints after major Iceberg upgrades"],"tags":["flink","deserialization","version-compatibility","statistics"],"backgroundTag":"unexpected-response-shape","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}