laskoviymishka commented on code in PR #1283: URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3466843036
########## cmd/iceberg/rewrite_manifests.go: ########## @@ -0,0 +1,67 @@ +// 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 main + +import ( + "context" + "fmt" + "os" + + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/table" +) + +func runRewriteManifests(ctx context.Context, output Output, cat catalog.Catalog, cmd *RewriteManifestsCmd) { + tbl := loadTable(ctx, output, cat, cmd.TableID) + + if tbl.CurrentSnapshot() == nil { + output.Text("No current snapshot; nothing to rewrite.") + + return + } + + var opts []table.RewriteManifestsOpt + if cmd.TargetManifestSize > 0 { + opts = append(opts, table.WithManifestTargetSize(int(cmd.TargetManifestSize))) Review Comment: `cmd.TargetManifestSize` is `int64` but `WithManifestTargetSize` takes `int`, so this truncates on 32-bit builds. Java's target size is a `long` and `ManifestFile.Length()` is `int64`, so I'd make the option `int64` to match — that also drops the cast here. It's a public API, so easier to get right now than to widen later. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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 + } + + // Capture results and summary once, from the first (synchronous) pass so + // they reach the attempt-0 summary; retries reuse that summary. + r.once.Do(func() { r.record(toRewrite, merged, kept) }) Review Comment: I think there's a correctness bug here once an OCC retry fires. On retry, `commit()` has cleared `ownManifests`, so the rebuild re-runs `processManifests` against the fresh parent and `mergeManifests` writes a brand-new set of merged manifests. But `r.once` already fired on attempt 0, so `record()` is skipped — which means `r.added`/`r.rewritten` and the four `snapshotProps` summary keys still describe the attempt-0 manifests, not the ones actually committed. So both the `RewriteManifestsResult` we hand back and the committed snapshot summary are frozen at values that don't match what's on disk. I'd drop the `sync.Once` and make `record()` re-entrant — reset `r.added`/`r.rewritten` at the top and call it on every pass. The last pass is the one that commits, so its counts are the ones that should win. wdyt? ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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) Review Comment: Related to the retry path: each `processManifests` call writes fresh merged `.avro` files via `createManifest`, but only the final attempt's manifests end up referenced by the committed snapshot. The earlier attempts' files are never referenced and never cleaned up, so every OCC retry leaks a full set of merged manifests permanently. The manifest-list orphan handling we already have only covers manifest-list files, not the inner manifest avro. This exists in a milder form in `mergeAppendFiles`, but it's structurally worse here because a rewrite re-merges everything on every retry rather than appending. Java tracks created manifests and runs `cleanUncommitted` for exactly this. I'd either track the paths we write so they can be cleaned on a superseded attempt, or skip the re-merge when the input already equals the fresh parent. Either works — I'd just want one of them before merge. ########## cmd/iceberg/rewrite_manifests.go: ########## @@ -0,0 +1,67 @@ +// 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 main + +import ( + "context" + "fmt" + "os" + + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/table" +) + +func runRewriteManifests(ctx context.Context, output Output, cat catalog.Catalog, cmd *RewriteManifestsCmd) { + tbl := loadTable(ctx, output, cat, cmd.TableID) + + if tbl.CurrentSnapshot() == nil { Review Comment: We've got two different answers for the empty-snapshot case: the CLI returns exit-0 with a "nothing to rewrite" message here, but the library returns a hard error from `RewriteManifests`. That's inconsistent and means the check lives in two places. I'd pick one. Simplest is to drop this CLI pre-check and have the API return `&RewriteManifestsResult{}, nil` for the no-snapshot case so it flows through the same empty-result path as "nothing eligible." Alternatively, export a sentinel (`ErrNoCurrentSnapshot`, like the existing `ErrSnapshotNotFound`) and have the CLI translate it. Either is fine, but it should be one source of truth. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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 + } + + // Capture results and summary once, from the first (synchronous) pass so + // they reach the attempt-0 summary; retries reuse that summary. + r.once.Do(func() { 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) { + 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. + 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) + } + } + + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(r.added)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(r.rewritten)) Review Comment: The summary counts diverge from Java's Appendix-F accounting. Java counts `manifests-replaced` as all predicate-matching manifests and `entries-processed` as all live entries across them — including the single-manifest bins that pass through unchanged — whereas here both are computed off `r.rewritten`, which excludes the pass-throughs. The pass-through optimization itself is correct and I wouldn't touch it; this is purely about what the summary reports. I'd compute these against the eligible set instead: `manifestsReplacedKey = len(toRewrite)`, `manifestsCreatedKey = len(merged)`, and `entriesProcessedKey = manifestActiveFiles(toRewrite)`. That way the summary reflects everything we considered, pass-throughs included, and reads the same as a Java-written snapshot. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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 + } + + // Capture results and summary once, from the first (synchronous) pass so + // they reach the attempt-0 summary; retries reuse that summary. + r.once.Do(func() { 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) { + 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. + 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) + } + } + + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(r.added)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(r.rewritten)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(merged) - len(r.added) + len(kept)) + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(manifestActiveFiles(r.rewritten), 10) +} + +// rebuildFromInheritedOnly tells commit() to drop ownManifests so OCC retries +// re-merge the fresh parent's manifests rather than the stale rewrite. +func (r *rewriteManifests) rebuildFromInheritedOnly() bool { return true } + +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. +func (t *Transaction) RewriteManifests(ctx context.Context, opts ...RewriteManifestsOpt) (*RewriteManifestsResult, error) { + if t.meta.currentSnapshot() == nil { + return nil, errors.New("cannot rewrite manifests: table has no current snapshot") + } + + cfg := rewriteManifestsCfg{ + targetSizeBytes: t.meta.props.GetInt(ManifestTargetSizeBytesKey, ManifestTargetSizeBytesDefault), + } + for _, o := range opts { + o(&cfg) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + wfs, ok := fs.(iceio.WriteFileIO) + if !ok { + return nil, ErrWriteIORequired + } + + prod := newRewriteManifestsProducer(t, wfs, iceberg.Properties{}, cfg) + updates, reqs, err := prod.commit(ctx) + if err != nil { + return nil, err + } + if err := t.apply(updates, reqs); err != nil { + return nil, err + } + + impl := prod.producerImpl.(*rewriteManifests) Review Comment: Bare type assertion will panic if the producer impl is ever anything other than `*rewriteManifests`. The function already returns an error, so I'd use comma-ok: ```go impl, ok := prod.producerImpl.(*rewriteManifests) if !ok { return nil, fmt.Errorf("internal error: unexpected producer type %T", prod.producerImpl) } ``` ########## table/rewrite_manifests_test.go: ########## @@ -0,0 +1,155 @@ +// 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" + "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. +func activeFiles(t *testing.T, manifests []iceberg.ManifestFile) int { + t.Helper() + var n int + for _, m := range manifests { + n += int(m.AddedDataFiles()) + int(m.ExistingDataFiles()) + } + + return n +} + +// 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{} + + arrowSchema := arrow.NewSchema([]arrow.Field{ + {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + }, nil) + + writeParquet := func(path string) { + bldr := array.NewInt32Builder(memory.DefaultAllocator) + defer bldr.Release() + + bldr.AppendValues([]int32{1}, nil) + col := bldr.NewArray() + defer col.Release() + + rec := array.NewRecordBatch(arrowSchema, []arrow.Array{col}, 1) + defer rec.Release() + + arrTbl := array.NewTableFromRecords(arrowSchema, []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())) + } + + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ) + + // Merge disabled, so each append leaves its own manifest behind. + meta, err := table.NewMetadata(schema, 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 + for i := range numCommits { + filePath := fmt.Sprintf("%s/data-%d.parquet", dir, i) + writeParquet(filePath) + + txn := tbl.NewTransaction() + require.NoError(t, txn.AddFiles(ctx, []string{filePath}, nil, false)) + tbl, err = txn.Commit(ctx) + require.NoError(t, err) + } + + 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"]) + + 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") Review Comment: The happy path is covered well, but the OCC-retry path isn't tested at all — and that's exactly where the `record()`/`sync.Once` staleness bug lives, so right now nothing would catch it. I'd add a case that forces a retry (the `blockingTrackingIO` + `occ_scenario_test.go` setup shows the pattern) and asserts that both the `RewriteManifestsResult` and the committed summary counts match the committed manifests, not attempt 0. While we're in here, a couple of cheaper gaps worth closing: `WithRewriteSpecID`, `WithRewriteManifestPredicate`, and `WithManifestTargetSize` binning are all unexercised, and this test doesn't prove delete-manifest isolation — adding a delete manifest and asserting it survives untouched would lock that in. ########## table/rewrite_manifests_test.go: ########## @@ -0,0 +1,155 @@ +// 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" + "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. +func activeFiles(t *testing.T, manifests []iceberg.ManifestFile) int { + t.Helper() + var n int + for _, m := range manifests { + n += int(m.AddedDataFiles()) + int(m.ExistingDataFiles()) Review Comment: This helper doesn't mirror the production `manifestActiveFiles`, which returns -1 when a manifest reports an unknown count. Here a -1 widens into a large positive int and silently corrupts the sum, so a test could pass on a count that production would have flagged as unvalidatable. I'd mirror production and `require.False` on a negative count instead of folding it in. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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 + } + + // Capture results and summary once, from the first (synchronous) pass so + // they reach the attempt-0 summary; retries reuse that summary. + r.once.Do(func() { 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) { + 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. + 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) + } + } + + r.base.snapshotProps[manifestsCreatedKey] = strconv.Itoa(len(r.added)) + r.base.snapshotProps[manifestsReplacedKey] = strconv.Itoa(len(r.rewritten)) + r.base.snapshotProps[manifestsKeptKey] = strconv.Itoa(len(merged) - len(r.added) + len(kept)) + r.base.snapshotProps[entriesProcessedKey] = strconv.FormatInt(manifestActiveFiles(r.rewritten), 10) +} + +// rebuildFromInheritedOnly tells commit() to drop ownManifests so OCC retries +// re-merge the fresh parent's manifests rather than the stale rewrite. +func (r *rewriteManifests) rebuildFromInheritedOnly() bool { return true } Review Comment: On your note in the PR description — yeah, I'd express this differently. This reads as a producer-lifecycle concern (how `commit()` should handle a retry) rather than a producer-contract concern (what the producer does), so putting it on the `producerImpl` interface means every implementor has to stub `return false` for something that isn't really about them. I'd make it a plain field on `snapshotProducer` instead and have `newRewriteManifestsProducer` set it: ```go type snapshotProducer struct { // ... rebuildMode bool // re-derive manifests from fresh parent on OCC retry } ``` Then `commit()` checks `sp.rebuildMode` directly and the three `return false` stubs on `fastAppendFiles`/`overwriteFiles`/`errorOnDeletedEntries` all go away. Keeps the interface about producer behavior and the lifecycle knob where it's actually consumed. wdyt? ########## cmd/iceberg/rewrite_manifests.go: ########## @@ -0,0 +1,67 @@ +// 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 main + +import ( + "context" + "fmt" + "os" + + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/table" +) + +func runRewriteManifests(ctx context.Context, output Output, cat catalog.Catalog, cmd *RewriteManifestsCmd) { + tbl := loadTable(ctx, output, cat, cmd.TableID) + + if tbl.CurrentSnapshot() == nil { + output.Text("No current snapshot; nothing to rewrite.") + + return + } + + var opts []table.RewriteManifestsOpt + if cmd.TargetManifestSize > 0 { + opts = append(opts, table.WithManifestTargetSize(int(cmd.TargetManifestSize))) + } + if cmd.SpecID >= 0 { + opts = append(opts, table.WithRewriteSpecID(cmd.SpecID)) + } + + tx := tbl.NewTransaction() + res, err := tx.RewriteManifests(ctx, opts...) + if err != nil { + output.Error(fmt.Errorf("rewrite manifests failed: %w", err)) + os.Exit(1) Review Comment: The repo wires `os.Exit` through the `osExit` seam in `exit.go` (`var osExit = os.Exit`) so exit paths stay testable. I'd use `osExit(1)` here and at the commit-failure site below to match the rest of the CLI. ########## table/rewrite_manifests.go: ########## @@ -0,0 +1,261 @@ +// 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" + "errors" + "fmt" + "slices" + "strconv" + "sync" + + "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 int + 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 int) 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 + + once sync.Once + rewritten []iceberg.ManifestFile + added []iceberg.ManifestFile +} + +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} + + 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 + } + + // Capture results and summary once, from the first (synchronous) pass so + // they reach the attempt-0 summary; retries reuse that summary. + r.once.Do(func() { 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 { Review Comment: `WithRewriteSpecID` with a spec ID that doesn't exist in `t.meta.specs` silently matches nothing and rewrites zero manifests, which is hard to distinguish from "already optimal." I'd validate the id against the known specs up front and error on an unknown one. Minor, but it turns a silent no-op into a clear signal. -- 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]
