{"record":{"id":"65fb830c89fd9ae9","repo":"apache/iceberg","slug":"hash-distribute-rows-by-equality-fields-even-thou-65fb83","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/FlinkSink.java","lineNumber":667,"sourceCode":"              for (PartitionField partitionField : partitionSpec.fields()) {\n                Preconditions.checkState(\n                    equalityFieldIds.contains(partitionField.sourceId()),\n                    \"In 'hash' distribution mode with equality fields set, source column '%s' of partition field '%s' \"\n                        + \"should be included in equality fields: '%s'\",\n                    table.schema().findColumnName(partitionField.sourceId()),\n                    partitionField,\n                    equalityFieldColumns);\n              }\n              return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));\n            }\n          }\n\n        case RANGE:\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(\n                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);","sourceCodeStart":649,"sourceCodeEnd":685,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L649-L685","documentation":"This is a warning logged by FlinkSink when the table's write.distribution-mode is 'range' but the write also uses equality fields (upsert/primary-key mode). Range distribution combined with equality fields is not safe for Flink streaming writers, so the sink silently falls back to hashing rows by the equality fields via keyBy instead of throwing an exception, preserving backward compatibility.","triggerScenarios":"A Flink job writes to an Iceberg table with write.distribution-mode=range (or DistributionMode.RANGE) while equality-field IDs are configured (primary key / upsert writes) via FlinkSink.builderFor(table)...upsert(true) or equalityFields().","commonSituations":"Developers copying batch/table configs that use range distribution into a streaming upsert job; tables migrated from Spark where range distribution is safe; explicit WRITE_DISTRIBUTION_MODE='range' set on a keyed table.","solutions":["Set write.distribution-mode to 'hash' on the table or via .distributionMode(DistributionMode.HASH) in the sink builder for primary-key streaming writes","If hashing by equality fields is intended behavior, verify the equality field set covers your key so records for the same key land in the same writer","If range distribution is truly required, do it in a batch (non-streaming) job without equality fields"],"exampleFix":"// before\nFlinkSink.forRowData(input)\n    .upsert(true)\n    .distributionMode(DistributionMode.RANGE)\n// after\nFlinkSink.forRowData(input)\n    .upsert(true)\n    .distributionMode(DistributionMode.HASH)","handlingStrategy":"validation","validationCode":"if (table.properties().getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE_NAME, \"none\").equals(\"range\") && !equalityFieldIds.isEmpty()) {\n  throw new IllegalArgumentException(\"Set write.distribution-mode=hash for primary-key Flink streaming writes\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set DistributionMode.HASH for upsert/equality-field streaming writes","Review table properties before reusing tables across batch and streaming jobs","Watch for this warning when enabling Flink job logs in CI smoke tests"],"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"}