{"record":{"id":"310d60175aabd2c2","repo":"apache/iceberg","slug":"incoming-records-violate-the-writer-assumption-tha-310d60","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:\npartition '<path>' in spec <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:\npartition '<path>' in spec <spec>","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java","lineNumber":96,"sourceCode":"\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 =\n            String.format(\"partition '%s' in spec %s\", spec.partitionToPath(partition), spec);\n        throw new IllegalStateException(NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE + errorCtx);\n      }\n\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\n    currentWriter.write(row);\n  }\n\n  @Override\n  public void close() throws IOException {\n    if (!closed) {\n      closeCurrentWriter();\n      this.closed = true;\n    }\n  }\n","sourceCodeStart":78,"sourceCodeEnd":114,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java#L78-L114","documentation":"ClusteredWriter.write also enforces clustering by partition within a spec: once a partition's writer is closed, records for that same partition must not appear again. When a record's partition key matches an already-completed partition, it throws IllegalStateException naming the partition path and spec.","triggerScenarios":"Writing records to a clustered writer where partition P, then a different partition, then partition P again occur within the same spec — the writer for P was already closed so it cannot be reopened.","commonSituations":"Input sorted by spec but not fully by partition (e.g., partial sort, multiple upstream tasks each emitting overlapping partitions); key-reuse bugs where the comparator sees the partition as equal but subsequent writes revisit it.","solutions":["Fully sort/repartition records by partition within each spec before writing.","Switch to a fanout writer that keeps writers open for all partitions.","Batch records grouped by partition and flush each group to a dedicated ClusteredWriter instance."],"exampleFix":"// before\nrecords.forEach(clusteredWriter::write); // interleaved partitions\n// after\nrecords.stream()\n    .sorted(Comparator.comparing(r -> partitionKey(r)))\n    .forEach(clusteredWriter::write);","handlingStrategy":"validation","validationCode":"// ensure records are sorted by partition within each spec\nrecords = records.sorted(Comparator.comparing((T r) -> specIdOf(r)).thenComparing(r -> partitionKey(r)));","typeGuard":null,"tryCatchPattern":"try {\n  writer.write(record);\n} catch (IllegalStateException e) {\n  // partition reappeared after close: sort input or use fanout writer\n}","preventionTips":["Fully sort by partition within spec, not just by spec.","Use fanout writers for inputs with interleaved partitions.","Group records per partition into separate write batches when sorting is impractical."],"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"}