apache/iceberg · error · IllegalArgumentException

Invalid statistics type

Error message

Invalid statistics type: %s. Should be Map or Sketch

What it means

RangePartitioner.delegatePartitioner() builds the per-record partitioner from the redistributed GlobalStatistics. Only StatisticsType.Map (MapRangePartitioner) and StatisticsType.Sketch (SketchRangePartitioner) are supported; any other type makes the range partitioner unable to produce range bounds, so it throws IllegalArgumentException naming the offending type. It means the partitioner received statistics it cannot interpret.

Solutions

1. Ensure the upstream statistics-collect operator is configured to emit Map or Sketch statistics (do not disable statistics while resuming old range-partition state).
2. Restore from a savepoint whose GlobalStatistics type matches the current operator configuration.
3. Version-align Iceberg flink modules across the pipeline.
4. As a workaround switch write.distribution.mode to 'hash' which bypasses the range partitioner.

Example fix

// before: resuming with statistics collection turned off
-- write.distribution-mode range  +  statistics disabled -> NONE type
// after: keep statistics enabled while using range mode
ALTER TABLE t SET ('write.distribution-mode'='range', 'write.distribution-mode.sort-order-adaptive'='true');

if (statistics.type() != StatisticsType.Map && statistics.type() != StatisticsType.Sketch) {
  throw new IllegalArgumentException("Partitioner requires Map/Sketch stats, got " + statistics.type());
}

boolean partitionable(GlobalStatistics g) {
  return g.type() == StatisticsType.Map || g.type() == StatisticsType.Sketch;
}

try {
  partitioner = delegatePartitioner(statistics);
} catch (IllegalArgumentException e) {
  LOG.warn("Stats type {} unusable; using hash partitioner", statistics.type());
  partitioner = new HashPartitioner();
}

When it happens

Trigger: The shuffle operator receives a GlobalStatistics whose type() is neither Map nor Sketch — e.g. StatisticsType.NONE after statistics collection was disabled or fell back, or a GlobalStatistics restored from state written by an incompatible version.

Common situations: Enabling/disabling key-level statistics collection mid-job so restored state type mismatches the running operator config; version skew between checkpoint writer and reader; misuse of the operator API passing a GlobalStatistics with NONE type.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Already closed files for partition:
• Altering partition keys is not supported yet.
• Altering partition keys is not supported yet.
• Altering partition keys is not supported yet.
• Altering partition keys is not supported yet.

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

Appendix: source

Thrown at flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java:79

    }
  }

  private AtomicLong roundRobinCounter(int numPartitions) {
    if (roundRobinCounter == null) {
      // randomize the starting point to avoid synchronization across subtasks
      this.roundRobinCounter = new AtomicLong(new Random().nextInt(numPartitions));
    }

    return roundRobinCounter;
  }

  private Partitioner<RowData> delegatePartitioner(GlobalStatistics statistics) {
    if (statistics.type() == StatisticsType.Map) {
      return new MapRangePartitioner(schema, sortOrder, statistics.mapAssignment());
    } else if (statistics.type() == StatisticsType.Sketch) {
      return new SketchRangePartitioner(schema, sortOrder, statistics.rangeBounds());
    } else {
      throw new IllegalArgumentException(
          String.format("Invalid statistics type: %s. Should be Map or Sketch", statistics.type()));
    }
  }

  /**
   * Util method that handles rescale (write parallelism / numPartitions change).
   *
   * @param partition partition calculated based on the existing statistics
   * @param numPartitionsStatsCalculation number of partitions when the assignment was calculated
   *     based on
   * @param numPartitions current number of partitions
   * @return adjusted partition if necessary.
   */
  static int adjustPartitionWithRescale(
      int partition, int numPartitionsStatsCalculation, int numPartitions) {
    if (numPartitionsStatsCalculation <= numPartitions) {
      // no rescale or scale-up case.
      // new subtasks are ignored and not assigned any keys, which is sub-optimal and only

View on GitHub (pinned to 86d9c8fc54)