{"record":{"id":"4935facdfe8ae222","repo":"apache/iceberg","slug":"invalid-statistics-type-s-should-be-map-or-sket-4935fa","errorCode":null,"errorMessage":"Invalid statistics type: %s. Should be Map or Sketch","messagePattern":"Invalid statistics type: (.+?)\\. Should be Map or Sketch","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java","lineNumber":79,"sourceCode":"    }\n  }\n\n  private AtomicLong roundRobinCounter(int numPartitions) {\n    if (roundRobinCounter == null) {\n      // randomize the starting point to avoid synchronization across subtasks\n      this.roundRobinCounter = new AtomicLong(new Random().nextInt(numPartitions));\n    }\n\n    return roundRobinCounter;\n  }\n\n  private Partitioner<RowData> delegatePartitioner(GlobalStatistics statistics) {\n    if (statistics.type() == StatisticsType.Map) {\n      return new MapRangePartitioner(schema, sortOrder, statistics.mapAssignment());\n    } else if (statistics.type() == StatisticsType.Sketch) {\n      return new SketchRangePartitioner(schema, sortOrder, statistics.rangeBounds());\n    } else {\n      throw new IllegalArgumentException(\n          String.format(\"Invalid statistics type: %s. Should be Map or Sketch\", statistics.type()));\n    }\n  }\n\n  /**\n   * Util method that handles rescale (write parallelism / numPartitions change).\n   *\n   * @param partition partition calculated based on the existing statistics\n   * @param numPartitionsStatsCalculation number of partitions when the assignment was calculated\n   *     based on\n   * @param numPartitions current number of partitions\n   * @return adjusted partition if necessary.\n   */\n  static int adjustPartitionWithRescale(\n      int partition, int numPartitionsStatsCalculation, int numPartitions) {\n    if (numPartitionsStatsCalculation <= numPartitions) {\n      // no rescale or scale-up case.\n      // new subtasks are ignored and not assigned any keys, which is sub-optimal and only","sourceCodeStart":61,"sourceCodeEnd":97,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java#L61-L97","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the upstream statistics-collect operator is configured to emit Map or Sketch statistics (do not disable statistics while resuming old range-partition state).","Restore from a savepoint whose GlobalStatistics type matches the current operator configuration.","Version-align Iceberg flink modules across the pipeline.","As a workaround switch write.distribution.mode to 'hash' which bypasses the range partitioner."],"exampleFix":"// before: resuming with statistics collection turned off\n-- write.distribution-mode range  +  statistics disabled -> NONE type\n// after: keep statistics enabled while using range mode\nALTER TABLE t SET ('write.distribution-mode'='range', 'write.distribution-mode.sort-order-adaptive'='true');","handlingStrategy":"validation","validationCode":"if (statistics.type() != StatisticsType.Map && statistics.type() != StatisticsType.Sketch) {\n  throw new IllegalArgumentException(\"Partitioner requires Map/Sketch stats, got \" + statistics.type());\n}","typeGuard":"boolean partitionable(GlobalStatistics g) {\n  return g.type() == StatisticsType.Map || g.type() == StatisticsType.Sketch;\n}","tryCatchPattern":"try {\n  partitioner = delegatePartitioner(statistics);\n} catch (IllegalArgumentException e) {\n  LOG.warn(\"Stats type {} unusable; using hash partitioner\", statistics.type());\n  partitioner = new HashPartitioner();\n}","preventionTips":["Keep statistics collection enabled whenever write.distribution-mode is range.","Restore checkpoints only with configurations matching those at write time.","Align Iceberg versions across operators."],"tags":["flink","partitioning","range-partitioner"],"backgroundTag":"invalid-enum-value","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}