{"record":{"id":"db84fc48fa414045","repo":"apache/iceberg","slug":"hash-distribute-rows-by-equality-fields-even-thou","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/FlinkSink.java","lineNumber":665,"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":647,"sourceCodeEnd":683,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L647-L683","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Change write.distribution-mode to hash for the primary-key table (recommended)","Set distribution-mode=none if explicit control over keying is not needed","Remove the range setting from table properties inherited by Flink jobs","Drop identifier fields only if range append is truly intended and dedup handled elsewhere","Ignore the warning if you accept the hash-by-equality-fields fallback behavior"],"exampleFix":"// before\nALTER TABLE db.tbl SET TBLPROPERTIES ('write.distribution-mode'='range');\n// after\nALTER TABLE db.tbl SET TBLPROPERTIES ('write.distribution-mode'='hash');","handlingStrategy":"validation","validationCode":"if (\"range\".equals(table.properties().get(\"write.distribution-mode\"))\n    && !table.schema().identifierFieldIds().isEmpty()) {\n  // switch to hash for streaming upsert writers\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Use hash distribution for primary-key tables in Flink streaming writes","Don't copy batch/Spark-oriented range distribution settings to streaming upsert jobs","Audit cluster-wide table property defaults for conflicts with upsert mode","Re-validate table properties after adding identifier fields"],"tags":["flink","sink","distribution","upsert","configuration"],"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"}