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

In FlinkSink.distributeDataStream(), when write.distribution-mode=range is configured but equality fields are present (primary-key tables), range distribution is unsafe for Flink streaming upsert writers. The sink logs this warning and falls back to hashing rows by the equality fields via EqualityFieldKeySelector, keeping backward-compatible behavior instead of throwing.

Source

Thrown at flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:665

              for (PartitionField partitionField : partitionSpec.fields()) {
                Preconditions.checkState(
                    equalityFieldIds.contains(partitionField.sourceId()),
                    "In 'hash' distribution mode with equality fields set, source column '%s' of partition field '%s' "
                        + "should be included in equality fields: '%s'",
                    table.schema().findColumnName(partitionField.sourceId()),
                    partitionField,
                    equalityFieldColumns);
              }
              return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
            }
          }

        case RANGE:
          // 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);

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Change write.distribution-mode to hash for the primary-key table (recommended)
2. Set distribution-mode=none if explicit control over keying is not needed
3. Remove the range setting from table properties inherited by Flink jobs
4. Drop identifier fields only if range append is truly intended and dedup handled elsewhere
5. Ignore the warning if you accept the hash-by-equality-fields fallback behavior

Example fix

// before
ALTER TABLE db.tbl SET TBLPROPERTIES ('write.distribution-mode'='range');
// after
ALTER TABLE db.tbl SET TBLPROPERTIES ('write.distribution-mode'='hash');

if ("range".equals(table.properties().get("write.distribution-mode"))
    && !table.schema().identifierFieldIds().isEmpty()) {
  // switch to hash for streaming upsert writers
}

When it happens

Trigger: distributeDataStream() hits the RANGE case with a non-empty equalityFieldIds — i.e. a primary-key/equality-field job (upsert) whose table property write.distribution-mode is set to range.

Common situations: Table property set to range by a batch-oriented convention (e.g. copied from Spark write guidance) while the Flink streaming upsert writer requires hash-by-key; schema identifier fields added later making equalityFieldIds non-empty; global cluster defaults forcing range mode.

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

• The configured equality field column IDs {} are not matched
• Fallback to use 'none' distribution mode, because there are
• Unrecognized WRITE_DISTRIBUTION_MODE:
• Unrecognized WRITE_DISTRIBUTION_MODE:
• Fallback to use 'none' distribution mode, because there are

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