You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "fallintoplace (via GitHub)" <gi...@apache.org> on 2026/08/28 11:56:30 UTC
[PR] feat(table): add RewriteManifests clustering [iceberg-go]
fallintoplace opened a new pull request, #1940:
URL: https://github.com/apache/iceberg-go/pull/1940
## What changed
- add `WithRewriteManifestClusterBy` for opt-in clustering
- group live data entries by **cluster key + partition spec**
- roll each key at the configured manifest target size
- keep the default size-only path unchanged
- clean open output manifests when a callback or writer fails
## Why
Clustering can keep files with the same query-relevant key together. This can reduce unrelated manifests during scan planning.
## Benchmark
Command: `go test ./table -run '^$' -bench '^BenchmarkManifestMergeModes$' -benchmem -benchtime=1s -count=5`\n\nApple M1 Pro, 64 one-entry input manifests, 8 output groups:\n\n- size-only: **16.85–16.94 ms/op**\n- cluster-by: **17.10–17.26 ms/op**\n- both: about **192.9k allocs/op**\n\nThe small rewrite-time cost is expected. The goal of this option is better manifest locality for reads.\n\n## Tests\n\n- `go test ./table/...`\n- `go vet ./table/...`\n- `go test -race ./table -run 'TestRewriteManifests(ClusterBy|CleansOrphansOnInvalidClusterKey)' -count=1`\n
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #1940:
URL: https://github.com/apache/iceberg-go/pull/1940
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1940:
URL: https://github.com/apache/iceberg-go/pull/1940#discussion_r3883191183
##########
table/rewrite_manifests_cluster.go:
##########
@@ -0,0 +1,221 @@
+// 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)
+ if !typ.Comparable() {
Review Comment:
[P1] Reject cluster keys that cannot safely round-trip through the writer map
`reflect.Type.Comparable()` only checks the static type. For example, `struct{ Value any }{Value: []int{1}}` passes this check and then `writers[key]` panics with `hash of unhashable type: []int`. A `math.NaN()` key also passes, but because it is not equal to itself the inserted writer cannot be retrieved during finalization, causing a nil-pointer panic at line 205. I reproduced both through the exported `Transaction.RewriteManifests` API; NaN is a realistic value when clustering by an identity float partition. Please validate the value itself (for example, `reflect.Value.Comparable()` plus a reflexivity check) and add public-API regressions for both cases.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1940:
URL: https://github.com/apache/iceberg-go/pull/1940#discussion_r3937421844
##########
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 closeErr := closeWriter(writer); closeErr != nil {
+ return nil, closeErr
+ }
+ }
+ 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** — `writer.abort()` before the error return is redundant with the deferred cleanup
The deferred cleanup at lines 145-157 iterates `for _, writer := range writers { writer.abort() }`, and this writer is a member of `writers`. `abort()` is idempotent (it nils both `w.writer` and `w.fileCloser` under nil guards), so the explicit call is a harmless no-op duplicate. Verified the defer alone is sufficient: the forced-branch probe reported `orphans=[]` with all removal performed by the deferred path loop. Not worth a round-trip on its own.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add RewriteManifests clustering [iceberg-go]
Posted by "fallintoplace (via GitHub)" <gi...@apache.org>.
fallintoplace closed pull request #1940: feat(table): add RewriteManifests clustering
URL: https://github.com/apache/iceberg-go/pull/1940
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org