zeroshade commented on code in PR #1283:
URL: https://github.com/apache/iceberg-go/pull/1283#discussion_r3521443556
##########
table/table.go:
##########
@@ -526,6 +526,20 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
}
}
+ // Inner data manifests written by superseded retry attempts (a rewrite
+ // re-merges everything on each retry) are orphaned objects: on success
the
+ // committed snapshot references only the final attempt's, and on a
failed
+ // commit nothing references any of them. They are always safe to
delete —
+ // the winning attempt's manifests are never added to this set — so
collect
+ // them on both the success and the safe-failure (exhausted
ErrCommitFailed)
+ // paths. The defer skips cleanup only on the unsafe non-ErrCommitFailed
+ // path, which already returned above with cleanupOrphans = false.
+ for _, u := range updates {
+ if su, ok := u.(*addSnapshotUpdate); ok &&
su.supersededManifests != nil {
+ orphanedManifests = append(orphanedManifests,
*su.supersededManifests...)
Review Comment:
This only collects `supersededManifests`, but `record()` only moves the
previous attempt's `r.added` into that slice
(table/rewrite_manifests.go:200-203). When retries are exhausted with
`ErrCommitFailed`, the final failed attempt's merged manifest was written but
is never moved into `supersededManifests`, so this cleanup misses it; the new
test currently codifies that final leak as allowed. Please also clean the
current attempt's uncommitted rewrite outputs on the safe exhausted-failure
path.
##########
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
Review Comment:
`WithManifestTargetSize` silently ignores zero or negative sizes. Please
either document that non-positive values mean 'use the table property' or
return an error/validate earlier so invalid caller input is not silently
accepted.
##########
table/snapshot_producers.go:
##########
@@ -60,6 +60,15 @@ type producerImpl interface {
// unconditionally a no-op; commit() skips validator registration
// entirely when this returns false, so validate will never run.
needsValidation() bool
+ // rebuildFromInheritedOnly reports whether OCC retries should re-derive
+ // this producer's manifests from the fresh parent rather than carrying
+ // forward the manifests built on the first attempt. A rewrite
re-expresses
Review Comment:
These hooks are rewrite-only behavior, but adding them to `producerImpl`
forces every producer and test fake to implement no-op methods. Please keep the
shared interface focused on common producer behavior and use unexported
optional capability interfaces at the call sites that need rewrite-specific
retry/cleanup behavior.
##########
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 = ""
Review Comment:
`NoOpNone` is the empty-string zero value while `IsNoOp()` derives no-op
state from slice lengths, so the explicit reason can drift from the result
state and the zero value is indistinguishable from unset. Please make this an
iota-style enum with a concrete `NoOpNone`, and make `NoOpReason` the single
source of truth for whether the result is a no-op.
##########
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 {
Review Comment:
`mergeManifests` can write merged manifests before
`validateRewriteFileCounts` runs. If validation (or any later path before
`record()`) returns an error, those newly created paths have not been
registered in `added`/`superseded`, so they are orphaned. Please register the
created manifest paths immediately and then hand cleanup ownership to the
commit/retry mechanism.
--
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]
