laskoviymishka commented on code in PR #1940: URL: https://github.com/apache/iceberg-go/pull/1940#discussion_r3893141849
########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +func (m *manifestMergeManager) clusterManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) { Review Comment: Not a blocker, the committed output is correct either way. But every OCC retry reruns the whole clustering pass from scratch (processManifests, mergeManifests, then here), so under contention we write a full new set of clustered manifests on each attempt. Java's BaseRewriteManifests avoids that with requiresRewrite: if the manifests it already processed are all still present in the fresh parent, it reuses the previously written output instead of re-clustering. The extra I/O aside, there's a subtler divergence. If clusterBy is non-deterministic, Java locks in the first attempt's layout when nothing was displaced, but we'd produce a fresh clustering each retry. I'd at least decide explicitly whether to mirror requiresRewrite here or document that clusterBy must be deterministic. wdyt? ########## table/rewrite_manifests_bench_test.go: ########## @@ -0,0 +1,245 @@ +// 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 ( + "bytes" + "context" + "fmt" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// BenchmarkManifestMergeModes compares the existing size-only merge with the +// cluster-by path on the same 64 one-entry manifests. Both modes use eight +// output-sized groups and a single merge worker so the benchmark focuses on +// entry routing and manifest writing rather than scheduling. +func BenchmarkManifestMergeModes(b *testing.B) { + for _, cluster := range []bool{false, true} { + name := "BySize" + if cluster { + name = "ByClusterKey" + } + b.Run(name, func(b *testing.B) { + benchmarkManifestMergeMode(b, cluster) + }) + } +} + +func benchmarkManifestMergeMode(b *testing.B, cluster bool) { + spec := iceberg.NewPartitionSpec() + schema := simpleSchema() + mem := newMemIO(1<<30, errLimitedWrite) + meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "table-location", nil) + if err != nil { + b.Fatal(err) + } + tbl := New(Identifier{"db", "benchmark"}, meta, "metadata.json", + func(context.Context) (iceio.IO, error) { return mem, nil }, nil) + txn := tbl.NewTransaction() + prod := newRewriteManifestsProducer(txn, mem, iceberg.Properties{}, rewriteManifestsCfg{}) + + const inputCount = 64 + const clusterCount = 8 + inputSnapshotID := int64(1) + inputSequenceNumber := int64(1) + manifests := make([]iceberg.ManifestFile, 0, inputCount) + clusterKeys := make(map[string]int, inputCount) + var maxLength int64 + for i := range inputCount { + path := fmt.Sprintf("file://data-%d.parquet", i) + builder, err := iceberg.NewDataFileBuilder( + spec, iceberg.EntryContentData, path, iceberg.ParquetFile, + nil, nil, nil, 1, 100, + ) + if err != nil { + b.Fatal(err) + } + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &inputSnapshotID, &inputSequenceNumber, nil, builder.Build()) + manifestPath := fmt.Sprintf("table-location/metadata/input-%d.avro", i) + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &buf, 2, spec, schema, inputSnapshotID, []iceberg.ManifestEntry{entry}) + if err != nil { + b.Fatal(err) + } + if err := mem.WriteFile(manifestPath, buf.Bytes()); err != nil { + b.Fatal(err) + } + manifests = append(manifests, manifest) + clusterKeys[path] = i % clusterCount + maxLength = max(maxLength, manifest.Length()) + } + + mgr := manifestMergeManager{ + targetSizeBytes: 8 * maxLength, + mergeEnabled: true, + mergeConcurrency: 1, + snap: prod, + } + if cluster { + mgr.clusterBy = func(df iceberg.DataFile) any { return clusterKeys[df.FilePath()] } + } + + b.ResetTimer() + for range b.N { + output, err := mgr.mergeManifests(manifests) + if err != nil { + b.Fatal(err) + } + b.StopTimer() + for _, manifest := range output { + if err := mem.Remove(manifest.FilePath()); err != nil { + b.Fatal(err) + } + } + b.StartTimer() + } +} + +// BenchmarkManifestPruningModes measures the read-side reason for clustering. +// The input manifests interleave 32 partition values. A size-only merge keeps +// that mix in every output group, while cluster-by partition value lets the +// manifest evaluator reject unrelated groups before opening their entries. +func BenchmarkManifestPruningModes(b *testing.B) { + for _, cluster := range []bool{false, true} { + name := "BySize" + if cluster { + name = "ByClusterKey" + } + b.Run(name, func(b *testing.B) { + benchmarkManifestPruningMode(b, cluster) + }) + } +} + +func benchmarkManifestPruningMode(b *testing.B, cluster bool) { + b.Helper() + + const ( + inputCount = 512 + clusterCount = 32 + targetValue = int32(7) + ) + + spec := partitionedSpec() + schema := simpleSchema() + mem := newMemIO(1<<30, errLimitedWrite) + meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "table-location", nil) + if err != nil { + b.Fatal(err) + } + tbl := New(Identifier{"db", "benchmark-pruning"}, meta, "metadata.json", + func(context.Context) (iceio.IO, error) { return mem, nil }, nil) + prod := newRewriteManifestsProducer(tbl.NewTransaction(), mem, iceberg.Properties{}, rewriteManifestsCfg{}) + + inputSnapshotID := int64(1) + inputSequenceNumber := int64(1) + manifests := make([]iceberg.ManifestFile, 0, inputCount) + for i := range inputCount { + partitionValue := int32((i / 8) % (clusterCount - 1)) + if partitionValue >= targetValue { + partitionValue++ + } + if i%8 == 0 { + // Put the queried value in every size-only input group. The + // size-only output therefore cannot prune any group, while the + // clustered output keeps these entries together. + partitionValue = targetValue + } + builder, err := iceberg.NewDataFileBuilder( + spec, iceberg.EntryContentData, + fmt.Sprintf("file://data-%d.parquet", i), iceberg.ParquetFile, + map[int]any{1000: partitionValue}, + nil, nil, 1, 100, + ) + if err != nil { + b.Fatal(err) + } + entry := iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, + &inputSnapshotID, + &inputSequenceNumber, + nil, + builder.Build(), + ) + + manifestPath := fmt.Sprintf("table-location/metadata/input-%d.avro", i) + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest( + manifestPath, &buf, 2, spec, schema, inputSnapshotID, + []iceberg.ManifestEntry{entry}, + ) + if err != nil { + b.Fatal(err) + } + if err := mem.WriteFile(manifestPath, buf.Bytes()); err != nil { + b.Fatal(err) + } + manifests = append(manifests, manifest) + } + + var maxLength int64 + for _, manifest := range manifests { + maxLength = max(maxLength, manifest.Length()) + } + mgr := manifestMergeManager{ + targetSizeBytes: 8 * maxLength, + mergeEnabled: true, + mergeConcurrency: 1, + snap: prod, + } + if cluster { + mgr.clusterBy = func(df iceberg.DataFile) any { + return df.Partition()[1000] + } + } + merged, err := mgr.mergeManifests(manifests) + if err != nil { + b.Fatal(err) + } + + scan := tbl.Scan(WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), targetValue))) + filtered, err := scan.filterManifestsWithSchema(merged, schema, &scanMetricsAccumulator{}) + if err != nil { + b.Fatal(err) + } + wantSelected := len(filtered) + if cluster { + if wantSelected >= len(merged) { + b.Fatalf("clustered layout selected %d/%d manifests, want pruning", wantSelected, len(merged)) Review Comment: This gate is the only place we assert that clustering actually lets the evaluator prune manifests, which is the read-side payoff the whole feature is for, and it only runs under `go test -bench`. Regular `go test ./...` skips benchmark bodies, so CI never checks it. The unit tests confirm entries share a key, but not that pruning follows. I'd lift the clustered-vs-size-only filterManifestsWithSchema comparison into a normal Test on a smaller dataset asserting `len(selected) < len(total)` for the clustered layout, so the guarantee is enforced in CI. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +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 := validateManifestClusterKey(clusterValue); clusterErr != nil { Review Comment: validateManifestClusterKey runs the full reflect pass (TypeOf, ValueOf, Comparable, the kind switch, and the reflexivity Equal) on every entry, but once a key is in `writers` we've already proven it's a safe map key. On a table with millions of entries across a handful of clusters that's millions of reflect round-trips where a dozen would do. We still need the comparability guard before `writers[key]` to avoid the panic, but that can be the cheap half: `reflect.ValueOf(clusterValue).Comparable()` per entry, with the full validateManifestClusterKey (including the NaN reflexivity check) deferred to the `!ok` new-key branch. wdyt? ########## table/snapshot_producers.go: ########## @@ -501,6 +502,9 @@ func (m *manifestMergeManager) mergeManifests(manifests []iceberg.ManifestFile) if !m.mergeEnabled || len(manifests) == 0 { return manifests, nil } + if m.clusterBy != nil { Review Comment: This bails before groupBySpec and minCountToMerge. Correct for RewriteManifests since it pins minCountToMerge=1, but manifestMergeManager is a shared struct, so a future caller that sets both clusterBy and minCountToMerge would have the latter silently dropped. A short comment that clustering replaces bin-packing (minCountToMerge included) rather than composing with it would save someone that surprise. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +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)) Review Comment: Worth documenting here: one writer stays open per distinct (specID, clusterKey) for the entire pass, so a high-cardinality clusterBy (say, keying on file path) keeps that many files open against the object store at once and can hit fd or handle limits. Java's clusterBy has the same one-in-flight-file-per-key shape, so I'd just note the cardinality expectation in the godoc rather than change behavior. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +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 := validateManifestClusterKey(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 { + writer, entryErr = m.newClusterWriter(specID) + if entryErr != nil { + return nil, entryErr + } + writers[key] = writer + order = append(order, key) + paths = append(paths, writer.path) + } + + if writer.writer != nil && writer.hasEntries && m.targetSizeBytes > 0 && writer.counter.Count >= m.targetSizeBytes { Review Comment: counter.Count only advances when the Avro writer flushes a block, not per entry (the createManifest comment at snapshot_producers.go:422 calls this out), so this fires at block granularity and a manifest can overshoot targetSizeBytes by close to a full block before it rolls. Small for a 128MB target, but worth a line in the godoc. Related: TestRewriteManifestsClusterByRollsAtTargetSize passes with targetSizeBytes=1 only because the Avro header already clears 1 byte before any entry, so it isn't really exercising a mid-stream roll. A test with a target that trips on the block boundary would be a more honest signal. ########## table/rewrite_manifests_bench_test.go: ########## @@ -0,0 +1,245 @@ +// 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 ( + "bytes" + "context" + "fmt" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// BenchmarkManifestMergeModes compares the existing size-only merge with the +// cluster-by path on the same 64 one-entry manifests. Both modes use eight +// output-sized groups and a single merge worker so the benchmark focuses on +// entry routing and manifest writing rather than scheduling. +func BenchmarkManifestMergeModes(b *testing.B) { + for _, cluster := range []bool{false, true} { + name := "BySize" + if cluster { + name = "ByClusterKey" + } + b.Run(name, func(b *testing.B) { + benchmarkManifestMergeMode(b, cluster) + }) + } +} + +func benchmarkManifestMergeMode(b *testing.B, cluster bool) { Review Comment: benchmarkManifestPruningMode opens with b.Helper() but this identically-shaped sibling doesn't, so a b.Fatal here points inside the helper instead of at the b.Run callsite. Adding b.Helper() as the first statement lines the two up. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +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 := validateManifestClusterKey(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 { + writer, entryErr = m.newClusterWriter(specID) Review Comment: Small one: entryErr is the range iterator's error variable, and reusing it for the writer-creation result conflates "iterator failed" with "writer creation failed". Harmless today since the range rebinds it each iteration, but a future edit between here and the next entryErr check could end up testing the wrong error. I'd give this its own var. ########## table/rewrite_manifests_test.go: ########## @@ -163,6 +164,159 @@ func TestRewriteManifests(t *testing.T) { assert.Equal(t, wantFiles, activeFiles(t, after), "rewrite must preserve the data file count") } +// TestRewriteManifestsClusterBy keeps files with the same user-provided key in +// separate manifests, even when their input manifests were interleaved. +func TestRewriteManifestsClusterBy(t *testing.T) { Review Comment: Two combos I'd add alongside this while we're here. A conflict+retry in clustering mode: clustering rewrites everything as new files each attempt, so the superseded-generation cleanup is leaned on harder here than in bin-packing, and none of the existing OCC-retry tests pass WithRewriteManifestClusterBy (this ties into the re-cluster note on clusterManifests). And clusterBy combined with a predicate or specID: both narrow r.eligible before clustering and are publicly supported, but the kept-vs-rewritten split against the file-count guard is currently untested. ########## table/rewrite_manifests_cluster.go: ########## @@ -0,0 +1,224 @@ +// 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 +} + +// 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. +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 := validateManifestClusterKey(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 { + writer, entryErr = m.newClusterWriter(specID) + if entryErr != nil { + return nil, entryErr + } + writers[key] = writer + order = append(order, key) + paths = append(paths, writer.path) + } + + 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 { + continue + } + if err := closeWriter(writer); err != nil { + return nil, err + } + } + + result := make([]iceberg.ManifestFile, 0) Review Comment: len(order) is a safe lower-bound capacity here since each key contributes at least one manifest, so `make([]iceberg.ManifestFile, 0, len(order))` skips the early regrows. -- 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]
