{"record":{"id":"07c5f8196c4d1513","repo":"apache/iceberg","slug":"staging-snapshot-s-on-branch-s-removes-data-fi","errorCode":null,"errorMessage":"Staging snapshot %s on branch '%s' removes data files; equality delete conversion does not support rewrites on the staging branch. Run compaction on the target branch instead.","messagePattern":"Staging snapshot (.+?) on branch '(.+?)' removes data files; equality delete conversion does not support rewrites on the staging branch\\. Run compaction on the target branch instead\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java","lineNumber":550,"sourceCode":"  /**\n   * Classifies the files added by {@code stagingSnapshot} into data files, eq delete files, and DV\n   * files. Throws if the snapshot:\n   *\n   * <ul>\n   *   <li>Removes data files (rewrites on the staging branch aren't supported).\n   *   <li>Contains V2 positional delete files (the converter expects a V3 staging branch written by\n   *       Flink, which produces only deletion vectors for deletes).\n   *   <li>Contains an eq-delete file whose {@code equalityFieldIds()} doesn't match the\n   *       builder-configured set (silent wrong-key serialization otherwise).\n   * </ul>\n   */\n  private StagingInputs retrieveStagingFiles(Snapshot stagingSnapshot) {\n    SnapshotChanges changes = SnapshotChanges.builderFor(table).snapshot(stagingSnapshot).build();\n\n    // Rewrites on the staging branch would require rewriting the corresponding DVs against new\n    // data files on target. Not implemented; fail fast instead of silently dropping work.\n    if (changes.removedDataFiles().iterator().hasNext()) {\n      throw new IllegalStateException(\n          String.format(\n              \"Staging snapshot %s on branch '%s' removes data files; \"\n                  + \"equality delete conversion does not support rewrites on the staging branch. \"\n                  + \"Run compaction on the target branch instead.\",\n              stagingSnapshot.snapshotId(), stagingBranch));\n    }\n\n    List<DataFile> newDataFiles = Lists.newArrayList();\n    List<DeleteFile> stagingDVFiles = Lists.newArrayList();\n    List<DeleteFile> eqDeleteFiles = Lists.newArrayList();\n\n    for (DataFile dataFile : changes.addedDataFiles()) {\n      newDataFiles.add(dataFile);\n    }\n\n    for (DeleteFile deleteFile : changes.addedDeleteFiles()) {\n      if (deleteFile.content() == FileContent.EQUALITY_DELETES) {\n        Set<Integer> deleteFieldIds = Sets.newHashSet(deleteFile.equalityFieldIds());","sourceCodeStart":532,"sourceCodeEnd":568,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java#L532-L568","documentation":"The equality-delete conversion planner detected that the staging-branch snapshot removed (rewrote away) data files. Compaction-style rewrites on the staging branch would require rewriting the corresponding DVs against new data files on the target, which is not implemented, so the planner fails fast rather than silently dropping those deletes.","triggerScenarios":"A snapshot on the staging branch was produced by a rewrite/compaction (replace/delete of data files) instead of only appends plus DV writes, and EqualityConvertPlanner.inputs() -> retrieveStagingFiles() inspects SnapshotChanges.removedDataFiles().","commonSituations":"Someone ran compaction (rewrite_data_files) on the staging branch; a maintenance configuration points the planner at a branch that receives regular compaction; copied snapshots include rewrite commits.","solutions":["Run compaction on the target branch, not the staging branch, as the message advises.","Keep the staging branch reserved for DV-writing rewrites only (equality delete conversion commits).","Re-create/rebase the staging branch from a snapshot that contains no rewrite commits and re-run the conversion.","If compaction on staging is required in your pipeline, disable it or partition it out from the conversion branch."],"exampleFix":"// before\nSparkActions.get(spark).rewriteDataFiles(table).onBranch(\"staging\").execute();\n// after\nSparkActions.get(spark).rewriteDataFiles(table).onBranch(\"main\").execute();","handlingStrategy":"validation","validationCode":"// before running conversion, assert the staging branch has no rewrite commits\nSnapshot staging = table.snapshot(branch);\nboolean hasRewrites = SnapshotChanges.builderFor(table).snapshot(staging).build()\n    .removedDataFiles().iterator().hasNext();\nif (hasRewrites) throw new IllegalStateException(\"Compaction detected on staging branch\");","typeGuard":"boolean stagingFreeOfRewrites(Table table, String branch) {\n  Snapshot s = table.snapshot(branch);\n  return s == null || !SnapshotChanges.builderFor(table).snapshot(s).build()\n      .removedDataFiles().iterator().hasNext();\n}","tryCatchPattern":"try {\n  planner.inputs();\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"does not support rewrites\")) {\n    moveCompactionToTargetBranch(); rerun();\n  } else throw e;\n}","preventionTips":["Configure compaction jobs to run on the target/main branch only.","Keep the staging branch dedicated to DV-writing maintenance commits.","Audit branch-level scheduled jobs (rewrite_data_files targets) before enabling conversion.","Gate conversion on a pre-flight check of SnapshotChanges for the staging snapshot."],"tags":["flink","compaction","branch"],"backgroundTag":"unsupported-operation","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"}