{"record":{"id":"d61c2ad072d571d5","repo":"apache/iceberg","slug":"incoming-records-violate-the-writer-assumption-tha","errorCode":null,"errorMessage":"Incoming records violate the writer assumption that records are clustered by spec and by partition within each spec. Either cluster the incoming records or switch to fanout writers.\nEncountered records that belong to already closed files:\nspec <spec>","messagePattern":"Incoming records violate the writer assumption that records are clustered by spec and by partition within each spec\\. Either cluster the incoming records or switch to fanout writers\\.\nEncountered records that belong to already closed files:\nspec <spec>","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java","lineNumber":76,"sourceCode":"\n  protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);\n\n  protected abstract void addResult(R result);\n\n  protected abstract R aggregatedResult();\n\n  @Override\n  public void write(T row, PartitionSpec spec, StructLike partition) {\n    if (!spec.equals(currentSpec)) {\n      if (currentSpec != null) {\n        closeCurrentWriter();\n        completedSpecIds.add(currentSpec.specId());\n        completedPartitions.clear();\n      }\n\n      if (completedSpecIds.contains(spec.specId())) {\n        String errorCtx = String.format(\"spec %s\", spec);\n        throw new IllegalStateException(NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE + errorCtx);\n      }\n\n      StructType partitionType = spec.partitionType();\n\n      this.currentSpec = spec;\n      this.partitionComparator = Comparators.forType(partitionType);\n      this.completedPartitions = StructLikeSet.create(partitionType);\n      // copy the partition key as the key object may be reused\n      this.currentPartition = StructLikeUtil.copy(partition);\n      this.currentWriter = newWriter(currentSpec, currentPartition);\n\n    } else if (partition != currentPartition\n        && partitionComparator.compare(partition, currentPartition) != 0) {\n      closeCurrentWriter();\n      completedPartitions.add(currentPartition);\n\n      if (completedPartitions.contains(partition)) {\n        String errorCtx =","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java#L58-L94","documentation":"ClusteredWriter.write enforces that incoming records are clustered by partition spec: once a spec has been completed (its writer closed), records for that spec must not reappear. When a record's specId belongs to an already-closed spec, it throws IllegalStateException describing the offending spec.","triggerScenarios":"Writing records to a clustered (non-fanout) writer where spec A, then spec B, then spec A again occur in the input stream, causing the writer for spec A to be reopened after it was closed.","commonSituations":"Unsorted/unclustered input from joins or multiple table sources with different partition specs feeding one writer; using ClusteredWriter (e.g., ClusteredDataWriter) where records mix specs intermittently instead of using a fanout writer.","solutions":["Sort or repartition records by spec and partition before writing (e.g., a distributed sort by spec id then partition).","Switch to a fanout writer (e.g., FanoutDataWriter/FanoutPositionDeleteWriter) which keeps one open writer per spec/partition.","Split the input into separate write tasks per spec so each clustered writer sees only one spec."],"exampleFix":"// before\nClusteredDataWriter<T> writer = new ClusteredDataWriter<>(...); // assumes clustered input\nrecords.forEach(writer::write);\n// after\nFanoutDataWriter<T> writer = new FanoutDataWriter<>(...); // tolerates interleaved specs\nrecords.forEach(writer::write);","handlingStrategy":"validation","validationCode":"// ensure input is clustered by spec id before writing\nrecords = records.sorted(Comparator.comparing(r -> specIdOf(r)));","typeGuard":null,"tryCatchPattern":"try {\n  writer.write(record);\n} catch (IllegalStateException e) {\n  // records not clustered: switch to fanout writer or sort input\n}","preventionTips":["Sort or repartition by spec id before using any ClusteredWriter.","Prefer FanoutDataWriter when input ordering cannot be guaranteed.","Keep records for one spec grouped together in each write task."],"tags":["writer","clustering","partitioning","invalid-state"],"backgroundTag":"invalid-state-transition","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"}