{"record":{"id":"1365106dc86aed68","repo":"apache/iceberg","slug":"main-branch-snapshot-changed-since-planning-expec","errorCode":null,"errorMessage":"Main branch snapshot changed since planning: expected <planResult.mainSnapshotId()> but found: <mainSnapshot.snapshotId()>","messagePattern":"Main branch snapshot changed since planning: expected <planResult\\.mainSnapshotId\\(\\)> but found: <mainSnapshot\\.snapshotId\\(\\)>","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java#L158-L194","documentation":"Thrown by the Flink maintenance equality-delete-to-DV converter when the main table branch's current snapshot no longer matches the snapshot that was used at planning time. The operator fails fast because the committer's validateFromSnapshot would reject any DV files written against the stale plan. The next maintenance cycle will re-plan and re-index.","triggerScenarios":"A concurrent commit (e.g. an ingest job or another maintenance task) advanced the main branch between the planner's snapshot capture (planResult.mainSnapshotId()) and the writer's check of the live mainSnapshot at processWatermark time.","commonSituations":"Running table maintenance concurrently with streaming ingestion; two maintenance jobs racing on the same table; long backpressure/watermark delays letting the main branch move before the writer fires.","solutions":["Do not write to the main branch concurrently with this maintenance job; ensure ingestion pauses or commits are serialized with the maintenance cycle.","Simply re-run the maintenance cycle - the error is intentionally fail-fast and the next cycle re-plans against the new snapshot.","Use a staging branch for the rewrite so main-branch commits during processing do not invalidate the plan.","Reduce the window between planning and writing (fewer queued records, faster checkpoints) to lower race likelihood."],"exampleFix":"// before: maintenance and ingestion both commit to main concurrently\n// after: schedule the rewrite via the Iceberg rewrite API which plans and commits on a staging branch, or pause the sink during maintenance","handlingStrategy":"retry","validationCode":"// before running/committing the cycle\nSnapshot main = table.currentSnapshot();\nif (!main.snapshotId().equals(planResult.mainSnapshotId())) {\n  LOG.warn(\"Main moved since planning; restarting cycle\"); // skip write, re-plan\n}","typeGuard":"boolean planStillValid(Table table, Long plannedSnapshotId) {\n  return plannedSnapshotId != null\n      && table.currentSnapshot() != null\n      && table.currentSnapshot().snapshotId() == plannedSnapshotId;\n}","tryCatchPattern":"try {\n  writer.processWatermark(watermark);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"changed since planning\")) {\n    // expected under concurrency: restart the cycle, no data loss\n    restartCycle();\n  } else throw e;\n}","preventionTips":["Do not commit to the main branch while the conversion cycle runs; serialize commits with maintenance.","Use a staging branch for rewrites so main commits cannot invalidate the plan.","Keep cycles short to shrink the planning-to-write window.","Monitor main-branch commit rates and schedule maintenance in quiet windows."],"tags":["flink","snapshot-conflict","concurrency"],"backgroundTag":"snapshot-changed-since-planning","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"}