{"record":{"id":"342b6143c29409bf","repo":"apache/iceberg","slug":"main-branch-snapshot-changed-since-planning-expec-342b61","errorCode":null,"errorMessage":"Main branch snapshot changed since planning: expected {} but found: {}","messagePattern":"Main branch snapshot changed since planning: expected (.+?) but found: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java","lineNumber":176,"sourceCode":"\n    super.processWatermark(mark);\n  }\n\n  private void resolveAndWrite() throws IOException {\n    if (positionsByFile.isEmpty()) {\n      return;\n    }\n\n    table.refresh();\n\n    Snapshot mainSnapshot = table.snapshot(targetBranch);\n\n    // Fail fast if the main branch changed since planning, to avoid writing DV files that the\n    // committer would reject via validateFromSnapshot. The next cycle will reindex.\n    if (mainSnapshot != null\n        && planResult.mainSnapshotId() != null\n        && mainSnapshot.snapshotId() != planResult.mainSnapshotId()) {\n      throw new IllegalStateException(\n          \"Main branch snapshot changed since planning: expected \"\n              + planResult.mainSnapshotId()\n              + \" but found: \"\n              + mainSnapshot.snapshotId());\n    }\n\n    Map<String, DeleteFile> dvs = collectExistingDVs(mainSnapshot, positionsByFile.keySet());\n\n    // Fold staging DVs into the rewrite so the writer emits one DV per data file (V3 rule). Flink\n    // writes a staging DV only for a newly added data file, so it never collides with a distinct\n    // existing DV: on a separate target branch collectExistingDVs has not seen it yet; on a shared\n    // branch it IS that existing DV, so the put is idempotent.\n    for (DeleteFile sd : planResult.stagingDVFiles()) {\n      if (ContentFileUtil.isDV(sd) && sd.referencedDataFile() != null) {\n        dvs.put(sd.referencedDataFile(), sd);\n      }\n    }\n","sourceCodeStart":158,"sourceCodeEnd":194,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java#L158-L194","documentation":"EqualityConvertDVWriter.resolveAndWrite() fails fast when the main branch snapshot observed at runtime differs from the snapshot id the DV plan was built against (planResult.mainSnapshotId()). Writing deletion vectors against a stale plan could produce files the committer's validateFromSnapshot would reject, so the writer aborts and the next cycle reindexes.","triggerScenarios":"Between scan planning and watermark-triggered write, another commit advanced the main branch (e.g. a concurrent writer or a prior maintenance cycle committed), so mainSnapshot.snapshotId() != planResult.mainSnapshotId().","commonSituations":"Concurrent stream/batch writers committing to the same table during equality-delete conversion; overlapping maintenance jobs; manual Spark/Flink commits landing mid-cycle.","solutions":["Don't write to the main table concurrently with the conversion job, or run conversion on an isolated branch schedule","Let the next maintenance cycle reindex — this error is designed to be recovered automatically","Reduce the window by triggering processing sooner after planning (commit-triggered rather than watermark-delayed)","Check the table's commit history to identify the competing writer"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// before running conversion, ensure no concurrent commits to main branch\nSnapshot current = table.snapshot(mainBranch);\nif (plannedMainSnapshotId != current.snapshotId()) {\n  // re-plan the cycle before writing DVs\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Avoid concurrent writers to the main branch during conversion cycles","Schedule conversion during write-quiescent windows","Trigger processing immediately after planning to shrink the race window","Design the task to tolerate the automatic reindex on the next cycle"],"tags":["snapshot","concurrency","deletion-vector","flink"],"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"}