rambleraptor commented on code in PR #1283: URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3531795992
########## table/rewrite_manifests.go: ########## @@ -0,0 +1,368 @@ +// 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" +) + +// 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 string + +const ( + // NoOpNone means the rewrite produced changes (not a no-op). + NoOpNone NoOpReason = "" + // NoOpNoSnapshot means the table had no current snapshot to rewrite. + NoOpNoSnapshot NoOpReason = "no current snapshot" + // NoOpAlreadyOptimal means the eligible manifests were already optimal. + NoOpAlreadyOptimal NoOpReason = "manifests already optimal" +) + +// 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. Empty 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. +func (r *RewriteManifestsResult) IsNoOp() bool { + return len(r.AddedManifests) == 0 && len(r.RewrittenManifests) == 0 +} + +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 + + // 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 is true: a rewrite re-expresses inherited manifests, +// so OCC retries re-derive from the fresh parent rather than carry forward. +func (r *rewriteManifests) rebuildFromInheritedOnly() bool { return true } + +// supersededManifests exposes the accumulator so doCommit can clean up merged +// manifests orphaned across OCC retries. +func (r *rewriteManifests) supersededManifests() *[]string { return &r.superseded } + +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(r.base.io, 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.superseded = append(r.superseded, 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, which counts only the + // manifests a merge actually wrote and consumed — not single-manifest bins + // that pass through untouched. Driving it off the added/rewritten diff (and + // counting entries over the rewritten set) makes a Go-written snapshot read + // the same as a Java one. + 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(kept)) + // Best-effort: the rewritten set was just read by validateRewriteFileCounts, + // so a count here should succeed; drop the key rather than fail if it can't. + if entries, err := manifestActiveFiles(r.base.io, r.rewritten); err == nil { + 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 the active (added + existing) data files across +// manifests. It uses the manifest header counts when present and falls back to +// reading and counting the live entries when a count is absent — V1 manifests +// don't populate the count fields, so without the fallback the count guard +// would silently not run on exactly the tables that most need it. +func manifestActiveFiles(fs iceio.IO, manifests []iceberg.ManifestFile) (int64, error) { + var total int64 + for _, m := range manifests { + added, existing := m.AddedDataFiles(), m.ExistingDataFiles() + if added < 0 || existing < 0 { + // discardDeleted drops DELETED entries, leaving added + existing. + for _, err := range m.Entries(fs, true) { + if err != nil { + return 0, fmt.Errorf("count active files in manifest %s: %w", m.FilePath(), err) + } + total++ + } + + continue + } + total += int64(added) + int64(existing) + } + + return total, nil +} + +// validateRewriteFileCounts guards against dropping or duplicating data files: +// the rewritten manifests must hold the same active files as their inputs. +func validateRewriteFileCounts(fs iceio.IO, before, after []iceberg.ManifestFile) error { + in, err := manifestActiveFiles(fs, before) + if err != nil { + return err + } + out, err := manifestActiveFiles(fs, after) + if err != nil { + return err + } + 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. +// +// Manifests are clustered by size only (bin-packed toward the target size); +// clustering by partition or sort key, which Java exposes via clusterBy, is a +// future extension. +// +// A no-op result (IsNoOp) means there was nothing to do — either the table has +// no current snapshot (NoOpNoSnapshot) or the eligible manifests are already +// optimal (NoOpAlreadyOptimal). A no-op writes no manifest list and stages +// nothing on the transaction, so a following Commit is a true no-op; callers +// can skip it either way. +func (t *Transaction) RewriteManifests(ctx context.Context, opts ...RewriteManifestsOpt) (*RewriteManifestsResult, error) { Review Comment: Yeah, that's a valid concern. I end up biasing towards "if Java does it, it's fine + doesn't need explanation", but I added the comment. -- 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]
