tanmayrauth commented on code in PR #1283: URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3464590475
########## table/rewrite_manifests_test.go: ########## @@ -0,0 +1,155 @@ +// Licensed to the Apache Software Foundation (ASF) under one Review Comment: Could we add coverage for the spec-id filter, the predicate option, delete manifests being left untouched, and the already-optimal no-op path? The OCC-retry rebuild in particular (tied to the once.Do question above) is currently untested. ########## 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) Review Comment: `manifestActiveFiles` returns -1 on an unknown count, so this can write "entries-processed": "-1" into the summary. Might be cleaner to omit the key when the count is unknown. ########## 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: The `once.Do` freezes `rewritten`/`added` and the summary props at attempt 0, but on an OCC retry `rebuildFn` re-runs `processManifests` against the fresh parent and writes a different set of merged manifests. So under contention the returned `RewriteManifestsResult` points at attempt-0 manifest files that got superseded/orphaned, and the committed `manifests-created`/`manifests-replaced` counts may not match what was actually written. Is reusing attempt-0 results here intentional, or should the result/summary reflect the winning attempt? ########## 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) Review Comment: This always stages a REPLACE snapshot even when nothing was eligible (merge returns the inputs unchanged, added/rewritten empty) — only the CLI guards against committing the empty snapshot. A library caller doing RewriteManifests + Commit would get an empty REPLACE snapshot. Worth detecting the no-op here, or at least documenting that callers must check the result before committing? -- 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]
