{"record":{"id":"4f4e6162110dac61","repo":"apache/iceberg","slug":"hash-distribute-rows-by-equality-fields-even-thou-4f4e61","errorCode":null,"errorMessage":"Hash distribute rows by equality fields, even though {}=range is set. Range distribution for primary keys are not always safe in Flink streaming writer.","messagePattern":"Hash distribute rows by equality fields, even though (.+?)=range is set\\. Range distribution for primary keys are not always safe in Flink streaming writer\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java","lineNumber":1050,"sourceCode":"    return Optional.ofNullable(flinkWriteConf.writeParallelism()).orElseGet(input::getParallelism);\n  }\n\n  private DataStream<RowData> distributeDataStreamByRangeDistributionMode(\n      DataStream<RowData> input,\n      Schema iSchema,\n      PartitionSpec partitionSpec,\n      SortOrder sortOrderParam) {\n\n    int writerParallelism = resolveWriterParallelism(input);\n\n    // needed because of checkStyle not allowing us to change the value of an argument\n    SortOrder sortOrder = sortOrderParam;\n\n    // Ideally, exception should be thrown in the combination of range distribution and\n    // equality fields. Primary key case should use hash distribution mode.\n    // Keep the current behavior of falling back to keyBy for backward compatibility.\n    if (!equalityFieldIds.isEmpty()) {\n      LOG.warn(\n          \"Hash distribute rows by equality fields, even though {}=range is set. \"\n              + \"Range distribution for primary keys are not always safe in \"\n              + \"Flink streaming writer.\",\n          WRITE_DISTRIBUTION_MODE);\n      return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));\n    }\n\n    // range distribute by partition key or sort key if table has an SortOrder\n    Preconditions.checkState(\n        sortOrder.isSorted() || partitionSpec.isPartitioned(),\n        \"Invalid write distribution mode: range. Need to define sort order or partition spec.\");\n    if (sortOrder.isUnsorted()) {\n      sortOrder = Partitioning.sortOrderFor(partitionSpec);\n      LOG.info(\"Construct sort order from partition spec\");\n    }\n\n    LOG.info(\"Range distribute rows by sort order: {}\", sortOrder);\n    StatisticsOrRecordTypeInformation statisticsOrRecordTypeInformation =","sourceCodeStart":1032,"sourceCodeEnd":1068,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L1032-L1068","documentation":"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.","triggerScenarios":"Setting write.distribution-mode=range on a table that has identifier fields, while the job specifies equality fields for an upsert/rowdelta sink.","commonSituations":"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.","solutions":["Change write.distribution-mode to 'hash' (or 'none') for the Flink upsert job.","Keep range distribution only for append-only tables without equality fields.","If range is truly desired, remove the equality fields / upsert semantics so the fallback doesn't apply."],"exampleFix":"// before\ntable.updateProperties().set(\"write.distribution-mode\", \"range\").commit();\n// after\ntable.updateProperties().set(\"write.distribution-mode\", \"hash\").commit();","handlingStrategy":"validation","validationCode":"if (\"range\".equals(table.properties().get(\"write.distribution-mode\"))\n    && !table.schema().identifierFieldIds().isEmpty()) {\n  throw new IllegalArgumentException(\"Range distribution is unsafe with equality fields in Flink streaming; use hash\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Use hash distribution for primary-key/upsert Flink jobs.","Audit table properties when reusing tables across engines."],"tags":["flink","iceberg-sink","distribution-mode","upsert"],"backgroundTag":"conflicting-config-options","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"}