(iceberg-go) branch main updated: perf(table): flatten manifest entries after parallel reads (#1976)

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git


The following commit(s) were added to refs/heads/main by this push:
     new 877891025 perf(table): flatten manifest entries after parallel reads (#1976)
877891025 is described below

commit 87789102577ce00f050b43fa5eb9f667ad31642a
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Tue Sep 1 05:43:16 2026 +0200

    perf(table): flatten manifest entries after parallel reads (#1976)
    
    * perf(table): flatten manifest entries after parallel reads
    
    Signed-off-by: Minh Vu <vu...@gmail.com>
    
    * fix(table): satisfy lint checks for manifest flattening
    
    Signed-off-by: Minh Vu <vu...@gmail.com>
    
    ---------
    
    Signed-off-by: Minh Vu <vu...@gmail.com>
---
 table/scanner.go                             | 45 +++++++++++++--
 table/scanner_manifest_entries_bench_test.go | 72 +++++++++++++++++-------
 table/scanner_manifest_entries_test.go       | 82 ++++++++++++++++++++++++++++
 3 files changed, 175 insertions(+), 24 deletions(-)

diff --git a/table/scanner.go b/table/scanner.go
index 00a0e4ef4..aec98bfea 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -315,6 +315,41 @@ func (m *manifestEntries) merge(entries []iceberg.ManifestEntry) error {
 	return err
 }
 
+func flattenClassifiedManifestEntries(results []classifiedManifestEntries) *manifestEntries {
+	var counts [manifestEntryKindCount]int
+	for _, result := range results {
+		counts[manifestEntryData] += len(result.dataEntries)
+		counts[manifestEntryPositionalDelete] += len(result.positionalDeleteEntries)
+		counts[manifestEntryEqualityDelete] += len(result.equalityDeleteEntries)
+		counts[manifestEntryDV] += len(result.dvEntries)
+	}
+
+	flattened := &manifestEntries{
+		dataEntries:             make([]iceberg.ManifestEntry, counts[manifestEntryData]),
+		positionalDeleteEntries: make([]iceberg.ManifestEntry, counts[manifestEntryPositionalDelete]),
+		equalityDeleteEntries:   make([]iceberg.ManifestEntry, counts[manifestEntryEqualityDelete]),
+		dvEntries:               make([]iceberg.ManifestEntry, counts[manifestEntryDV]),
+	}
+
+	var offsets [manifestEntryKindCount]int
+	for _, result := range results {
+		offsets[manifestEntryData] += copy(
+			flattened.dataEntries[offsets[manifestEntryData]:], result.dataEntries,
+		)
+		offsets[manifestEntryPositionalDelete] += copy(
+			flattened.positionalDeleteEntries[offsets[manifestEntryPositionalDelete]:], result.positionalDeleteEntries,
+		)
+		offsets[manifestEntryEqualityDelete] += copy(
+			flattened.equalityDeleteEntries[offsets[manifestEntryEqualityDelete]:], result.equalityDeleteEntries,
+		)
+		offsets[manifestEntryDV] += copy(
+			flattened.dvEntries[offsets[manifestEntryDV]:], result.dvEntries,
+		)
+	}
+
+	return flattened
+}
+
 func newPartitionRecord(partitionData map[int]any, partitionType *iceberg.StructType) partitionRecord {
 	out := make(partitionRecord, len(partitionType.FieldList))
 	for i, f := range partitionType.FieldList {
@@ -1053,7 +1088,7 @@ func (scan *Scan) collectManifestEntriesWithSchema(
 	concurrencyLimit := min(scan.concurrency, len(manifestList))
 	manifestIO := newManifestIOBatch(scan.ioF, concurrencyLimit)
 
-	entries := newManifestEntries()
+	manifestResults := make([]classifiedManifestEntries, len(manifestList))
 	g, gctx := errgroup.WithContext(ctx)
 	g.SetLimit(concurrencyLimit)
 
@@ -1061,7 +1096,7 @@ func (scan *Scan) collectManifestEntriesWithSchema(
 		return buildPartitionEvaluator(specID, scan.metadata, schema, partitionFilters, scan.caseSensitive)
 	})
 
-	for _, mf := range manifestList {
+	for manifestIndex, mf := range manifestList {
 		if !scan.checkSequenceNumber(minSeqNum, mf) {
 			continue
 		}
@@ -1079,9 +1114,11 @@ func (scan *Scan) collectManifestEntriesWithSchema(
 			if err != nil {
 				return err
 			}
-			if err := entries.merge(manifestEntries); err != nil {
+			classified, err := classifyManifestEntries(manifestEntries)
+			if err != nil {
 				return err
 			}
+			manifestResults[manifestIndex] = classified
 
 			return nil
 		})
@@ -1091,7 +1128,7 @@ func (scan *Scan) collectManifestEntriesWithSchema(
 		return nil, err
 	}
 
-	return entries, nil
+	return flattenClassifiedManifestEntries(manifestResults), nil
 }
 
 // PlanFiles orchestrates the fetching and filtering of manifests, building a
diff --git a/table/scanner_manifest_entries_bench_test.go b/table/scanner_manifest_entries_bench_test.go
index 42330d7dc..71caf1493 100644
--- a/table/scanner_manifest_entries_bench_test.go
+++ b/table/scanner_manifest_entries_bench_test.go
@@ -60,31 +60,63 @@ func BenchmarkManifestEntryCollection(b *testing.B) {
 		{name: "data", content: iceberg.ManifestContentData},
 		{name: "deletes", content: iceberg.ManifestContentDeletes},
 	} {
-		for _, benchmarkCase := range benchmarkCases {
-			name := fmt.Sprintf("content=%s/manifests=%d/concurrency=%d/entries=%d",
-				workload.name, benchmarkCase.manifestCount, benchmarkCase.concurrency, benchmarkCase.entryCount)
-			b.Run(name, func(b *testing.B) {
-				entries := benchmarkManifestEntries(benchmarkCase.entryCount, workload.content)
-				b.ReportAllocs()
-				b.ResetTimer()
-				b.ReportMetric(float64(benchmarkCase.manifestCount*benchmarkCase.entryCount), "entries/op")
-
-				var result *manifestEntries
-				for range b.N {
-					result = collectManifestEntryBatches(entries, benchmarkCase.manifestCount, benchmarkCase.concurrency)
-				}
-
-				if got, want := len(result.dataEntries)+len(result.positionalDeleteEntries)+
-					len(result.equalityDeleteEntries)+len(result.dvEntries), benchmarkCase.manifestCount*benchmarkCase.entryCount; got != want {
-					b.Fatalf("collected %d entries, want %d", got, want)
-				}
-				runtime.KeepAlive(result)
-			})
+		for _, method := range []struct {
+			name    string
+			collect func([]iceberg.ManifestEntry, int, int) *manifestEntries
+		}{
+			{name: "merge", collect: collectManifestEntryBatchesWithMerge},
+			{name: "flatten", collect: collectManifestEntryBatches},
+		} {
+			for _, benchmarkCase := range benchmarkCases {
+				name := fmt.Sprintf("method=%s/content=%s/manifests=%d/concurrency=%d/entries=%d",
+					method.name, workload.name, benchmarkCase.manifestCount, benchmarkCase.concurrency, benchmarkCase.entryCount)
+				b.Run(name, func(b *testing.B) {
+					entries := benchmarkManifestEntries(benchmarkCase.entryCount, workload.content)
+					b.ReportAllocs()
+					b.ResetTimer()
+					b.ReportMetric(float64(benchmarkCase.manifestCount*benchmarkCase.entryCount), "entries/op")
+
+					var result *manifestEntries
+					for range b.N {
+						result = method.collect(entries, benchmarkCase.manifestCount, benchmarkCase.concurrency)
+					}
+
+					if got, want := len(result.dataEntries)+len(result.positionalDeleteEntries)+
+						len(result.equalityDeleteEntries)+len(result.dvEntries), benchmarkCase.manifestCount*benchmarkCase.entryCount; got != want {
+						b.Fatalf("collected %d entries, want %d", got, want)
+					}
+					runtime.KeepAlive(result)
+				})
+			}
 		}
 	}
 }
 
 func collectManifestEntryBatches(entries []iceberg.ManifestEntry, manifestCount, concurrency int) *manifestEntries {
+	results := make([]classifiedManifestEntries, manifestCount)
+	var g errgroup.Group
+	g.SetLimit(min(concurrency, manifestCount))
+	for manifestIndex := range manifestCount {
+		g.Go(func() error {
+			manifestEntries := append(make([]iceberg.ManifestEntry, 0, len(entries)), entries...)
+			classified, err := classifyManifestEntries(manifestEntries)
+			if err != nil {
+				return err
+			}
+			results[manifestIndex] = classified
+
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		panic(err)
+	}
+
+	return flattenClassifiedManifestEntries(results)
+}
+
+func collectManifestEntryBatchesWithMerge(entries []iceberg.ManifestEntry, manifestCount, concurrency int) *manifestEntries {
 	collected := newManifestEntries()
 	var g errgroup.Group
 	g.SetLimit(min(concurrency, manifestCount))
diff --git a/table/scanner_manifest_entries_test.go b/table/scanner_manifest_entries_test.go
new file mode 100644
index 000000000..10f7cb66c
--- /dev/null
+++ b/table/scanner_manifest_entries_test.go
@@ -0,0 +1,82 @@
+// 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 (
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFlattenClassifiedManifestEntriesPreservesManifestOrder(t *testing.T) {
+	snapshotID := int64(1)
+	dataOne := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, &mockDataFile{
+		path:        "data-1.parquet",
+		contentType: iceberg.EntryContentData,
+	})
+	dataTwo := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, &mockDataFile{
+		path:        "data-2.parquet",
+		contentType: iceberg.EntryContentData,
+	})
+	positionalDelete := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, &mockDataFile{
+		path:        "pos-delete.parquet",
+		contentType: iceberg.EntryContentPosDeletes,
+	})
+	equalityDelete := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, &mockDataFile{
+		path:        "eq-delete.parquet",
+		contentType: iceberg.EntryContentEqDeletes,
+	})
+	dv := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, &mockDataFile{
+		path:        "deletion-vector.puffin",
+		contentType: iceberg.EntryContentPosDeletes,
+		format:      iceberg.PuffinFile,
+	})
+
+	got := flattenClassifiedManifestEntries([]classifiedManifestEntries{
+		{
+			dataEntries:           []iceberg.ManifestEntry{dataOne},
+			equalityDeleteEntries: []iceberg.ManifestEntry{equalityDelete},
+			dvEntries:             []iceberg.ManifestEntry{dv},
+		},
+		{},
+		{dataEntries: []iceberg.ManifestEntry{dataTwo}, positionalDeleteEntries: []iceberg.ManifestEntry{positionalDelete}},
+	})
+
+	assert.Equal(t, []iceberg.ManifestEntry{dataOne, dataTwo}, got.dataEntries)
+	assert.Equal(t, []iceberg.ManifestEntry{positionalDelete}, got.positionalDeleteEntries)
+	assert.Equal(t, []iceberg.ManifestEntry{equalityDelete}, got.equalityDeleteEntries)
+	assert.Equal(t, []iceberg.ManifestEntry{dv}, got.dvEntries)
+	assert.Equal(t, len(got.dataEntries), cap(got.dataEntries))
+	assert.Equal(t, len(got.positionalDeleteEntries), cap(got.positionalDeleteEntries))
+	assert.Equal(t, len(got.equalityDeleteEntries), cap(got.equalityDeleteEntries))
+	assert.Equal(t, len(got.dvEntries), cap(got.dvEntries))
+}
+
+func TestFlattenClassifiedManifestEntriesEmpty(t *testing.T) {
+	got := flattenClassifiedManifestEntries(nil)
+
+	assert.NotNil(t, got.dataEntries)
+	assert.NotNil(t, got.positionalDeleteEntries)
+	assert.NotNil(t, got.equalityDeleteEntries)
+	assert.NotNil(t, got.dvEntries)
+	assert.Empty(t, got.dataEntries)
+	assert.Empty(t, got.positionalDeleteEntries)
+	assert.Empty(t, got.equalityDeleteEntries)
+	assert.Empty(t, got.dvEntries)
+}