apache/iceberg · warning

Hash distribute rows by equality fields, even though {}=rang

Error message

Hash distribute rows by equality fields, even though {}=range is set. Range distribution for primary keys are not always safe in Flink streaming writer.

What it means

When write.distribution-mode=range is combined with equality fields (primary keys), the sink cannot safely use range shuffle for upserts in a Flink streaming writer, so it warns and falls back to a hash keyBy on the equality fields. Range distribution for keyed rows can skew and break delete-file correctness assumptions.

Source

Thrown at flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:1050

    return Optional.ofNullable(flinkWriteConf.writeParallelism()).orElseGet(input::getParallelism);
  }

  private DataStream<RowData> distributeDataStreamByRangeDistributionMode(
      DataStream<RowData> input,
      Schema iSchema,
      PartitionSpec partitionSpec,
      SortOrder sortOrderParam) {

    int writerParallelism = resolveWriterParallelism(input);

    // needed because of checkStyle not allowing us to change the value of an argument
    SortOrder sortOrder = sortOrderParam;

    // Ideally, exception should be thrown in the combination of range distribution and
    // equality fields. Primary key case should use hash distribution mode.
    // Keep the current behavior of falling back to keyBy for backward compatibility.
    if (!equalityFieldIds.isEmpty()) {
      LOG.warn(
          "Hash distribute rows by equality fields, even though {}=range is set. "
              + "Range distribution for primary keys are not always safe in "
              + "Flink streaming writer.",
          WRITE_DISTRIBUTION_MODE);
      return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
    }

    // range distribute by partition key or sort key if table has an SortOrder
    Preconditions.checkState(
        sortOrder.isSorted() || partitionSpec.isPartitioned(),
        "Invalid write distribution mode: range. Need to define sort order or partition spec.");
    if (sortOrder.isUnsorted()) {
      sortOrder = Partitioning.sortOrderFor(partitionSpec);
      LOG.info("Construct sort order from partition spec");
    }

    LOG.info("Range distribute rows by sort order: {}", sortOrder);
    StatisticsOrRecordTypeInformation statisticsOrRecordTypeInformation =

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Change write.distribution-mode to 'hash' (or 'none') for the Flink upsert job.
2. Keep range distribution only for append-only tables without equality fields.
3. If range is truly desired, remove the equality fields / upsert semantics so the fallback doesn't apply.

Example fix

// before
table.updateProperties().set("write.distribution-mode", "range").commit();
// after
table.updateProperties().set("write.distribution-mode", "hash").commit();

if ("range".equals(table.properties().get("write.distribution-mode"))
    && !table.schema().identifierFieldIds().isEmpty()) {
  throw new IllegalArgumentException("Range distribution is unsafe with equality fields in Flink streaming; use hash");
}

When it happens

Trigger: Setting write.distribution-mode=range on a table that has identifier fields, while the job specifies equality fields for an upsert/rowdelta sink.

Common situations: Reusing batch/Spark-style range-distribution config in a Flink streaming upsert job; tables with primary keys configured with a range distribution inherited from another engine's defaults.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

• Fallback to use 'none' distribution mode, because there are
• Hash distribute rows by equality fields, even though {}=rang
• Hash distribute rows by equality fields, even though {}=rang
• Failed to serialize PK index key
• ${tableName}: Unrecognized ${WRITE_DISTRIBUTION_MODE}: ${mod

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