zeroshade commented on code in PR #1283: URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3553634343
########## table/rewrite_manifests.go: ########## @@ -0,0 +1,424 @@ +// 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" + "log" + "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" // new manifests the merge wrote + manifestsReplacedKey = "manifests-replaced" // input manifests the merge consumed + manifestsKeptKey = "manifests-kept" // manifests left untouched, delete manifests included + entriesProcessedKey = "entries-processed" // data-file entries, not rows +) + +// NoOpReason explains why a rewrite changed nothing. It lets callers tell a +// table with no current snapshot apart from one whose manifests are already +// optimal, which would otherwise both surface as an empty result. +type NoOpReason int + +const ( + // NoOpNone means the rewrite produced changes (not a no-op). + NoOpNone NoOpReason = iota + // NoOpNoSnapshot means the table had no current snapshot to rewrite. + NoOpNoSnapshot + // NoOpAlreadyOptimal means the eligible manifests were already optimal. + NoOpAlreadyOptimal +) + +func (r NoOpReason) String() string { + switch r { + case NoOpNoSnapshot: + return "no current snapshot" + case NoOpAlreadyOptimal: + return "manifests already optimal" + default: + return "none" + } +} + +// 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 + // NoOpReason is set when the rewrite changed nothing, distinguishing a + // missing snapshot from an already-optimal layout. NoOpNone otherwise. + NoOpReason NoOpReason +} + +// IsNoOp reports whether the rewrite changed nothing. Callers should skip the +// commit in that case rather than staging an empty REPLACE snapshot. NoOpReason +// is the single source of truth. +func (r *RewriteManifestsResult) IsNoOp() bool { + return r.NoOpReason != NoOpNone +} + +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. A +// non-positive size is ignored, leaving the default 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 + + // superseded accumulates merged manifest files written by a rebuild on a + // prior OCC attempt that a later attempt replaced. doCommit removes them + // after a successful commit so retries don't leak orphaned manifests. + superseded []string + + // 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{}} + + return prod +} + +// rebuildFromInheritedOnly marks this producer: a rewrite re-expresses inherited +// manifests, so OCC retries re-derive from the fresh parent rather than carry +// forward. +func (r *rewriteManifests) rebuildFromInheritedOnly() {} + +// supersededManifests returns the merged manifests orphaned across OCC retries, +// safe to delete once the commit resolves. When the commit never landed +// (committed is false) the final attempt's manifests are orphaned too — a +// snapshot that never committed references nothing — so they are appended. +func (r *rewriteManifests) supersededManifests(committed bool) []string { + if committed { + return r.superseded + } + + paths := slices.Clone(r.superseded) + for _, m := range r.added { + paths = append(paths, m.FilePath()) + } + + return paths +} + +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 Review Comment: If `mergeManifests` errors after one or more bins have already written merged `.avro` files (bins are written concurrently in `mergeGroup`), this early return leaks them — the `deleteWritten` defer is only installed after a successful return a few lines below. This is the remaining orphan-manifest cleanup gap from the earlier review. -- 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]
