{"record":{"id":"1c1e21f51927efb0","repo":"apache/iceberg","slug":"fallback-to-use-none-distribution-mode-beca-1c1e21","errorCode":null,"errorMessage":"{}: Fallback to use 'none' distribution mode, because there are no equality fields set and {}='range' is not supported yet in flink","messagePattern":"(.+?): Fallback to use 'none' distribution mode, because there are no equality fields set and (.+?)='range' is not supported yet in flink","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java","lineNumber":177,"sourceCode":"              Types.NestedField sourceField = schema.findField(partitionField.sourceId());\n              Preconditions.checkState(\n                  sourceField != null && equalityFields.contains(sourceField.name()),\n                  \"%s: In 'hash' distribution mode with equality fields set, partition field '%s' \"\n                      + \"should be included in equality fields: '%s'\",\n                  tableName,\n                  partitionField,\n                  schema.columns().stream()\n                      .filter(c -> equalityFields.contains(c.name()))\n                      .collect(Collectors.toList()));\n            }\n            return partitionKeySelector(\n                tableName, schema, spec, writeParallelism, maxWriteParallelism);\n          }\n        }\n\n      case RANGE:\n        if (equalityFields.isEmpty()) {\n          LOG.warn(\n              \"{}: Fallback to use 'none' distribution mode, because there are no equality fields set \"\n                  + \"and {}='range' is not supported yet in flink\",\n              tableName,\n              WRITE_DISTRIBUTION_MODE);\n          return tableKeySelector(tableName, writeParallelism, maxWriteParallelism);\n        } else {\n          LOG.info(\n              \"{}: Distribute rows by equality fields, because there are equality fields set \"\n                  + \"and {}='range' is not supported yet in flink\",\n              tableName,\n              WRITE_DISTRIBUTION_MODE);\n          return equalityFieldKeySelector(\n              tableName, schema, equalityFields, writeParallelism, maxWriteParallelism);\n        }\n\n      default:\n        throw new IllegalArgumentException(\n            tableName + \": Unrecognized \" + WRITE_DISTRIBUTION_MODE + \": \" + mode);","sourceCodeStart":159,"sourceCodeEnd":195,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java#L159-L195","documentation":"HashKeyGenerator.getKeySelector logs this warning when RANGE distribution mode is requested but no equality fields are set. Range distribution by partition key is not supported in the Flink sink, so with no equality fields there is nothing to range-assign on; the generator falls back to 'none' distribution mode via tableKeySelector.","triggerScenarios":"DynamicRecord/sink config with WRITE_DISTRIBUTION_MODE='range' and an empty equalityFields set; the RANGE branch of getKeySelector warns and returns tableKeySelector instead of a range key selector.","commonSituations":"Migrating a Spark Iceberg writer configuration (which supports range distribution) to the Flink sink without adjusting write.distribution-mode; generic table properties like write.distribution-mode=range applied to Flink jobs.","solutions":["Set equality fields so the sink can key records for distribution.","Change write.distribution-mode to 'none' or 'hash' in the Flink sink configuration.","Do range clustering upstream (e.g. sort/rewrite with Spark) instead of relying on Flink range distribution."],"exampleFix":"// before\ntable property: write.distribution-mode=range  (Flink job, no equality fields)\n\n// after\ntable property: write.distribution-mode=hash  + equalityFields=[\"id\"]\n// or write.distribution-mode=none","handlingStrategy":"validation","validationCode":"if (\"range\".equals(writeDistributionMode) && equalityFields.isEmpty()) {\n  throw new IllegalArgumentException(\n      \"Flink sink does not support range distribution without equality fields; use 'hash' or 'none'\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Do not copy Spark writer distribution settings (range) into Flink sink configs unmodified.","Choose hash+equality-fields when clustering is needed in Flink.","Handle range clustering offline via compaction/rewrite tooling instead."],"tags":["flink","dynamic-sink","distribution-mode","range"],"backgroundTag":"unsupported-operation","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}