apache/iceberg · error · UncheckedIOException

Fail to deserialize aggregated statistics

Error message

Fail to deserialize aggregated statistics

What it means

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.

Source

Thrown at flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java:97

      CompletedStatistics completedStatistics = statisticsSerializer.deserialize(input);
      if (!completedStatistics.isValid()) {
        throw new RuntimeException("Fail to deserialize aggregated statistics,change to v1");
      }

      return completedStatistics;
    } catch (Exception e) {
      try {
        // If we restore from a lower version, the new version of SortKeySerializer cannot correctly
        // parse the checkpointData, so we need to first switch the version to v1. Once the state
        // data is successfully parsed, we need to switch the serialization version to the latest
        // version to parse the subsequent data passed from the TM.
        statisticsSerializer.changeSortKeySerializerVersion(1);
        DataInputDeserializer input = new DataInputDeserializer(bytes);
        CompletedStatistics deserialize = statisticsSerializer.deserialize(input);
        statisticsSerializer.changeSortKeySerializerVersionLatest();
        return deserialize;
      } catch (IOException ioException) {
        throw new UncheckedIOException("Fail to deserialize aggregated statistics", ioException);
      }
    }
  }

  static byte[] serializeGlobalStatistics(
      GlobalStatistics globalStatistics, TypeSerializer<GlobalStatistics> statisticsSerializer) {
    try {
      DataOutputSerializer out = new DataOutputSerializer(1024);
      statisticsSerializer.serialize(globalStatistics, out);
      return out.getCopyOfBuffer();
    } catch (IOException e) {
      throw new UncheckedIOException("Fail to serialize aggregated statistics", e);
    }
  }

  static GlobalStatistics deserializeGlobalStatistics(
      byte[] bytes, TypeSerializer<GlobalStatistics> statisticsSerializer) {
    try {

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Verify the table sort key (partition/cluster-by columns) has not changed between when the statistics bytes were serialized and deserialization
2. Restore the job from a checkpoint/savepoint matching the current schema, or drop the stale state
3. Ensure the same statisticsSerializer (same serializer version) is used at both serialize and deserialize sites
4. Upgrade both writer and coordinator nodes to the same Iceberg version to keep serializer versions aligned

Example fix

// before: bytes from an old savepoint with different sort key
CompletedStatistics stats = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);
// after: guard for stale state after schema change
if (stateBytesVersion != serializer.currentVersion()) {
  LOG.warn("Discarding statistics serialized with old serializer version");
  return null;
}
CompletedStatistics stats = StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);

// verify serializer version before deserializing
Preconditions.checkState(
    serializerVersion == expectedSortKeyVersion,
    "Statistics serializer version mismatch");

if (bytes == null || bytes.length == 0) { return null; }

try {
  return StatisticsUtil.deserializeCompletedStatistics(bytes, serializer);
} catch (UncheckedIOException e) {
  LOG.warn("Dropping unreadable aggregated statistics", e);
  return null;
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• Failed to serialize sort key
• Failed to close encryption manager
• Failed to close equality delete source
• Failed to close position delete source
• Failed to close current writer

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/5b4bb801dec8ecb5. Report an issue: GitHub.