zeroshade commented on code in PR #1940: URL: https://github.com/apache/iceberg-go/pull/1940#discussion_r3935765412
########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,247 @@ +// 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 ( + "errors" + "fmt" + "io" + "log" + "reflect" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/internal" +) + +type manifestClusterKey struct { + specID int + value any +} + +type manifestClusterWriter struct { + writer *iceberg.ManifestWriter + path string + counter *internal.CountingWriter + fileCloser io.Closer + hasEntries bool + manifests []iceberg.ManifestFile +} + +func (w *manifestClusterWriter) close() (mf iceberg.ManifestFile, err error) { + if w.writer == nil { + return nil, nil + } + + writer := w.writer + w.writer = nil + defer func() { + if w.fileCloser != nil { + err = errors.Join(err, w.fileCloser.Close()) + w.fileCloser = nil + } + }() + + if err := writer.Close(); err != nil { + return nil, err + } + + return writer.ToManifestFile(w.path, w.counter.Count) +} + +func (w *manifestClusterWriter) abort() { + if w.writer != nil { + _ = w.writer.Close() + w.writer = nil + } + if w.fileCloser != nil { + _ = w.fileCloser.Close() + w.fileCloser = nil + } +} + +func (m *manifestMergeManager) newClusterWriter(specID int) (*manifestClusterWriter, error) { + spec, err := m.snap.spec(specID) + if err != nil { + return nil, err + } + + writer, path, counter, fileCloser, err := m.snap.newManifestWriter(spec) + if err != nil { + return nil, err + } + + return &manifestClusterWriter{ + writer: writer, + path: path, + counter: counter, + fileCloser: fileCloser, + }, nil +} + +func validateManifestClusterKey(key any) error { + if key == nil { + return errors.New("manifest cluster key must be non-nil") + } + + typ := reflect.TypeOf(key) + value := reflect.ValueOf(key) + if !value.Comparable() { + return fmt.Errorf("manifest cluster key type %s is not comparable", typ) + } + + switch value.Kind() { + case reflect.Chan, reflect.Pointer, reflect.UnsafePointer: + if value.IsNil() { + return errors.New("manifest cluster key must be non-nil") + } + } + if !value.Equal(value) { + return fmt.Errorf("manifest cluster key type %s is not reflexive", typ) + } + + return nil +} + +func validateManifestClusterKeyComparable(key any) error { + if key == nil { + return errors.New("manifest cluster key must be non-nil") + } + + value := reflect.ValueOf(key) + if !value.Comparable() { + return fmt.Errorf("manifest cluster key type %s is not comparable", value.Type()) + } + + return nil +} + +// clusterManifests rewrites entries into one rolling writer per cluster key and +// partition spec. Writers stay open while entries for other keys are read so a +// later file with the same key is still written beside the earlier files. This +// assumes a producer that only reorganizes live data entries; it does not add +// new files or preserve tombstones. +func (m *manifestMergeManager) clusterManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + // One output writer is tracked per key, so reserve space for the common + // case where each input manifest introduces a new cluster. + writers := make(map[manifestClusterKey]*manifestClusterWriter, len(manifests)) + order := make([]manifestClusterKey, 0, len(manifests)) + paths := make([]string, 0, len(manifests)) + completed := false + + defer func() { + if completed { + return + } + + for _, writer := range writers { + writer.abort() + } + for _, path := range paths { + if removeErr := m.snap.io.Remove(path); removeErr != nil { + log.Printf("Warning: failed to delete orphaned clustered manifest %s: %v", path, removeErr) + } + } + }() + + closeWriter := func(writer *manifestClusterWriter) error { + manifest, closeErr := writer.close() + if closeErr != nil { + return closeErr + } + if manifest != nil { + writer.manifests = append(writer.manifests, manifest) + } + + return nil + } + + for _, manifest := range manifests { + specID := int(manifest.PartitionSpecID()) + for entry, entryErr := range m.snap.iterManifestEntries(manifest, true) { + if entryErr != nil { + return nil, entryErr + } + + clusterValue := m.clusterBy(entry.DataFile()) + if clusterErr := validateManifestClusterKeyComparable(clusterValue); clusterErr != nil { + return nil, fmt.Errorf("cluster data file %q: %w", entry.DataFile().FilePath(), clusterErr) + } + key := manifestClusterKey{specID: specID, value: clusterValue} + writer, ok := writers[key] + if !ok { + if clusterErr := validateManifestClusterKey(clusterValue); clusterErr != nil { + return nil, fmt.Errorf("cluster data file %q: %w", entry.DataFile().FilePath(), clusterErr) + } + var openErr error + writer, openErr = m.newClusterWriter(specID) + if openErr != nil { + return nil, openErr + } + writers[key] = writer + order = append(order, key) + paths = append(paths, writer.path) + } + + // The counter advances when the Avro writer flushes a block, so a + // manifest can exceed targetSizeBytes by nearly one block. + if writer.writer != nil && writer.hasEntries && m.targetSizeBytes > 0 && writer.counter.Count >= m.targetSizeBytes { + if entryErr := closeWriter(writer); entryErr != nil { + return nil, entryErr + } + } + if writer.writer == nil { + next, openErr := m.newClusterWriter(specID) + if openErr != nil { + return nil, openErr + } + writer.writer = next.writer + writer.path = next.path + writer.counter = next.counter + writer.fileCloser = next.fileCloser + writer.hasEntries = false + paths = append(paths, writer.path) + } + + if err := writer.writer.Existing(entry); err != nil { + return nil, err + } + writer.hasEntries = true + } + } + + for _, key := range order { + writer := writers[key] + if writer.writer == nil || !writer.hasEntries { + writer.abort() Review Comment: **nit** — Unreachable defensive branch in the finalization loop can silently orphan a file if it ever becomes reachable `if writer.writer == nil || !writer.hasEntries { writer.abort(); continue }` cannot fire: every writer placed in `order` is opened immediately before an `Existing(entry)` call (:220-223), and the roll path at :207-218 re-opens and then writes before the loop advances, so at finalization every writer has a live *ManifestWriter and hasEntries==true. Any error path returns early instead of reaching here. The branch is harmless today, but it is the one place that abandons a writer while `completed` is later set to true at :244 — which means the file at `writer.path`, already appended to `paths` at :197/:217, would be skipped by the cleanup defer at :154-158 and left as a silent orphan. Either drop the branch, or if it is kept as a guard, remove its path from `paths` (or delete the file) so the invariant 'a manifest we abort is never left behind' holds unconditionally. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,247 @@ +// 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 ( + "errors" + "fmt" + "io" + "log" + "reflect" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/internal" +) + +type manifestClusterKey struct { + specID int + value any +} + +type manifestClusterWriter struct { + writer *iceberg.ManifestWriter + path string + counter *internal.CountingWriter + fileCloser io.Closer + hasEntries bool + manifests []iceberg.ManifestFile +} + +func (w *manifestClusterWriter) close() (mf iceberg.ManifestFile, err error) { + if w.writer == nil { + return nil, nil + } + + writer := w.writer + w.writer = nil + defer func() { + if w.fileCloser != nil { + err = errors.Join(err, w.fileCloser.Close()) + w.fileCloser = nil + } + }() + + if err := writer.Close(); err != nil { + return nil, err + } + + return writer.ToManifestFile(w.path, w.counter.Count) +} + +func (w *manifestClusterWriter) abort() { + if w.writer != nil { + _ = w.writer.Close() + w.writer = nil + } + if w.fileCloser != nil { + _ = w.fileCloser.Close() + w.fileCloser = nil + } +} + +func (m *manifestMergeManager) newClusterWriter(specID int) (*manifestClusterWriter, error) { + spec, err := m.snap.spec(specID) + if err != nil { + return nil, err + } + + writer, path, counter, fileCloser, err := m.snap.newManifestWriter(spec) + if err != nil { + return nil, err + } + + return &manifestClusterWriter{ + writer: writer, + path: path, + counter: counter, + fileCloser: fileCloser, + }, nil +} + +func validateManifestClusterKey(key any) error { + if key == nil { + return errors.New("manifest cluster key must be non-nil") + } + + typ := reflect.TypeOf(key) + value := reflect.ValueOf(key) + if !value.Comparable() { + return fmt.Errorf("manifest cluster key type %s is not comparable", typ) + } + + switch value.Kind() { + case reflect.Chan, reflect.Pointer, reflect.UnsafePointer: + if value.IsNil() { + return errors.New("manifest cluster key must be non-nil") + } + } + if !value.Equal(value) { + return fmt.Errorf("manifest cluster key type %s is not reflexive", typ) + } + + return nil +} + +func validateManifestClusterKeyComparable(key any) error { + if key == nil { + return errors.New("manifest cluster key must be non-nil") + } + + value := reflect.ValueOf(key) + if !value.Comparable() { + return fmt.Errorf("manifest cluster key type %s is not comparable", value.Type()) + } + + return nil +} + +// clusterManifests rewrites entries into one rolling writer per cluster key and +// partition spec. Writers stay open while entries for other keys are read so a +// later file with the same key is still written beside the earlier files. This +// assumes a producer that only reorganizes live data entries; it does not add +// new files or preserve tombstones. +func (m *manifestMergeManager) clusterManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { + // One output writer is tracked per key, so reserve space for the common + // case where each input manifest introduces a new cluster. + writers := make(map[manifestClusterKey]*manifestClusterWriter, len(manifests)) + order := make([]manifestClusterKey, 0, len(manifests)) + paths := make([]string, 0, len(manifests)) + completed := false + + defer func() { + if completed { + return + } + + for _, writer := range writers { + writer.abort() + } + for _, path := range paths { + if removeErr := m.snap.io.Remove(path); removeErr != nil { + log.Printf("Warning: failed to delete orphaned clustered manifest %s: %v", path, removeErr) + } + } + }() + + closeWriter := func(writer *manifestClusterWriter) error { + manifest, closeErr := writer.close() + if closeErr != nil { + return closeErr + } + if manifest != nil { + writer.manifests = append(writer.manifests, manifest) + } + + return nil + } + + for _, manifest := range manifests { + specID := int(manifest.PartitionSpecID()) + for entry, entryErr := range m.snap.iterManifestEntries(manifest, true) { + if entryErr != nil { + return nil, entryErr + } + + clusterValue := m.clusterBy(entry.DataFile()) + if clusterErr := validateManifestClusterKeyComparable(clusterValue); clusterErr != nil { + return nil, fmt.Errorf("cluster data file %q: %w", entry.DataFile().FilePath(), clusterErr) + } + key := manifestClusterKey{specID: specID, value: clusterValue} + writer, ok := writers[key] + if !ok { + if clusterErr := validateManifestClusterKey(clusterValue); clusterErr != nil { + return nil, fmt.Errorf("cluster data file %q: %w", entry.DataFile().FilePath(), clusterErr) + } + var openErr error + writer, openErr = m.newClusterWriter(specID) + if openErr != nil { + return nil, openErr + } + writers[key] = writer + order = append(order, key) + paths = append(paths, writer.path) + } + + // The counter advances when the Avro writer flushes a block, so a + // manifest can exceed targetSizeBytes by nearly one block. + if writer.writer != nil && writer.hasEntries && m.targetSizeBytes > 0 && writer.counter.Count >= m.targetSizeBytes { + if entryErr := closeWriter(writer); entryErr != nil { + return nil, entryErr Review Comment: **nit** — Residual `entryErr` shadow at the roll site Thread 8 asked for the writer-result error to stop borrowing the range iterator's `entryErr` name. Both writer-creation sites were changed to `openErr` (:190, :208) but the roll site still reads `if entryErr := closeWriter(writer); entryErr != nil`. It is a distinct shadowing declaration so the code is correct, but it re-introduces exactly the naming the thread flagged. Suggest `closeErr` for symmetry with `openErr`. -- 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]
