{"record":{"id":"8f0f24d762566c35","repo":"apache/iceberg","slug":"invalid-statistics-type-s-should-be-map-or-sket","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/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java#L61-L97","documentation":"RangePartitioner.delegatePartitioner selects the actual partitioner based on the GlobalStatistics type: Map yields MapRangePartitioner, Sketch yields SketchRangePartitioner. Any other type throws IllegalArgumentException('Invalid statistics type: %s. Should be Map or Sketch'), because the partitioner cannot interpret other statistics payloads.","triggerScenarios":"partition() receives a GlobalStatistics whose type() is neither Map nor Sketch — e.g. an uninitialized/placeholder GlobalStatistics, or statistics deserialized from a different connector version with an extra type.","commonSituations":"GlobalStatistics propagated to writers before the coordinator produced real statistics; version skew between writer tasks and coordinator; savepoint restore across incompatible connector versions.","solutions":["Ensure the coordinator has completed at least one checkpoint so real global statistics (Map or Sketch) are distributed to writers","Align Iceberg connector versions across the job, especially when restoring from a savepoint","Log/inspect the type value printed in the message and trace where the GlobalStatistics instance was constructed","If the type is a valid new type from a newer release, upgrade the partitioner side"],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":"if (statistics != null && statistics.type() != StatisticsType.Map && statistics.type() != StatisticsType.Sketch) {\n  throw new IllegalArgumentException(\"GlobalStatistics type not partitioner-compatible: \" + statistics.type());\n}","typeGuard":"static boolean partitionerUsable(GlobalStatistics gs) {\n  return gs != null && (gs.type() == StatisticsType.Map || gs.type() == StatisticsType.Sketch);\n}","tryCatchPattern":"try {\n  Partitioner<RowData> p = partitioner.partition(statistics);\n} catch (IllegalArgumentException e) {\n  // fall back to default (round-robin/hash) partitioning until valid statistics arrive\n}","preventionTips":["Wait for the first completed checkpoint before expecting range partitioning to engage","Keep writer and coordinator connector versions identical","Validate savepoint compatibility before restore"],"tags":["flink","range-partitioning","data-statistics","type-mismatch"],"backgroundTag":"unsupported-enum-value","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}