{"record":{"id":"b7518a99e4459c48","repo":"apache/iceberg","slug":"fallback-to-use-none-distribution-mode-because-b7518a","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":"info","filePath":"flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L616-L652","documentation":"In FlinkSink's HASH distribution mode, rows are keyed by equality fields (upsert) or partition key. With no equality fields configured and an unpartitioned table there is nothing to hash by, so the sink logs this warning and falls back to 'none' distribution, sending rows as-is.","triggerScenarios":"Builder has distribution mode HASH (or WRITE_DISTRIBUTION_MODE default resolves to it in streaming) while equalityFieldIds is empty and partitionSpec.isUnpartitioned() in distributeDataStream.","commonSituations":"Writing append-only data to an unpartitioned table with distribution-mode=hash configured; default streaming writes where users assumed hash is safe without a partition or PK.","solutions":["Use distribution mode 'none' explicitly for unpartitioned append-only tables","Partition the table so HASH mode can key by partition","Set equality fields (primary key) if writing upserts","Remove the hash distribution setting if no key exists"],"exampleFix":"// before\nFlinkSink.forRowData(input).distributionMode(DistributionMode.HASH)...\n// after\nFlinkSink.forRowData(input).distributionMode(DistributionMode.NONE)...","handlingStrategy":"validation","validationCode":"if (distributionMode == DistributionMode.HASH\n    && equalityFieldIds.isEmpty()\n    && table.spec().isUnpartitioned()) {\n  // no key exists; request NONE explicitly or add partitions/PK\n  distributionMode = DistributionMode.NONE;\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Only request HASH distribution when a partition spec or primary key exists","Prefer NONE for unpartitioned append-only streams","Set distribution mode deliberately per table, not globally"],"tags":["flink","iceberg","distribution","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"}