laskoviymishka commented on code in PR #1283: URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3481847888
########## table/rewrite_manifests.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + "strconv" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// Snapshot summary keys for a manifest rewrite. +const ( + manifestsCreatedKey = "manifests-created" + manifestsReplacedKey = "manifests-replaced" + manifestsKeptKey = "manifests-kept" + entriesProcessedKey = "entries-processed" +) + +// RewriteManifestsResult reports the manifests changed by a rewrite. +type RewriteManifestsResult struct { + // RewrittenManifests are the old manifests that were replaced. + RewrittenManifests []iceberg.ManifestFile + // AddedManifests are the new manifests written in their place. + AddedManifests []iceberg.ManifestFile +} + +type rewriteManifestsCfg struct { + targetSizeBytes int64 + specID *int + predicate func(iceberg.ManifestFile) bool +} + +// RewriteManifestsOpt configures [Transaction.RewriteManifests]. +type RewriteManifestsOpt func(*rewriteManifestsCfg) + +// WithManifestTargetSize overrides the target manifest size in bytes. +// The default comes from the commit.manifest.target-size-bytes property. +func WithManifestTargetSize(size int64) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { + if size > 0 { + c.targetSizeBytes = size + } + } +} + +// WithRewriteSpecID restricts the rewrite to manifests of one partition spec. +func WithRewriteSpecID(id int) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.specID = &id } +} + +// WithRewriteManifestPredicate only rewrites manifests for which pred is true. +// Manifests that don't match are left untouched. +func WithRewriteManifestPredicate(pred func(iceberg.ManifestFile) bool) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.predicate = pred } +} + +// rewriteManifests is a producer that merges small data manifests into +// fewer, target-sized ones, committed as a metadata-only REPLACE snapshot. +type rewriteManifests struct { + base *snapshotProducer + cfg rewriteManifestsCfg + + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile + + // result is the value returned to the caller. record() rewrites its + // fields on every pass, including OCC retries, so the pointer the caller + // holds reflects the manifests actually committed, not attempt 0's. + result *RewriteManifestsResult +} + +func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer { + prod := createSnapshotProducer(OpReplace, txn, fs, nil, props) + prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result: &RewriteManifestsResult{}} + // A rewrite re-expresses inherited manifests, so OCC retries must + // re-derive from the fresh parent rather than carry these forward. + prod.rebuildMode = true + + return prod +} + +func (r *rewriteManifests) existingManifests() ([]iceberg.ManifestFile, error) { + snap := r.base.txn.meta.currentSnapshot() + if snap == nil { + return nil, nil + } + + return snap.Manifests(r.base.io) +} + +func (r *rewriteManifests) deletedEntries(context.Context) ([]iceberg.ManifestEntry, error) { + return nil, nil +} + +func (r *rewriteManifests) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + var toRewrite, kept []iceberg.ManifestFile + for _, m := range manifests { + if r.eligible(m) { + toRewrite = append(toRewrite, m) + } else { + kept = append(kept, m) + } + } + + mgr := manifestMergeManager{ + targetSizeBytes: r.cfg.targetSizeBytes, + minCountToMerge: 1, // force a merge regardless of count + mergeEnabled: true, // explicit op ignores commit.manifest-merge.enabled + snap: r.base, + } + merged, err := mgr.mergeManifests(toRewrite) + if err != nil { + return nil, err + } + + if err := validateRewriteFileCounts(toRewrite, merged); err != nil { + return nil, err + } + + // Record on every pass, not just the first. An OCC retry re-runs this + // against the fresh parent and writes a different merged set; the last + // pass is the one that commits, so its counts are the ones that must win. + r.record(toRewrite, merged, kept) + + return slices.Concat(merged, kept), nil +} + +// eligible reports whether m is a data manifest selected for rewrite. +func (r *rewriteManifests) eligible(m iceberg.ManifestFile) bool { + if m.ManifestContent() != iceberg.ManifestContentData { + return false + } + if r.cfg.specID != nil && int(m.PartitionSpecID()) != *r.cfg.specID { + return false + } + if r.cfg.predicate != nil && !r.cfg.predicate(m) { + return false + } + + return true +} + +func (r *rewriteManifests) record(toRewrite, merged, kept []iceberg.ManifestFile) { + // A prior pass (a superseded OCC attempt) wrote its own merged manifests + // that the catalog never referenced. Carry their paths into the + // producer's superseded set so commit() can delete them on success; + // otherwise every retry leaks a full set of merged .avro files. + for _, m := range r.added { + r.base.supersededManifests = append(r.base.supersededManifests, m.FilePath()) + } + r.added, r.rewritten = nil, nil + + inPaths := make(map[string]struct{}, len(toRewrite)) + for _, m := range toRewrite { + inPaths[m.FilePath()] = struct{}{} + } + outPaths := make(map[string]struct{}, len(merged)) + for _, m := range merged { + outPaths[m.FilePath()] = struct{}{} + } + + // Bins of a single manifest pass through unchanged; only the manifests + // that actually appear on one side and not the other are added/replaced. + // This drives the returned RewriteManifestsResult. + for _, m := range merged { + if _, ok := inPaths[m.FilePath()]; !ok { + r.added = append(r.added, m) + } + } + for _, m := range toRewrite { + if _, ok := outPaths[m.FilePath()]; !ok { + r.rewritten = append(r.rewritten, m) + } + } + + // Publish into the shared result the caller holds so a retry's counts win. + r.result.RewrittenManifests = r.rewritten + r.result.AddedManifests = r.added + + // The summary follows Java's Appendix-F accounting over the full eligible + // set, pass-throughs included, so it reads the same as a Java-written + // snapshot — distinct from the added/rewritten diff above. + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(merged)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(toRewrite)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(kept)) + if entries := manifestActiveFiles(toRewrite); entries >= 0 { + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(entries, 10) + } else { + delete(r.base.snapshotProps, entriesProcessedKey) + } +} + +func (r *rewriteManifests) validate(*conflictContext) error { return nil } +func (r *rewriteManifests) needsValidation() bool { return false } + +// manifestActiveFiles sums added + existing data files, or -1 if any manifest +// reports an unknown count. +func manifestActiveFiles(manifests []iceberg.ManifestFile) int64 { + var total int64 + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + if added < 0 || existing < 0 { + return -1 + } + total += int64(added) + int64(existing) + } + + return total +} + +// validateRewriteFileCounts guards against dropping or duplicating data files: +// the rewritten manifests must hold the same active files as their inputs. +func validateRewriteFileCounts(before, after []iceberg.ManifestFile) error { + in, out := manifestActiveFiles(before), manifestActiveFiles(after) + if in < 0 || out < 0 { Review Comment: This guard is doing important work — it's the main thing protecting against a merge accidentally dropping or duplicating data files. The one thing I'd love to tighten: it returns nil the moment a manifest reports an unknown count, and V1 tables don't populate the count fields at all, so on those tables the check quietly doesn't run — which is a shame, because they're exactly the ones where I'd most want it. Could we fall back to reading the live entries and counting them directly when the header counts are missing? If that feels like too much for this PR, even a loud log plus a godoc note that the guard is a no-op on V1 would keep us honest about what it does and doesn't cover. wdyt? ########## table/table.go: ########## @@ -530,6 +530,15 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen return nil, err } + // The commit succeeded. Inner data manifests written by superseded retry + // attempts (a rewrite re-merges everything on each retry) are now orphaned + // objects; the committed snapshot references only the final attempt's. + for _, u := range updates { + if su, ok := u.(*addSnapshotUpdate); ok && su.supersededManifests != nil { + orphanedManifests = append(orphanedManifests, *su.supersededManifests...) Review Comment: This handles the happy path well. The only gap I see is the error path: if `CommitTable` returns something other than `ErrCommitFailed` (a 5xx after a few retries, say), cleanup is skipped, and the manifests written by the earlier attempts — which nothing ever referenced — become orphans no one cleans up. Definitely fine as a follow-up rather than a blocker. The nice thing is those pre-failure superseded manifests are always safe to delete, unlike the winning attempt's, so cleaning them unconditionally on the error path (or just documenting the leak as a known edge) would close it. ########## table/snapshot_producers.go: ########## @@ -524,6 +524,19 @@ type snapshotProducer struct { deletedFiles map[string]iceberg.DataFile deletedDeleteFiles map[string]iceberg.DataFile snapshotProps iceberg.Properties + + // rebuildMode re-derives manifests from the fresh parent on an OCC + // retry rather than carrying forward this producer's manifests. It is a + // lifecycle knob consumed by commit(), not part of the producer + // contract; newRewriteManifestsProducer sets it. + rebuildMode bool Review Comment: Picking up the question from your PR description about whether there's a cleaner way to express this — I think your earlier instinct toward `rebuildFromInheritedOnly() bool` on `producerImpl` was the right one. As a field on the shared `snapshotProducer`, every other producer ends up carrying `rebuildMode` and `supersededManifests` zero-valued, and `commit()` grows an `if sp.rebuildMode` branch that (as the comment itself notes) isn't really part of the producer contract. Putting it back behind an interface method, and scoping `supersededManifests` to the `rewriteManifests` type — surfaced to `doCommit` through the `addSnapshotUpdate` you're already threading — would keep the base struct generic. Worth settling before more producers grow on top of it. wdyt? ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + "strconv" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// Snapshot summary keys for a manifest rewrite. +const ( + manifestsCreatedKey = "manifests-created" + manifestsReplacedKey = "manifests-replaced" + manifestsKeptKey = "manifests-kept" + entriesProcessedKey = "entries-processed" +) + +// RewriteManifestsResult reports the manifests changed by a rewrite. +type RewriteManifestsResult struct { + // RewrittenManifests are the old manifests that were replaced. + RewrittenManifests []iceberg.ManifestFile + // AddedManifests are the new manifests written in their place. + AddedManifests []iceberg.ManifestFile +} + +type rewriteManifestsCfg struct { + targetSizeBytes int64 + specID *int + predicate func(iceberg.ManifestFile) bool +} + +// RewriteManifestsOpt configures [Transaction.RewriteManifests]. +type RewriteManifestsOpt func(*rewriteManifestsCfg) + +// WithManifestTargetSize overrides the target manifest size in bytes. +// The default comes from the commit.manifest.target-size-bytes property. +func WithManifestTargetSize(size int64) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { + if size > 0 { + c.targetSizeBytes = size + } + } +} + +// WithRewriteSpecID restricts the rewrite to manifests of one partition spec. +func WithRewriteSpecID(id int) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.specID = &id } +} + +// WithRewriteManifestPredicate only rewrites manifests for which pred is true. +// Manifests that don't match are left untouched. +func WithRewriteManifestPredicate(pred func(iceberg.ManifestFile) bool) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.predicate = pred } +} + +// rewriteManifests is a producer that merges small data manifests into +// fewer, target-sized ones, committed as a metadata-only REPLACE snapshot. +type rewriteManifests struct { + base *snapshotProducer + cfg rewriteManifestsCfg + + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile + + // result is the value returned to the caller. record() rewrites its + // fields on every pass, including OCC retries, so the pointer the caller + // holds reflects the manifests actually committed, not attempt 0's. + result *RewriteManifestsResult +} + +func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer { + prod := createSnapshotProducer(OpReplace, txn, fs, nil, props) + prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result: &RewriteManifestsResult{}} + // A rewrite re-expresses inherited manifests, so OCC retries must + // re-derive from the fresh parent rather than carry these forward. + prod.rebuildMode = true + + return prod +} + +func (r *rewriteManifests) existingManifests() ([]iceberg.ManifestFile, error) { + snap := r.base.txn.meta.currentSnapshot() + if snap == nil { + return nil, nil + } + + return snap.Manifests(r.base.io) +} + +func (r *rewriteManifests) deletedEntries(context.Context) ([]iceberg.ManifestEntry, error) { + return nil, nil +} + +func (r *rewriteManifests) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + var toRewrite, kept []iceberg.ManifestFile + for _, m := range manifests { + if r.eligible(m) { + toRewrite = append(toRewrite, m) + } else { + kept = append(kept, m) + } + } + + mgr := manifestMergeManager{ + targetSizeBytes: r.cfg.targetSizeBytes, + minCountToMerge: 1, // force a merge regardless of count + mergeEnabled: true, // explicit op ignores commit.manifest-merge.enabled + snap: r.base, + } + merged, err := mgr.mergeManifests(toRewrite) + if err != nil { + return nil, err + } + + if err := validateRewriteFileCounts(toRewrite, merged); err != nil { + return nil, err + } + + // Record on every pass, not just the first. An OCC retry re-runs this + // against the fresh parent and writes a different merged set; the last + // pass is the one that commits, so its counts are the ones that must win. + r.record(toRewrite, merged, kept) + + return slices.Concat(merged, kept), nil +} + +// eligible reports whether m is a data manifest selected for rewrite. +func (r *rewriteManifests) eligible(m iceberg.ManifestFile) bool { + if m.ManifestContent() != iceberg.ManifestContentData { + return false + } + if r.cfg.specID != nil && int(m.PartitionSpecID()) != *r.cfg.specID { + return false + } + if r.cfg.predicate != nil && !r.cfg.predicate(m) { + return false + } + + return true +} + +func (r *rewriteManifests) record(toRewrite, merged, kept []iceberg.ManifestFile) { + // A prior pass (a superseded OCC attempt) wrote its own merged manifests + // that the catalog never referenced. Carry their paths into the + // producer's superseded set so commit() can delete them on success; + // otherwise every retry leaks a full set of merged .avro files. + for _, m := range r.added { + r.base.supersededManifests = append(r.base.supersededManifests, m.FilePath()) + } + r.added, r.rewritten = nil, nil + + inPaths := make(map[string]struct{}, len(toRewrite)) + for _, m := range toRewrite { + inPaths[m.FilePath()] = struct{}{} + } + outPaths := make(map[string]struct{}, len(merged)) + for _, m := range merged { + outPaths[m.FilePath()] = struct{}{} + } + + // Bins of a single manifest pass through unchanged; only the manifests + // that actually appear on one side and not the other are added/replaced. + // This drives the returned RewriteManifestsResult. + for _, m := range merged { + if _, ok := inPaths[m.FilePath()]; !ok { + r.added = append(r.added, m) + } + } + for _, m := range toRewrite { + if _, ok := outPaths[m.FilePath()]; !ok { + r.rewritten = append(r.rewritten, m) + } + } + + // Publish into the shared result the caller holds so a retry's counts win. + r.result.RewrittenManifests = r.rewritten + r.result.AddedManifests = r.added + + // The summary follows Java's Appendix-F accounting over the full eligible + // set, pass-throughs included, so it reads the same as a Java-written + // snapshot — distinct from the added/rewritten diff above. + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(merged)) Review Comment: The comment mentions matching Java's Appendix-F accounting — I think it's close but drifts a little once single-manifest pass-throughs are in play. Java counts created/replaced over the manifests actually written and consumed by a merge, and entries-processed over entries physically read, whereas here a bin of one manifest passes through untouched but still lands in `len(merged)` and `len(toRewrite)`. Concretely, with 10 eligible manifests where 5 are already above target and 5 merge into 1, this reports created=6/replaced=10 while Java reports created=1/replaced=5. The good news is you already compute the real added/rewritten diff for the result — driving the summary off `len(r.added)`/`len(r.rewritten)` (and counting entries over `r.rewritten`) would make a Go-written snapshot read the same as a Java one. wdyt? ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + "strconv" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// Snapshot summary keys for a manifest rewrite. +const ( + manifestsCreatedKey = "manifests-created" + manifestsReplacedKey = "manifests-replaced" + manifestsKeptKey = "manifests-kept" + entriesProcessedKey = "entries-processed" +) + +// RewriteManifestsResult reports the manifests changed by a rewrite. +type RewriteManifestsResult struct { + // RewrittenManifests are the old manifests that were replaced. + RewrittenManifests []iceberg.ManifestFile + // AddedManifests are the new manifests written in their place. + AddedManifests []iceberg.ManifestFile +} + +type rewriteManifestsCfg struct { + targetSizeBytes int64 + specID *int + predicate func(iceberg.ManifestFile) bool +} + +// RewriteManifestsOpt configures [Transaction.RewriteManifests]. +type RewriteManifestsOpt func(*rewriteManifestsCfg) + +// WithManifestTargetSize overrides the target manifest size in bytes. +// The default comes from the commit.manifest.target-size-bytes property. +func WithManifestTargetSize(size int64) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { + if size > 0 { + c.targetSizeBytes = size + } + } +} + +// WithRewriteSpecID restricts the rewrite to manifests of one partition spec. +func WithRewriteSpecID(id int) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.specID = &id } +} + +// WithRewriteManifestPredicate only rewrites manifests for which pred is true. +// Manifests that don't match are left untouched. +func WithRewriteManifestPredicate(pred func(iceberg.ManifestFile) bool) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.predicate = pred } +} + +// rewriteManifests is a producer that merges small data manifests into +// fewer, target-sized ones, committed as a metadata-only REPLACE snapshot. +type rewriteManifests struct { + base *snapshotProducer + cfg rewriteManifestsCfg + + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile + + // result is the value returned to the caller. record() rewrites its + // fields on every pass, including OCC retries, so the pointer the caller + // holds reflects the manifests actually committed, not attempt 0's. + result *RewriteManifestsResult +} + +func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer { + prod := createSnapshotProducer(OpReplace, txn, fs, nil, props) + prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result: &RewriteManifestsResult{}} + // A rewrite re-expresses inherited manifests, so OCC retries must + // re-derive from the fresh parent rather than carry these forward. + prod.rebuildMode = true + + return prod +} + +func (r *rewriteManifests) existingManifests() ([]iceberg.ManifestFile, error) { + snap := r.base.txn.meta.currentSnapshot() + if snap == nil { + return nil, nil + } + + return snap.Manifests(r.base.io) +} + +func (r *rewriteManifests) deletedEntries(context.Context) ([]iceberg.ManifestEntry, error) { + return nil, nil +} + +func (r *rewriteManifests) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + var toRewrite, kept []iceberg.ManifestFile + for _, m := range manifests { + if r.eligible(m) { + toRewrite = append(toRewrite, m) + } else { + kept = append(kept, m) + } + } + + mgr := manifestMergeManager{ + targetSizeBytes: r.cfg.targetSizeBytes, + minCountToMerge: 1, // force a merge regardless of count + mergeEnabled: true, // explicit op ignores commit.manifest-merge.enabled + snap: r.base, + } + merged, err := mgr.mergeManifests(toRewrite) + if err != nil { + return nil, err + } + + if err := validateRewriteFileCounts(toRewrite, merged); err != nil { + return nil, err + } + + // Record on every pass, not just the first. An OCC retry re-runs this + // against the fresh parent and writes a different merged set; the last + // pass is the one that commits, so its counts are the ones that must win. + r.record(toRewrite, merged, kept) + + return slices.Concat(merged, kept), nil +} + +// eligible reports whether m is a data manifest selected for rewrite. +func (r *rewriteManifests) eligible(m iceberg.ManifestFile) bool { + if m.ManifestContent() != iceberg.ManifestContentData { + return false + } + if r.cfg.specID != nil && int(m.PartitionSpecID()) != *r.cfg.specID { + return false + } + if r.cfg.predicate != nil && !r.cfg.predicate(m) { + return false + } + + return true +} + +func (r *rewriteManifests) record(toRewrite, merged, kept []iceberg.ManifestFile) { + // A prior pass (a superseded OCC attempt) wrote its own merged manifests + // that the catalog never referenced. Carry their paths into the + // producer's superseded set so commit() can delete them on success; + // otherwise every retry leaks a full set of merged .avro files. + for _, m := range r.added { + r.base.supersededManifests = append(r.base.supersededManifests, m.FilePath()) + } + r.added, r.rewritten = nil, nil + + inPaths := make(map[string]struct{}, len(toRewrite)) + for _, m := range toRewrite { + inPaths[m.FilePath()] = struct{}{} + } + outPaths := make(map[string]struct{}, len(merged)) + for _, m := range merged { + outPaths[m.FilePath()] = struct{}{} + } + + // Bins of a single manifest pass through unchanged; only the manifests + // that actually appear on one side and not the other are added/replaced. + // This drives the returned RewriteManifestsResult. + for _, m := range merged { + if _, ok := inPaths[m.FilePath()]; !ok { + r.added = append(r.added, m) + } + } + for _, m := range toRewrite { + if _, ok := outPaths[m.FilePath()]; !ok { + r.rewritten = append(r.rewritten, m) + } + } + + // Publish into the shared result the caller holds so a retry's counts win. + r.result.RewrittenManifests = r.rewritten + r.result.AddedManifests = r.added + + // The summary follows Java's Appendix-F accounting over the full eligible + // set, pass-throughs included, so it reads the same as a Java-written + // snapshot — distinct from the added/rewritten diff above. + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(merged)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(toRewrite)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(kept)) + if entries := manifestActiveFiles(toRewrite); entries >= 0 { + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(entries, 10) + } else { + delete(r.base.snapshotProps, entriesProcessedKey) + } +} + +func (r *rewriteManifests) validate(*conflictContext) error { return nil } +func (r *rewriteManifests) needsValidation() bool { return false } + +// manifestActiveFiles sums added + existing data files, or -1 if any manifest +// reports an unknown count. +func manifestActiveFiles(manifests []iceberg.ManifestFile) int64 { + var total int64 + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + if added < 0 || existing < 0 { + return -1 + } + total += int64(added) + int64(existing) + } + + return total +} + +// validateRewriteFileCounts guards against dropping or duplicating data files: +// the rewritten manifests must hold the same active files as their inputs. +func validateRewriteFileCounts(before, after []iceberg.ManifestFile) error { + in, out := manifestActiveFiles(before), manifestActiveFiles(after) + if in < 0 || out < 0 { + return nil // counts unknown; can't validate + } + if in != out { + return fmt.Errorf("rewrite manifests changed active file count: %d before, %d after", in, out) + } + + return nil +} + +// RewriteManifests merges small data manifests in the current snapshot into +// fewer, target-sized ones and stages the result as a REPLACE snapshot. It +// rewrites metadata only; no data files are read or written. Delete manifests +// are left untouched. +// +// An empty result (no AddedManifests and no RewrittenManifests) means there +// was nothing to do — either the table has no current snapshot or the +// manifests are already optimal. Callers should treat that as a no-op and +// skip the commit rather than staging an empty REPLACE snapshot. +func (t *Transaction) RewriteManifests(ctx context.Context, opts ...RewriteManifestsOpt) (*RewriteManifestsResult, error) { + if t.meta.currentSnapshot() == nil { + return &RewriteManifestsResult{}, nil + } + + cfg := rewriteManifestsCfg{ + targetSizeBytes: int64(t.meta.props.GetInt(ManifestTargetSizeBytesKey, ManifestTargetSizeBytesDefault)), + } + for _, o := range opts { + o(&cfg) + } + + if cfg.specID != nil { + if _, err := t.meta.GetSpecByID(*cfg.specID); err != nil { Review Comment: Tiny one: `GetSpecByID`'s error gets dropped and replaced with a fresh string, so the `ErrPartitionSpecNotFound` sentinel is lost and callers can't `errors.Is` it — which is why the test has to string-match on "unknown partition spec id 999". Wrapping it — `fmt.Errorf("cannot rewrite manifests: unknown partition spec id %d: %w", *cfg.specID, err)` — and switching the test to `errors.Is` would make the failure programmatically distinguishable. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + "strconv" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// Snapshot summary keys for a manifest rewrite. +const ( + manifestsCreatedKey = "manifests-created" + manifestsReplacedKey = "manifests-replaced" + manifestsKeptKey = "manifests-kept" + entriesProcessedKey = "entries-processed" +) + +// RewriteManifestsResult reports the manifests changed by a rewrite. +type RewriteManifestsResult struct { + // RewrittenManifests are the old manifests that were replaced. + RewrittenManifests []iceberg.ManifestFile + // AddedManifests are the new manifests written in their place. + AddedManifests []iceberg.ManifestFile +} + +type rewriteManifestsCfg struct { + targetSizeBytes int64 + specID *int + predicate func(iceberg.ManifestFile) bool +} + +// RewriteManifestsOpt configures [Transaction.RewriteManifests]. +type RewriteManifestsOpt func(*rewriteManifestsCfg) + +// WithManifestTargetSize overrides the target manifest size in bytes. +// The default comes from the commit.manifest.target-size-bytes property. +func WithManifestTargetSize(size int64) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { + if size > 0 { + c.targetSizeBytes = size + } + } +} + +// WithRewriteSpecID restricts the rewrite to manifests of one partition spec. +func WithRewriteSpecID(id int) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.specID = &id } +} + +// WithRewriteManifestPredicate only rewrites manifests for which pred is true. +// Manifests that don't match are left untouched. +func WithRewriteManifestPredicate(pred func(iceberg.ManifestFile) bool) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.predicate = pred } +} + +// rewriteManifests is a producer that merges small data manifests into +// fewer, target-sized ones, committed as a metadata-only REPLACE snapshot. +type rewriteManifests struct { + base *snapshotProducer + cfg rewriteManifestsCfg + + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile + + // result is the value returned to the caller. record() rewrites its + // fields on every pass, including OCC retries, so the pointer the caller + // holds reflects the manifests actually committed, not attempt 0's. + result *RewriteManifestsResult +} + +func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer { + prod := createSnapshotProducer(OpReplace, txn, fs, nil, props) + prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result: &RewriteManifestsResult{}} + // A rewrite re-expresses inherited manifests, so OCC retries must + // re-derive from the fresh parent rather than carry these forward. + prod.rebuildMode = true + + return prod +} + +func (r *rewriteManifests) existingManifests() ([]iceberg.ManifestFile, error) { + snap := r.base.txn.meta.currentSnapshot() + if snap == nil { + return nil, nil + } + + return snap.Manifests(r.base.io) +} + +func (r *rewriteManifests) deletedEntries(context.Context) ([]iceberg.ManifestEntry, error) { + return nil, nil +} + +func (r *rewriteManifests) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + var toRewrite, kept []iceberg.ManifestFile + for _, m := range manifests { + if r.eligible(m) { + toRewrite = append(toRewrite, m) + } else { + kept = append(kept, m) + } + } + + mgr := manifestMergeManager{ + targetSizeBytes: r.cfg.targetSizeBytes, + minCountToMerge: 1, // force a merge regardless of count + mergeEnabled: true, // explicit op ignores commit.manifest-merge.enabled + snap: r.base, + } + merged, err := mgr.mergeManifests(toRewrite) + if err != nil { + return nil, err + } + + if err := validateRewriteFileCounts(toRewrite, merged); err != nil { + return nil, err + } + + // Record on every pass, not just the first. An OCC retry re-runs this + // against the fresh parent and writes a different merged set; the last + // pass is the one that commits, so its counts are the ones that must win. + r.record(toRewrite, merged, kept) + + return slices.Concat(merged, kept), nil +} + +// eligible reports whether m is a data manifest selected for rewrite. +func (r *rewriteManifests) eligible(m iceberg.ManifestFile) bool { + if m.ManifestContent() != iceberg.ManifestContentData { + return false + } + if r.cfg.specID != nil && int(m.PartitionSpecID()) != *r.cfg.specID { + return false + } + if r.cfg.predicate != nil && !r.cfg.predicate(m) { + return false + } + + return true +} + +func (r *rewriteManifests) record(toRewrite, merged, kept []iceberg.ManifestFile) { + // A prior pass (a superseded OCC attempt) wrote its own merged manifests + // that the catalog never referenced. Carry their paths into the + // producer's superseded set so commit() can delete them on success; + // otherwise every retry leaks a full set of merged .avro files. + for _, m := range r.added { + r.base.supersededManifests = append(r.base.supersededManifests, m.FilePath()) + } + r.added, r.rewritten = nil, nil + + inPaths := make(map[string]struct{}, len(toRewrite)) + for _, m := range toRewrite { + inPaths[m.FilePath()] = struct{}{} + } + outPaths := make(map[string]struct{}, len(merged)) + for _, m := range merged { + outPaths[m.FilePath()] = struct{}{} + } + + // Bins of a single manifest pass through unchanged; only the manifests + // that actually appear on one side and not the other are added/replaced. + // This drives the returned RewriteManifestsResult. + for _, m := range merged { + if _, ok := inPaths[m.FilePath()]; !ok { + r.added = append(r.added, m) + } + } + for _, m := range toRewrite { + if _, ok := outPaths[m.FilePath()]; !ok { + r.rewritten = append(r.rewritten, m) + } + } + + // Publish into the shared result the caller holds so a retry's counts win. + r.result.RewrittenManifests = r.rewritten + r.result.AddedManifests = r.added + + // The summary follows Java's Appendix-F accounting over the full eligible + // set, pass-throughs included, so it reads the same as a Java-written + // snapshot — distinct from the added/rewritten diff above. + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(merged)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(toRewrite)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(kept)) + if entries := manifestActiveFiles(toRewrite); entries >= 0 { + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(entries, 10) + } else { + delete(r.base.snapshotProps, entriesProcessedKey) + } +} + +func (r *rewriteManifests) validate(*conflictContext) error { return nil } +func (r *rewriteManifests) needsValidation() bool { return false } + +// manifestActiveFiles sums added + existing data files, or -1 if any manifest +// reports an unknown count. +func manifestActiveFiles(manifests []iceberg.ManifestFile) int64 { + var total int64 + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + if added < 0 || existing < 0 { + return -1 + } + total += int64(added) + int64(existing) + } + + return total +} + +// validateRewriteFileCounts guards against dropping or duplicating data files: +// the rewritten manifests must hold the same active files as their inputs. +func validateRewriteFileCounts(before, after []iceberg.ManifestFile) error { + in, out := manifestActiveFiles(before), manifestActiveFiles(after) + if in < 0 || out < 0 { + return nil // counts unknown; can't validate + } + if in != out { + return fmt.Errorf("rewrite manifests changed active file count: %d before, %d after", in, out) + } + + return nil +} + +// RewriteManifests merges small data manifests in the current snapshot into Review Comment: One small doc note: Java's `RewriteManifests` exposes `clusterBy(...)` to group by partition/sort key, and this bin-packs by size only. That's a totally reasonable subset for a first cut — I'd just call it out in this godoc (size-only today, partition/sort clustering a future extension) so the contract stays honest until a `WithClusterBy` lands. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + "strconv" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// Snapshot summary keys for a manifest rewrite. +const ( + manifestsCreatedKey = "manifests-created" + manifestsReplacedKey = "manifests-replaced" + manifestsKeptKey = "manifests-kept" + entriesProcessedKey = "entries-processed" +) + +// RewriteManifestsResult reports the manifests changed by a rewrite. +type RewriteManifestsResult struct { + // RewrittenManifests are the old manifests that were replaced. + RewrittenManifests []iceberg.ManifestFile + // AddedManifests are the new manifests written in their place. + AddedManifests []iceberg.ManifestFile +} + +type rewriteManifestsCfg struct { + targetSizeBytes int64 + specID *int + predicate func(iceberg.ManifestFile) bool +} + +// RewriteManifestsOpt configures [Transaction.RewriteManifests]. +type RewriteManifestsOpt func(*rewriteManifestsCfg) + +// WithManifestTargetSize overrides the target manifest size in bytes. +// The default comes from the commit.manifest.target-size-bytes property. +func WithManifestTargetSize(size int64) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { + if size > 0 { + c.targetSizeBytes = size + } + } +} + +// WithRewriteSpecID restricts the rewrite to manifests of one partition spec. +func WithRewriteSpecID(id int) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.specID = &id } +} + +// WithRewriteManifestPredicate only rewrites manifests for which pred is true. +// Manifests that don't match are left untouched. +func WithRewriteManifestPredicate(pred func(iceberg.ManifestFile) bool) RewriteManifestsOpt { + return func(c *rewriteManifestsCfg) { c.predicate = pred } +} + +// rewriteManifests is a producer that merges small data manifests into +// fewer, target-sized ones, committed as a metadata-only REPLACE snapshot. +type rewriteManifests struct { + base *snapshotProducer + cfg rewriteManifestsCfg + + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile + + // result is the value returned to the caller. record() rewrites its + // fields on every pass, including OCC retries, so the pointer the caller + // holds reflects the manifests actually committed, not attempt 0's. + result *RewriteManifestsResult +} + +func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer { + prod := createSnapshotProducer(OpReplace, txn, fs, nil, props) + prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result: &RewriteManifestsResult{}} + // A rewrite re-expresses inherited manifests, so OCC retries must + // re-derive from the fresh parent rather than carry these forward. + prod.rebuildMode = true + + return prod +} + +func (r *rewriteManifests) existingManifests() ([]iceberg.ManifestFile, error) { + snap := r.base.txn.meta.currentSnapshot() + if snap == nil { + return nil, nil + } + + return snap.Manifests(r.base.io) +} + +func (r *rewriteManifests) deletedEntries(context.Context) ([]iceberg.ManifestEntry, error) { + return nil, nil +} + +func (r *rewriteManifests) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + var toRewrite, kept []iceberg.ManifestFile + for _, m := range manifests { + if r.eligible(m) { + toRewrite = append(toRewrite, m) + } else { + kept = append(kept, m) + } + } + + mgr := manifestMergeManager{ + targetSizeBytes: r.cfg.targetSizeBytes, + minCountToMerge: 1, // force a merge regardless of count + mergeEnabled: true, // explicit op ignores commit.manifest-merge.enabled + snap: r.base, + } + merged, err := mgr.mergeManifests(toRewrite) + if err != nil { + return nil, err + } + + if err := validateRewriteFileCounts(toRewrite, merged); err != nil { + return nil, err + } + + // Record on every pass, not just the first. An OCC retry re-runs this + // against the fresh parent and writes a different merged set; the last + // pass is the one that commits, so its counts are the ones that must win. + r.record(toRewrite, merged, kept) + + return slices.Concat(merged, kept), nil +} + +// eligible reports whether m is a data manifest selected for rewrite. +func (r *rewriteManifests) eligible(m iceberg.ManifestFile) bool { + if m.ManifestContent() != iceberg.ManifestContentData { + return false + } + if r.cfg.specID != nil && int(m.PartitionSpecID()) != *r.cfg.specID { + return false + } + if r.cfg.predicate != nil && !r.cfg.predicate(m) { + return false + } + + return true +} + +func (r *rewriteManifests) record(toRewrite, merged, kept []iceberg.ManifestFile) { + // A prior pass (a superseded OCC attempt) wrote its own merged manifests + // that the catalog never referenced. Carry their paths into the + // producer's superseded set so commit() can delete them on success; + // otherwise every retry leaks a full set of merged .avro files. + for _, m := range r.added { + r.base.supersededManifests = append(r.base.supersededManifests, m.FilePath()) + } + r.added, r.rewritten = nil, nil + + inPaths := make(map[string]struct{}, len(toRewrite)) + for _, m := range toRewrite { + inPaths[m.FilePath()] = struct{}{} + } + outPaths := make(map[string]struct{}, len(merged)) + for _, m := range merged { + outPaths[m.FilePath()] = struct{}{} + } + + // Bins of a single manifest pass through unchanged; only the manifests + // that actually appear on one side and not the other are added/replaced. + // This drives the returned RewriteManifestsResult. + for _, m := range merged { + if _, ok := inPaths[m.FilePath()]; !ok { + r.added = append(r.added, m) + } + } + for _, m := range toRewrite { + if _, ok := outPaths[m.FilePath()]; !ok { + r.rewritten = append(r.rewritten, m) + } + } + + // Publish into the shared result the caller holds so a retry's counts win. + r.result.RewrittenManifests = r.rewritten + r.result.AddedManifests = r.added + + // The summary follows Java's Appendix-F accounting over the full eligible + // set, pass-throughs included, so it reads the same as a Java-written + // snapshot — distinct from the added/rewritten diff above. + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(merged)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(toRewrite)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(kept)) + if entries := manifestActiveFiles(toRewrite); entries >= 0 { + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(entries, 10) + } else { + delete(r.base.snapshotProps, entriesProcessedKey) + } +} + +func (r *rewriteManifests) validate(*conflictContext) error { return nil } +func (r *rewriteManifests) needsValidation() bool { return false } + +// manifestActiveFiles sums added + existing data files, or -1 if any manifest +// reports an unknown count. +func manifestActiveFiles(manifests []iceberg.ManifestFile) int64 { + var total int64 + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + if added < 0 || existing < 0 { + return -1 + } + total += int64(added) + int64(existing) + } + + return total +} + +// validateRewriteFileCounts guards against dropping or duplicating data files: +// the rewritten manifests must hold the same active files as their inputs. +func validateRewriteFileCounts(before, after []iceberg.ManifestFile) error { + in, out := manifestActiveFiles(before), manifestActiveFiles(after) + if in < 0 || out < 0 { + return nil // counts unknown; can't validate + } + if in != out { + return fmt.Errorf("rewrite manifests changed active file count: %d before, %d after", in, out) + } + + return nil +} + +// RewriteManifests merges small data manifests in the current snapshot into +// fewer, target-sized ones and stages the result as a REPLACE snapshot. It +// rewrites metadata only; no data files are read or written. Delete manifests +// are left untouched. +// +// An empty result (no AddedManifests and no RewrittenManifests) means there +// was nothing to do — either the table has no current snapshot or the +// manifests are already optimal. Callers should treat that as a no-op and +// skip the commit rather than staging an empty REPLACE snapshot. +func (t *Transaction) RewriteManifests(ctx context.Context, opts ...RewriteManifestsOpt) (*RewriteManifestsResult, error) { + if t.meta.currentSnapshot() == nil { + return &RewriteManifestsResult{}, nil Review Comment: Two small things tangled together here, both about the no-op shape — flagging them now mostly because they're public API and harder to change later. The no-snapshot case and the already-optimal case both return the same empty `RewriteManifestsResult{}`, so a caller can't tell them apart — the CLI ends up printing "Manifests already optimal" even for an empty table. A sentinel error for no-snapshot (or a little `IsNoOp`/reason on the result) would let callers branch cleanly. And further down, `RewriteManifests` calls `prod.commit` + `t.apply` unconditionally, so an empty rewrite still stages a REPLACE. The godoc kindly tells callers to skip the commit, but nothing enforces it, so a `RewriteManifests` + `Commit` lands an empty snapshot. Could we make the no-op real — skip `t.apply` when the result is empty — so `Commit` is a true no-op no matter what the caller does? ########## table/rewrite_manifests_test.go: ########## @@ -0,0 +1,525 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table_test + +import ( + "context" + "fmt" + "strconv" + "strings" + "sync" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/parquet/pqarrow" + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// activeFiles sums the added + existing data files across manifests. It mirrors +// the production accounting in manifestActiveFiles: a manifest that reports a +// negative (unknown) count fails the test rather than being silently folded in, +// so a count we can't trust never masquerades as a valid total. +func activeFiles(t *testing.T, manifests []iceberg.ManifestFile) int { + t.Helper() + var n int + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + require.False(t, added < 0 || existing < 0, + "manifest %s reports an unknown data-file count", m.FilePath()) + n += int(added) + int(existing) + } + + return n +} + +// rewriteArrowSchema is the single-column schema used by the rewrite tests. +var rewriteArrowSchema = arrow.NewSchema([]arrow.Field{ + {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, +}, nil) + +// writeOneRowParquet writes a one-row Parquet file at path. +func writeOneRowParquet(t *testing.T, fs iceio.WriteFileIO, path string) { + t.Helper() + + bldr := array.NewInt32Builder(memory.DefaultAllocator) + defer bldr.Release() + + bldr.AppendValues([]int32{1}, nil) + col := bldr.NewArray() + defer col.Release() + + rec := array.NewRecordBatch(rewriteArrowSchema, []arrow.Array{col}, 1) + defer rec.Release() + + arrTbl := array.NewTableFromRecords(rewriteArrowSchema, []arrow.RecordBatch{rec}) + defer arrTbl.Release() + + fo, err := fs.Create(path) + require.NoError(t, err) + require.NoError(t, pqarrow.WriteTable(arrTbl, fo, arrTbl.NumRows(), + nil, pqarrow.DefaultWriterProps())) +} + +func rewriteSchema() *iceberg.Schema { + return iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ) +} + +// appendSeparateManifests performs n single-row appends, each in its own +// commit, leaving n separate data manifests behind (merge must be disabled on +// the table). It returns the table after the final commit. +func appendSeparateManifests(t *testing.T, ctx context.Context, tbl *table.Table, fs iceio.WriteFileIO, dir, prefix string, n int) *table.Table { + t.Helper() + var err error + for i := range n { + filePath := fmt.Sprintf("%s/%s-%d.parquet", dir, prefix, i) + writeOneRowParquet(t, fs, filePath) + + txn := tbl.NewTransaction() + require.NoError(t, txn.AddFiles(ctx, []string{filePath}, nil, false)) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + } + + return tbl +} + +// TestRewriteManifests merges several small data manifests into one without +// changing the data files they reference. +func TestRewriteManifests(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + // Merge disabled, so each append leaves its own manifest behind. + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, iceberg.Properties{ + table.ManifestMergeEnabledKey: "false", + }) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + ident := table.Identifier{"default", "test_rewrite_manifests"} + tbl := table.New(ident, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, + cat, + ) + + const numCommits = 3 + tbl = appendSeparateManifests(t, ctx, tbl, fs, dir, "data", numCommits) + + before, err := tbl.CurrentSnapshot().Manifests(fs) + require.NoError(t, err) + require.Len(t, before, numCommits, "each append should leave its own manifest") + wantFiles := activeFiles(t, before) + require.Equal(t, numCommits, wantFiles) + + txn := tbl.NewTransaction() + res, err := txn.RewriteManifests(ctx) + require.NoError(t, err) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + + assert.Len(t, res.AddedManifests, 1) + assert.Len(t, res.RewrittenManifests, numCommits) + + snap := tbl.CurrentSnapshot() + require.NotNil(t, snap) + assert.Equal(t, table.OpReplace, snap.Summary.Operation) + assert.Equal(t, "1", snap.Summary.Properties["manifests-created"]) + assert.Equal(t, strconv.Itoa(numCommits), snap.Summary.Properties["manifests-replaced"]) + // All eligible: nothing is kept, and every input file is accounted for. + assert.Equal(t, "0", snap.Summary.Properties["manifests-kept"]) + assert.Equal(t, strconv.Itoa(numCommits), snap.Summary.Properties["entries-processed"]) + + after, err := snap.Manifests(fs) + require.NoError(t, err) + assert.Len(t, after, 1, "manifests should be merged into one") + assert.Equal(t, wantFiles, activeFiles(t, after), "rewrite must preserve the data file count") +} + +// TestRewriteManifestsNoSnapshot is a no-op (empty result, no error) when the +// table has no current snapshot, so callers can skip staging an empty REPLACE. +func TestRewriteManifestsNoSnapshot(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, nil) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + tbl := table.New(table.Identifier{"default", "empty"}, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, cat) + + txn := tbl.NewTransaction() + res, err := txn.RewriteManifests(ctx) + require.NoError(t, err) + require.NotNil(t, res) + assert.Empty(t, res.AddedManifests) + assert.Empty(t, res.RewrittenManifests) +} + +// TestRewriteManifestsAlreadyOptimal is a no-op when a single data manifest +// already holds everything: there is nothing to merge, so the result is empty +// and the caller should skip the commit. +func TestRewriteManifestsAlreadyOptimal(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + // Merge enabled so the single append leaves exactly one manifest. + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, nil) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + tbl := table.New(table.Identifier{"default", "optimal"}, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, cat) + + tbl = appendSeparateManifests(t, ctx, tbl, fs, dir, "data", 1) + + before, err := tbl.CurrentSnapshot().Manifests(fs) + require.NoError(t, err) + require.Len(t, before, 1) + + txn := tbl.NewTransaction() + res, err := txn.RewriteManifests(ctx) + require.NoError(t, err) + assert.Empty(t, res.AddedManifests, "a single manifest is already optimal") + assert.Empty(t, res.RewrittenManifests) +} + +// TestRewriteManifestsUnknownSpecID errors when the requested partition spec id +// does not exist on the table. +func TestRewriteManifestsUnknownSpecID(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, iceberg.Properties{ + table.ManifestMergeEnabledKey: "false", + }) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + tbl := table.New(table.Identifier{"default", "spec"}, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, cat) + + // Need a current snapshot, otherwise the no-op short-circuit fires first. + tbl = appendSeparateManifests(t, ctx, tbl, fs, dir, "data", 1) + + txn := tbl.NewTransaction() + _, err = txn.RewriteManifests(ctx, table.WithRewriteSpecID(999)) + require.Error(t, err) + assert.Contains(t, err.Error(), "unknown partition spec id 999") +} + +// TestRewriteManifestsPredicate rewrites only the manifests the predicate +// selects and leaves the rest untouched. +func TestRewriteManifestsPredicate(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, iceberg.Properties{ + table.ManifestMergeEnabledKey: "false", + }) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + tbl := table.New(table.Identifier{"default", "pred"}, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, cat) + + const numCommits = 3 + tbl = appendSeparateManifests(t, ctx, tbl, fs, dir, "data", numCommits) + + before, err := tbl.CurrentSnapshot().Manifests(fs) + require.NoError(t, err) + require.Len(t, before, numCommits) + + // Select two of the three manifests by path; the third must be left alone. + selected := map[string]struct{}{ + before[0].FilePath(): {}, + before[1].FilePath(): {}, + } + pred := func(m iceberg.ManifestFile) bool { + _, ok := selected[m.FilePath()] + + return ok + } + + txn := tbl.NewTransaction() + res, err := txn.RewriteManifests(ctx, table.WithRewriteManifestPredicate(pred)) + require.NoError(t, err) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + + assert.Len(t, res.AddedManifests, 1, "the two selected manifests merge into one") + assert.Len(t, res.RewrittenManifests, 2) + + snap := tbl.CurrentSnapshot() + require.NotNil(t, snap) + assert.Equal(t, "1", snap.Summary.Properties["manifests-created"]) + assert.Equal(t, "2", snap.Summary.Properties["manifests-replaced"]) + assert.Equal(t, "1", snap.Summary.Properties["manifests-kept"]) + + after, err := snap.Manifests(fs) + require.NoError(t, err) + assert.Len(t, after, 2, "one merged manifest plus the untouched one") + assert.Equal(t, numCommits, activeFiles(t, after), "no data file may be dropped") + + // The untouched manifest must survive verbatim. + var keptPaths []string + for _, m := range after { + keptPaths = append(keptPaths, m.FilePath()) + } + assert.Contains(t, keptPaths, before[2].FilePath(), "unselected manifest must be kept as-is") +} + +// classifyManifests counts data vs delete manifests and returns the path of the +// (single) delete manifest, if any. +func classifyManifests(t *testing.T, manifests []iceberg.ManifestFile) (data, delete int, deletePath string) { + t.Helper() + for _, m := range manifests { + if m.ManifestContent() == iceberg.ManifestContentData { + data++ + } else { + delete++ + deletePath = m.FilePath() + } + } + + return data, delete, deletePath +} + +// TestRewriteManifestsLeavesDeleteManifests verifies that a rewrite merges only +// data manifests and leaves delete manifests untouched. +func TestRewriteManifestsLeavesDeleteManifests(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + fs := iceio.LocalFS{} + + // Delete files require format version >= 2. + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, iceberg.Properties{ + table.PropertyFormatVersion: "2", + table.ManifestMergeEnabledKey: "false", + }) + require.NoError(t, err) + + cat := &mergeCatalog{meta: meta} + tbl := table.New(table.Identifier{"default", "del"}, meta, dir+"/metadata/00000.json", + func(_ context.Context) (iceio.IO, error) { return fs, nil }, cat) + + const numData = 2 + tbl = appendSeparateManifests(t, ctx, tbl, fs, dir, "data", numData) + + // Add a positional-delete file. RowDelta only records its metadata, so the + // referenced path need not exist on disk. This leaves a delete manifest the + // rewrite must not touch. + txn := tbl.NewTransaction() + require.NoError(t, txn.NewRowDelta(nil). + AddDeletes(buildPosDeleteFile(t, dir+"/data/pos-del.parquet")). + Commit(ctx)) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + + before, err := tbl.CurrentSnapshot().Manifests(fs) + require.NoError(t, err) + dataN, delN, beforeDeletePath := classifyManifests(t, before) + require.Equal(t, numData, dataN) + require.Equal(t, 1, delN, "setup must leave one delete manifest") + + txn = tbl.NewTransaction() + res, err := txn.RewriteManifests(ctx) + require.NoError(t, err) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + + assert.Len(t, res.AddedManifests, 1) + assert.Len(t, res.RewrittenManifests, numData, "only data manifests are rewritten") + + snap := tbl.CurrentSnapshot() + require.NotNil(t, snap) + assert.Equal(t, "1", snap.Summary.Properties["manifests-created"]) + assert.Equal(t, strconv.Itoa(numData), snap.Summary.Properties["manifests-replaced"]) + assert.Equal(t, "1", snap.Summary.Properties["manifests-kept"], "the delete manifest is kept") + + after, err := snap.Manifests(fs) + require.NoError(t, err) + afterData, afterDel, afterDeletePath := classifyManifests(t, after) + assert.Equal(t, 1, afterData, "data manifests merged into one") + assert.Equal(t, 1, afterDel, "delete manifest preserved") + assert.Equal(t, beforeDeletePath, afterDeletePath, "delete manifest must survive verbatim") +} + +// trackingFS wraps LocalFS and records the paths it creates and removes so a +// test can assert that orphaned manifests are cleaned up after a commit. +type trackingFS struct { + iceio.LocalFS + mu sync.Mutex + created []string + removed []string +} + +func (f *trackingFS) Create(name string) (iceio.FileWriter, error) { + f.mu.Lock() + f.created = append(f.created, name) + f.mu.Unlock() + + return f.LocalFS.Create(name) +} + +func (f *trackingFS) Remove(name string) error { + f.mu.Lock() + f.removed = append(f.removed, name) + f.mu.Unlock() + + return f.LocalFS.Remove(name) +} + +func (f *trackingFS) snapshotCreated() map[string]struct{} { + f.mu.Lock() + defer f.mu.Unlock() + out := make(map[string]struct{}, len(f.created)) + for _, p := range f.created { + out[p] = struct{}{} + } + + return out +} + +func (f *trackingFS) wasRemoved(path string) bool { + f.mu.Lock() + defer f.mu.Unlock() + for _, p := range f.removed { + if p == path { + return true + } + } + + return false +} + +// TestRewriteManifestsStaleCountsAfterOCCRetry forces a commit conflict and +// asserts that the RewriteManifestsResult and the committed summary reflect the +// winning attempt — the one rebuilt against the fresh parent — not the stale +// attempt 0. It also asserts that the manifests written by the superseded +// attempt are deleted rather than leaked. +func TestRewriteManifestsStaleCountsAfterOCCRetry(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + track := &trackingFS{} + fsF := func(_ context.Context) (iceio.IO, error) { return track, nil } + + props := iceberg.Properties{ + table.PropertyFormatVersion: "2", + table.ManifestMergeEnabledKey: "false", + table.CommitMinRetryWaitMsKey: "0", + table.CommitMaxRetryWaitMsKey: "0", + table.CommitTotalRetryTimeoutMsKey: "60000", + table.CommitNumRetriesKey: "2", + } + meta, err := table.NewMetadata(rewriteSchema(), iceberg.UnpartitionedSpec, + table.UnsortedSortOrder, dir, props) + require.NoError(t, err) + + cat := &occScenarioCatalog{current: meta, conflictsLeft: 0, location: dir} + ident := table.Identifier{"default", "occ_rewrite"} + tbl := table.New(ident, meta, dir+"/metadata/00000.json", fsF, cat) + + // Writer A's stale view: three separate data manifests. + const staleCount = 3 + tbl = appendSeparateManifests(t, ctx, tbl, track, dir, "stale", staleCount) + staleTbl := tbl // metadata frozen at three manifests + + // A concurrent writer adds a fourth manifest the catalog now knows about + // but Writer A does not. The rewrite, on retry, must re-merge all four. + const freshCount = staleCount + 1 + _ = appendSeparateManifests(t, ctx, tbl, track, dir, "fresh", 1) + freshSnap := cat.current.CurrentSnapshot() + require.NotNil(t, freshSnap) + freshManifests, err := freshSnap.Manifests(track) + require.NoError(t, err) + require.Len(t, freshManifests, freshCount, "catalog should now hold four manifests") + + // Force exactly one conflict so Writer A retries against the fresh parent. + cat.conflictsLeft = 1 Review Comment: Really glad this test asserts the orphan cleanup at all — that's the part I'd most worry about. One extension I'd suggest: it only forces a single conflict, and `supersededManifests` is built to accumulate across multiple retries, so the per-generation `record()` accumulation only gets exercised once. A `conflictsLeft=2` case that checks every superseded generation gets cleaned up would lock that down, and pairing it with an error-path case (inject a `Create` failure on a later retry, per the table.go note) would cover the leak corner too. -- 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]
