{"record":{"id":"726084d16cf7f40e","repo":"apache/iceberg","slug":"fail-to-deserialize-aggregated-statistics-change-t-726084","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":"error","filePath":"flink/v2.2/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.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java#L63-L99","documentation":"Thrown by StatisticsUtil.deserializeCompletedStatistics when the deserialized CompletedStatistics fails its isValid() check, indicating the payload was written by a newer serializer version (e.g. v2 sort keys) that this code path cannot interpret as valid. It signals fallback to v1 serialization is needed; the same message is also used when falling back. As a RuntimeException it propagates when the v1 fallback itself cannot be prepared.","triggerScenarios":"Restoring aggregated statistics bytes written with the latest sort-key serializer version into a job whose current serializer version disagrees, so the deserialized object is invalid; failures inside the primary deserialize path caught by the generic catch(Exception).","commonSituations":"Flink job upgrade where the checkpointed aggregated statistics were written with v2 sort-key serialization but restored with v1 settings, or rolling upgrades mixing task versions.","solutions":["Let the built-in fallback run: it retries with statisticsSerializer.changeSortKeySerializerVersion(1) and restores the latest version afterwards.","If this RuntimeException escapes, the v1 fallback also failed — restore with the original Iceberg version that wrote the checkpoint.","Migrate state by draining the job with the old version and restarting clean with the new version."],"exampleFix":null,"handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  CompletedStatistics s = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);\n} catch (RuntimeException e) {\n  // both current and v1 fallback failed — restore state with the original Iceberg version\n  LOG.error(\"Aggregated statistics unreadable in any version\", e);\n}","preventionTips":["Upgrade and downgrade only along paths the v1 fallback supports.","Drain and restart jobs cleanly rather than restoring checkpoints across major serializer changes.","Keep checkpoints from the pre-upgrade version available for rollback."],"tags":["serialization","version-migration","flink"],"backgroundTag":"incompatible-source-type","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"}