kumarpritam863 opened a new pull request, #18583: URL: https://github.com/apache/iceberg/pull/18583
`TableUpdater.findOrCreateBranch` (Flink dynamic sink) no longer creates the `main` branch. For a table that has no snapshot yet, creating `main` committed an empty snapshot, and every writer subtask that reached the new table before the first data commit added another one. ### Why `findOrCreateBranch` calls `table.manageSnapshots().createBranch(branch).commit()` whenever the metadata cache does not list the branch. A new table has no `main` ref, so the cache never lists it before the first commit, and `SnapshotManager.createBranch(String)` on a table without a current snapshot does `transaction.newFastAppend().toBranch(name).commit()` (core, `SnapshotManager.java`, "Create an empty snapshot for the branch"). Concurrent subtasks each pass the "ref does not exist" check and each commit their own empty snapshot. A job with parallelism 5 left up to five empty `append` snapshots (`total-records=0`, no `flink.job-id`) ahead of the first data snapshot. ### Change Return early from `findOrCreateBranch` for `SnapshotRef.MAIN_BRANCH`. The first commit to a table creates `main`, and `DynamicCommitter` already handles a branch with no snapshot (`latestSnapshot != null ? ancestorsOf(...) : List.of()`). Other branches are created as before. The change and its test are identical in the three Flink version directories: `v1.20`, `v2.2`, `v2.3`. ### Why only `main` For another branch, `createBranch(name)` on a table that has a snapshot only adds a ref at the table's head and commits nothing, so there is no empty snapshot to avoid, and skipping it would make the branch start empty instead of from the head. On a table without a snapshot it does commit an empty snapshot on that branch, but that snapshot currently also gives the branch its own history: without it, the first commit to the branch is validated by `DynamicCommitter`'s `MaxCommittedCheckpointIdValidator` against the ancestry that `SnapshotUtil.latestSnapshot(metadata, branch)` returns for a branch that does not exist, which is the head of `main`. If `main` has already committed the same job and operator for that checkpoint, the commit to the new branch is treated as a duplicate and skipped. I reproduced this by also skipping creation for other branches: `TestDynamicIcebergSink#testMultipleBranches` then lost the commit to `branch1` in about one run in four. For `main` a missing ref h as no ancestry to fall back to, so the early return is safe. Handling the other branches would need the committer to validate against the branch's own ancestry first, which is a separate change. ### Testing - New `TestTableUpdater#testMainBranchIsNotCreatedAsAnEmptySnapshot`: updating a table that does not exist, twice, leaves it with no snapshots and no refs. It fails on current `main` (one empty `append` snapshot) and passes with the change. - `TestTableUpdater` passes in all three versions (15 tests each); `spotlessApply` is clean. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
