{"record":{"id":"ab7ec2d4884f8bca","repo":"apache/iceberg","slug":"fallback-to-use-none-distribution-mode-because-ab7ec2","errorCode":null,"errorMessage":"Fallback to use 'none' distribution mode, because there are no equality fields set and table is unpartitioned","messagePattern":"Fallback to use 'none' distribution mode, because there are no equality fields set and table is unpartitioned","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":634,"sourceCode":"\n      Schema iSchema = table.schema();\n      PartitionSpec partitionSpec = table.spec();\n      SortOrder sortOrder = table.sortOrder();\n\n      switch (writeMode) {\n        case NONE:\n          if (equalityFieldIds.isEmpty()) {\n            return input;\n          } else {\n            LOG.info(\"Distribute rows by equality fields, because there are equality fields set\");\n            return input.keyBy(\n                new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));\n          }\n\n        case HASH:\n          if (equalityFieldIds.isEmpty()) {\n            if (partitionSpec.isUnpartitioned()) {\n              LOG.warn(\n                  \"Fallback to use 'none' distribution mode, because there are no equality fields set \"\n                      + \"and table is unpartitioned\");\n              return input;\n            } else {\n              return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));\n            }\n          } else {\n            if (partitionSpec.isUnpartitioned()) {\n              LOG.info(\n                  \"Distribute rows by equality fields, because there are equality fields set \"\n                      + \"and table is unpartitioned\");\n              return input.keyBy(\n                  new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));\n            } else {\n              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' \"","sourceCodeStart":616,"sourceCodeEnd":652,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L616-L652","documentation":"When WRITE_DISTRIBUTION_MODE=hash is requested but the sink has no equality fields and the table is unpartitioned, there is no key to hash by, so FlinkSink silently falls back to the 'none' (no shuffle) distribution and logs this warning. Rows will not be clustered by any key.","triggerScenarios":"distributeDataStream with DistributionMode.HASH, equalityFieldIds empty, and partitionSpec.isUnpartitioned() true — e.g. setting write.distribution-mode=hash on an unpartitioned table without primary keys.","commonSituations":"Copy-pasting sink config with write.distribution-mode=hash onto an unpartitioned, key-less table; expecting hash mode to reduce small files when no partition or key exists.","solutions":["Remove write.distribution-mode=hash or set it to none for this table.","Add a partition spec to the table if clustering by partition is desired.","Configure equalityFieldColumns if the table actually has a logical key to hash by."],"exampleFix":"// before\ntable.updateProperties().set(TableProperties.WRITE_DISTRIBUTION_MODE, \"hash\").commit(); // unpartitioned table\n// after\ntable.updateProperties().set(TableProperties.WRITE_DISTRIBUTION_MODE, \"none\").commit();","handlingStrategy":"validation","validationCode":"if (mode == DistributionMode.HASH && eqFieldIds.isEmpty() && table.spec().isUnpartitioned()) {\n  mode = DistributionMode.NONE; // avoid the silent fallback\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Only set write.distribution-mode=hash on partitioned or key-ed tables","Check table.spec().isUnpartitioned() before choosing a distribution mode","Centralize distribution-mode selection logic in one helper"],"tags":["flink","distribution-mode","configuration","partitioning"],"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-23T08:17:48.524Z"}