{"record":{"id":"264e6ec8e4e91826","repo":"apache/iceberg","slug":"hash-distribute-rows-by-equality-fields-even-thou-264e6e","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/v2.3/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/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L1032-L1068","documentation":"Same family as the FlinkSink range warning: when WRITE_DISTRIBUTION_MODE is 'range' but equality fields (primary key / upsert) are present, IcebergSink cannot apply range distribution safely for streaming writes and instead hashes rows by the equality fields via keyBy, logging this warning for backward compatibility rather than failing.","triggerScenarios":"IcebergSink built with .equalityFields(...) (upsert) while the table or builder sets write.distribution-mode=range (DistributionMode.RANGE), possibly with a sortOrder configured.","commonSituations":"Tables configured with range distribution + sort order reused in Flink upsert streaming jobs; configs copied from batch engines; users expecting sorted output files who instead get hash-by-key distribution.","solutions":["Switch to distributionMode(DistributionMode.HASH) for primary-key streaming writes","If sorting is the goal, use batch mode or ensure your distribution/sort intent does not rely on range mode in streaming","Verify the equality fields are correct since they now determine record placement"],"exampleFix":"// before\nIcebergSink.builder().upsert(true).distributionMode(DistributionMode.RANGE).append();\n// after\nIcebergSink.builder().upsert(true).distributionMode(DistributionMode.HASH).append();","handlingStrategy":"validation","validationCode":"if (DistributionMode.RANGE.equals(distributionMode) && !equalityFieldIds.isEmpty()) {\n  throw new IllegalArgumentException(\"Use hash distribution with equality fields\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never combine range mode with upsert/equality fields in Flink","Standardize sink configuration in one helper","Treat this warning as a config bug, not noise"],"tags":["flink","iceberg","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"}