{"record":{"id":"1d885f93dd935030","repo":"apache/iceberg","slug":"fallback-to-use-none-distribution-mode-because","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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":632,"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":614,"sourceCodeEnd":650,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L614-L650","documentation":"In FlinkSink.distributeDataStream(), HASH distribution with no equality fields on an unpartitioned table has nothing meaningful to key by, so it logs this warning and returns the input unchanged — effectively falling back to NONE distribution. Rows are written without keying, which is fine for appends but changes expected shuffle behavior.","triggerScenarios":"write.distribution-mode=hash configured (or defaulted by upsert mode) while equalityFieldIds is empty AND the table partition spec is unpartitioned. Typical with equality-field-columns not set on an append job over an unpartitioned table.","commonSituations":"Setting distribution-mode=hash globally for a job writing to unpartitioned tables; upsert-related defaults applied but equality fields removed; table migrated from partitioned to unpartitioned while job config stayed fixed.","solutions":["Set write.distribution-mode=none in the table properties or via FlinkOptions to make the behavior explicit","Add equality-field-columns if keyed distribution is actually required","Partition the table if hash distribution by partition key is desired","Ignore the warning if unkeyed append is intended","Verify job config matches the current table layout after table evolution"],"exampleFix":"// before (table property)\n\"write.distribution-mode\" = \"hash\"  // unpartitioned table, no equality fields\n// after\nALTER TABLE db.tbl SET TBLPROPERTIES ('write.distribution-mode'='none');","handlingStrategy":"validation","validationCode":"if (\"hash\".equals(table.properties().get(\"write.distribution-mode\"))\n    && table.spec().isUnpartitioned()\n    && equalityFieldColumns.isEmpty()) {\n  // set mode=none explicitly or add equality fields\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Match distribution-mode to the table layout (hash needs partitioned or keyed tables)","Review table properties after partition-spec changes","Document intended distribution mode per table","Use write.distribution-mode=none for plain appends to unpartitioned tables"],"tags":["flink","sink","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"}