{"record":{"id":"5f7e0208f190cb39","repo":"apache/iceberg","slug":"fallback-to-use-none-distribution-mode-because-5f7e02","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/IcebergSink.java","lineNumber":1001,"sourceCode":"        throw new RuntimeException(\"Unrecognized \" + WRITE_DISTRIBUTION_MODE + \": \" + mode);\n    }\n  }\n\n  private DataStream<RowData> distributeDataStreamByNoneDistributionMode(\n      DataStream<RowData> input, Schema iSchema) {\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(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));\n    }\n  }\n\n  private DataStream<RowData> distributeDataStreamByHashDistributionMode(\n      DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {\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(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' \"\n                  + \"should be included in equality fields: '%s'\",","sourceCodeStart":983,"sourceCodeEnd":1019,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L983-L1019","documentation":"In the new IcebergSink builder's hash-distribution path, when no equality fields are configured and the table is unpartitioned there is no key to hash on, so the sink logs this warning and falls back to 'none' distribution (returns the input unchanged).","triggerScenarios":"IcebergSink constructor calls distributeDataStreamByHashDistributionMode with empty equalityFieldIds and partitionSpec.isUnpartitioned() (e.g. WRITE_DISTRIBUTION_MODE=hash or DistributionMode.HASH requested).","commonSituations":"Append-only writes to an unpartitioned table with hash mode requested; users copying config from a partitioned table's job; defaulting to hash in streaming without a primary key.","solutions":["Use DistributionMode.NONE for unpartitioned append-only streams","Add partitions to the table so HASH can key by partition","Configure equality fields/primary key for upsert workloads","Drop the explicit hash distribution setting"],"exampleFix":"// before\nIcebergSink.forRowData(input).distributionMode(DistributionMode.HASH)...\n// after\nIcebergSink.forRowData(input).distributionMode(DistributionMode.NONE)...","handlingStrategy":"validation","validationCode":"if (mode == DistributionMode.HASH && equalityFieldIds.isEmpty() && table.spec().isUnpartitioned()) {\n  mode = DistributionMode.NONE; // avoid pointless hash request\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Match distribution mode to table shape (partitioned/PK vs append-only)","Validate sink configuration in a pre-flight check before env.execute","Avoid copying distribution settings between different tables"],"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"}