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 07f52fb3e perf(table): read partition manifests concurrently (#2228)
07f52fb3e is described below
commit 07f52fb3e01421c75a2f163beda31f5ae4ba687a
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 19:25:40 2026 +0200
perf(table): read partition manifests concurrently (#2228)
* perf(table): aggregate partition manifests concurrently
* test(table): cover parallel partition manifest aggregation
* test(perf): benchmark partition metadata manifest reads
* fix(table): repair parallel partition inspection
* fix(table): honor cancellation while merging parallel manifest results
Signed-off-by: Minh Vu <[email protected]>
* test(table): cover canceled scans and deterministic partition timestamp
ties
Signed-off-by: Minh Vu <[email protected]>
* fix(table): avoid canceled manifest opens
Signed-off-by: Minh Vu <[email protected]>
* style(table): fix lint findings
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/inspect_partitions.go | 184 +++++++++++++++++-----
table/inspect_partitions_parallel_bench_test.go | 112 ++++++++++++++
table/inspect_partitions_parallel_test.go | 193 ++++++++++++++++++++++++
3 files changed, 454 insertions(+), 35 deletions(-)
diff --git a/table/inspect_partitions.go b/table/inspect_partitions.go
index c63dc4877..62f73efba 100644
--- a/table/inspect_partitions.go
+++ b/table/inspect_partitions.go
@@ -28,6 +28,8 @@ import (
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/scalar"
"github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "golang.org/x/sync/errgroup"
)
type inspectPartitionAggregate struct {
@@ -166,6 +168,10 @@ func (i InspectTable) Partitions(ctx context.Context)
(array.RecordReader, error
return rr, nil
}
+// inspectPartitionManifestConcurrency bounds concurrent manifest reads and
+// per-manifest partition aggregation state.
+const inspectPartitionManifestConcurrency = 8
+
func (i InspectTable) partitionAggregates(ctx context.Context, partitionType
*iceberg.StructType) ([]inspectPartitionAggregate, error) {
snapshot := i.tbl.Metadata().CurrentSnapshot()
if snapshot == nil {
@@ -189,46 +195,72 @@ func (i InspectTable) partitionAggregates(ctx
context.Context, partitionType *ic
aggregates := make([]*inspectPartitionAggregate, 0)
aggregateTree := newInspectPartitionAggregateTree()
- for manifest := range manifests {
- if err := ctx.Err(); err != nil {
- return nil, err
+
+ // Buffer only a bounded number of manifests. Workers group their
entries
+ // independently, then their results are merged in manifest order.
+ // This keeps equal-timestamp tie-breaking deterministic.
+ processBatch := func(batch []iceberg.ManifestFile) error {
+ if len(batch) == 1 {
+ return i.foldInspectPartitionManifest(ctx, fs,
batch[0], partitionType,
+ snapshotTimes, aggregateTree, &aggregates)
}
- for entry, err := range manifest.Entries(fs, true) {
- if err != nil {
- return nil, err
- }
- file := entry.DataFile()
- partitionValues := dataFilePartition(file)
- aggregate :=
aggregateTree.lookupPartition(partitionValues, partitionType)
- if aggregate == nil {
- partition :=
inspectCoercePartition(partitionValues, partitionType)
- record := newPartitionRecord(partition,
partitionType)
- aggregate = &inspectPartitionAggregate{
- partition:
cloneInspectPartition(partition),
- orderingKey:
inspectPartitionKey(partition),
+
+ partial := make([][]*inspectPartitionAggregate, len(batch))
+ g, gctx := errgroup.WithContext(ctx)
+ for index, manifest := range batch {
+ g.Go(func() error {
+ localTree := newInspectPartitionAggregateTree()
+
+ return i.foldInspectPartitionManifest(gctx, fs,
manifest, partitionType,
+ snapshotTimes, localTree,
&partial[index])
+ })
+ }
+ if err := g.Wait(); err != nil {
+ return err
+ }
+
+ merged := 0
+ for _, manifestAggregates := range partial {
+ for _, local := range manifestAggregates {
+ // A manifest can contain many distinct
partitions. Remain
+ // responsive to cancellation during the
ordered merge, too.
+ if merged&1023 == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
}
- aggregates = append(aggregates, aggregate)
- aggregateTree.insert(record, aggregate)
- }
+ merged++
+ aggregate :=
aggregateTree.lookupPartition(local.partition, partitionType)
+ if aggregate == nil {
+ aggregates = append(aggregates, local)
+
aggregateTree.insert(newPartitionRecord(local.partition, partitionType), local)
- switch file.ContentType() {
- case iceberg.EntryContentData:
- aggregate.dataRecordCount += file.Count()
- aggregate.dataFileCount++
- aggregate.dataFileSize += file.FileSizeBytes()
- case iceberg.EntryContentPosDeletes:
- aggregate.positionDeleteCount += file.Count()
- aggregate.positionDeleteFiles++
- case iceberg.EntryContentEqDeletes:
- aggregate.equalityDeleteCount += file.Count()
- aggregate.equalityDeleteFiles++
+ continue
+ }
+ mergeInspectPartitionAggregate(aggregate, local)
}
+ }
- if file.ContentType() == iceberg.EntryContentData ||
- file.ContentType() ==
iceberg.EntryContentPosDeletes ||
- file.ContentType() ==
iceberg.EntryContentEqDeletes {
- i.updatePartitionTimestamp(aggregate,
snapshotTimes, entry.SnapshotID(), file.SpecID())
- }
+ return nil
+ }
+
+ batch := make([]iceberg.ManifestFile, 0,
inspectPartitionManifestConcurrency)
+ for manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ batch = append(batch, manifest)
+ if len(batch) < inspectPartitionManifestConcurrency {
+ continue
+ }
+ if err := processBatch(batch); err != nil {
+ return nil, err
+ }
+ batch = batch[:0]
+ }
+ if len(batch) > 0 {
+ if err := processBatch(batch); err != nil {
+ return nil, err
}
}
@@ -241,6 +273,88 @@ func (i InspectTable) partitionAggregates(ctx
context.Context, partitionType *ic
return result, nil
}
+// foldInspectPartitionManifest performs the original partition accounting
using
+// a caller-owned tree. Each concurrent worker owns its own tree and result
slice.
+func (i InspectTable) foldInspectPartitionManifest(
+ ctx context.Context,
+ fs iceio.IO,
+ manifest iceberg.ManifestFile,
+ partitionType *iceberg.StructType,
+ snapshotTimes map[int64]int64,
+ aggregateTree *inspectPartitionAggregateTree,
+ aggregates *[]*inspectPartitionAggregate,
+) error {
+ // Avoid opening remote manifest files after the caller has canceled.
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ entryIndex := 0
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return err
+ }
+ if entryIndex&1023 == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ }
+ entryIndex++
+
+ file := entry.DataFile()
+ partitionValues := dataFilePartition(file)
+ aggregate := aggregateTree.lookupPartition(partitionValues,
partitionType)
+ if aggregate == nil {
+ partition := inspectCoercePartition(partitionValues,
partitionType)
+ record := newPartitionRecord(partition, partitionType)
+ aggregate = &inspectPartitionAggregate{
+ partition: cloneInspectPartition(partition),
+ orderingKey: inspectPartitionKey(partition),
+ }
+ *aggregates = append(*aggregates, aggregate)
+ aggregateTree.insert(record, aggregate)
+ }
+
+ switch file.ContentType() {
+ case iceberg.EntryContentData:
+ aggregate.dataRecordCount += file.Count()
+ aggregate.dataFileCount++
+ aggregate.dataFileSize += file.FileSizeBytes()
+ case iceberg.EntryContentPosDeletes:
+ aggregate.positionDeleteCount += file.Count()
+ aggregate.positionDeleteFiles++
+ case iceberg.EntryContentEqDeletes:
+ aggregate.equalityDeleteCount += file.Count()
+ aggregate.equalityDeleteFiles++
+ }
+
+ if file.ContentType() == iceberg.EntryContentData ||
+ file.ContentType() == iceberg.EntryContentPosDeletes ||
+ file.ContentType() == iceberg.EntryContentEqDeletes {
+ i.updatePartitionTimestamp(aggregate, snapshotTimes,
entry.SnapshotID(), file.SpecID())
+ }
+ }
+
+ return ctx.Err()
+}
+
+func mergeInspectPartitionAggregate(dst, src *inspectPartitionAggregate) {
+ dst.dataRecordCount += src.dataRecordCount
+ dst.dataFileCount += src.dataFileCount
+ dst.dataFileSize += src.dataFileSize
+ dst.positionDeleteCount += src.positionDeleteCount
+ dst.positionDeleteFiles += src.positionDeleteFiles
+ dst.equalityDeleteCount += src.equalityDeleteCount
+ dst.equalityDeleteFiles += src.equalityDeleteFiles
+
+ if src.lastUpdatedAt != nil &&
+ (dst.lastUpdatedAt == nil || *src.lastUpdatedAt >
*dst.lastUpdatedAt) {
+ dst.lastUpdatedAt = src.lastUpdatedAt
+ dst.lastUpdatedSnapshot = src.lastUpdatedSnapshot
+ dst.specID = src.specID
+ }
+}
+
func (i InspectTable) updatePartitionTimestamp(
aggregate *inspectPartitionAggregate,
snapshotTimes map[int64]int64,
diff --git a/table/inspect_partitions_parallel_bench_test.go
b/table/inspect_partitions_parallel_bench_test.go
new file mode 100644
index 000000000..fe15160d4
--- /dev/null
+++ b/table/inspect_partitions_parallel_bench_test.go
@@ -0,0 +1,112 @@
+// 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"
+ "github.com/stretchr/testify/require"
+)
+
+// BenchmarkInspectPartitionsParallelManifests exercises the full metadata
+// table read, including manifest decoding and partition aggregation.
+// The tracked FileIO adds 10ms latency per manifest open to model remote I/O.
+func BenchmarkInspectPartitionsParallelManifests(b *testing.B) {
+ for _, count := range []int{1, 8, 32} {
+ b.Run(fmt.Sprintf("manifests=%d", count), func(b *testing.B) {
+ tbl := benchmarkInspectPartitionTable(b, count)
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ reader, err :=
tbl.Inspect().Partitions(b.Context())
+ if err != nil {
+ b.Fatal(err)
+ }
+ reader.Release()
+ }
+ })
+ }
+}
+
+func benchmarkInspectPartitionTable(b *testing.B, manifestCount int) *Table {
+ b.Helper()
+
+ const snapshotID = int64(1)
+ sequenceNumber := int64(1)
+ spec := *iceberg.UnpartitionedSpec
+ schema := iceberg.NewSchema(0)
+ meta, err := NewMetadata(schema, &spec, UnsortedSortOrder,
"mem://inspect-bench/table", nil)
+ require.NoError(b, err)
+ builder, err := MetadataBuilderFromBase(meta, "metadata.json")
+ require.NoError(b, err)
+
+ fs := iceio.NewMemFS()
+ manifests := make([]iceberg.ManifestFile, 0, manifestCount)
+ for index := range manifestCount {
+ dataPath :=
fmt.Sprintf("mem://inspect-bench/table/data/file-%d.parquet", index)
+ dataFileBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentData, dataPath,
iceberg.ParquetFile,
+ nil, nil, nil, 100, 1_024,
+ )
+ require.NoError(b, err)
+
+ manifestPath :=
fmt.Sprintf("mem://inspect-bench/table/metadata/inspect-%d.avro", index)
+ var manifestBuf bytes.Buffer
+ manifest, err := iceberg.WriteManifest(
+ manifestPath, &manifestBuf, 2, spec, schema, snapshotID,
+ []iceberg.ManifestEntry{
+
iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
+ new(snapshotID), &sequenceNumber,
&sequenceNumber, dataFileBuilder.Build()),
+ },
+ )
+ require.NoError(b, err)
+ require.NoError(b, fs.WriteFile(manifestPath,
manifestBuf.Bytes()))
+ manifests = append(manifests, manifest)
+ }
+
+ manifestListPath :=
"mem://inspect-bench/table/metadata/manifest-list.avro"
+ var listBuf bytes.Buffer
+ require.NoError(b, iceberg.WriteManifestList(
+ 2, &listBuf, snapshotID, nil, &sequenceNumber, 0, manifests,
+ ))
+ require.NoError(b, fs.WriteFile(manifestListPath, listBuf.Bytes()))
+
+ schemaID := schema.ID
+ require.NoError(b, builder.AddSnapshot(&Snapshot{
+ SnapshotID: snapshotID,
+ SequenceNumber: sequenceNumber,
+ TimestampMs: meta.LastUpdatedMillis() + 1,
+ ManifestList: manifestListPath,
+ SchemaID: &schemaID,
+ Summary: &Summary{Operation: OpAppend},
+ }))
+ require.NoError(b, builder.SetSnapshotRef(MainBranch, snapshotID,
BranchRef))
+
+ built, err := builder.Build()
+ require.NoError(b, err)
+
+ delayed := &inspectPartitionManifestTrackingIO{IO: fs}
+
+ return New(Identifier{"db", "inspect-bench"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return delayed, nil
}, nil)
+}
diff --git a/table/inspect_partitions_parallel_test.go
b/table/inspect_partitions_parallel_test.go
new file mode 100644
index 000000000..d03e9c6c6
--- /dev/null
+++ b/table/inspect_partitions_parallel_test.go
@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "fmt"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+type inspectPartitionManifestTrackingIO struct {
+ iceio.IO
+ active atomic.Int32
+ maxActive atomic.Int32
+ failPath string
+}
+
+func (f *inspectPartitionManifestTrackingIO) Open(path string) (iceio.File,
error) {
+ if strings.Contains(path, "/metadata/inspect-") &&
strings.HasSuffix(path, ".avro") {
+ if f.failPath != "" && strings.Contains(path, f.failPath) {
+ return nil, fmt.Errorf("injected manifest read error
for %s", path)
+ }
+
+ active := f.active.Add(1)
+ for {
+ maxActive := f.maxActive.Load()
+ if active <= maxActive ||
f.maxActive.CompareAndSwap(maxActive, active) {
+ break
+ }
+ }
+ // Deliberately delay only manifest opens to expose concurrent
I/O
+ // without depending on timing assertions.
+ time.Sleep(10 * time.Millisecond)
+ f.active.Add(-1)
+ }
+
+ return f.IO.Open(path)
+}
+
+func inspectParallelPartitionManifestTable(t *testing.T, count int) *Table {
+ t.Helper()
+
+ spec := partitionedSpec()
+ const snapshotID = int64(1)
+ sequenceNumber := int64(1)
+ manifests := make([]inspectManifestSpec, 0, count)
+ for index := range count {
+ files := []struct {
+ partition int32
+ count int64
+ }{
+ {partition: 7, count: 3},
+ {partition: 8, count: 5},
+ }
+ entries := make([]iceberg.ManifestEntry, 0, len(files))
+ for _, f := range files {
+ path :=
fmt.Sprintf("mem://default/table-location/data/parallel-%d-%d.parquet", index,
f.partition)
+ dataFile := newTestDataFileWithCount(t, spec, path,
map[int]any{1000: f.partition}, f.count)
+ entries = append(entries, iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, new(snapshotID),
+ &sequenceNumber, &sequenceNumber, dataFile,
+ ))
+ }
+ manifests = append(manifests, inspectManifestSpec{
+ content: iceberg.ManifestContentData,
+ entries: entries,
+ })
+ }
+
+ return inspectFilesTableWithManifests(t, spec, manifests...)
+}
+
+func inspectWithTrackingIO(t *testing.T, tbl *Table) (*Table,
*inspectPartitionManifestTrackingIO) {
+ t.Helper()
+
+ fs, err := tbl.FS(t.Context())
+ require.NoError(t, err)
+
+ tracked := &inspectPartitionManifestTrackingIO{IO: fs}
+ wrapped := New(Identifier{"db", "parallel-inspect"}, tbl.Metadata(),
tbl.MetadataLocation(),
+ func(context.Context) (iceio.IO, error) { return tracked, nil
}, nil)
+
+ return wrapped, tracked
+}
+
+func TestInspectPartitionsParallelManifestAggregation(t *testing.T) {
+ for _, count := range []int{1, 8, 17} {
+ t.Run(fmt.Sprintf("manifests=%d", count), func(t *testing.T) {
+ tbl, tracked := inspectWithTrackingIO(t,
inspectParallelPartitionManifestTable(t, count))
+
+ reader, err := tbl.Inspect().Partitions(t.Context())
+ require.NoError(t, err)
+ defer reader.Release()
+
+ record := collectRecord(t, reader)
+ defer record.Release()
+ require.EqualValues(t, 2, record.NumRows())
+
+ partitions :=
record.Column(0).(*array.Struct).Field(0).(*array.Int32)
+ recordCounts := record.Column(2).(*array.Int64)
+ fileCounts := record.Column(3).(*array.Int32)
+ lastSnapshot := record.Column(10).(*array.Int64)
+ for index, partition := range []int32{7, 8} {
+ require.Equal(t, partition,
partitions.Value(index))
+ wantRows := int64(count) * 3
+ if partition == 8 {
+ wantRows = int64(count) * 5
+ }
+ require.Equal(t, wantRows,
recordCounts.Value(index))
+ require.EqualValues(t, count,
fileCounts.Value(index))
+ require.EqualValues(t, 1,
lastSnapshot.Value(index))
+ }
+ require.LessOrEqual(t, tracked.maxActive.Load(),
int32(inspectPartitionManifestConcurrency))
+ if count > 1 {
+ require.GreaterOrEqual(t,
tracked.maxActive.Load(), int32(2))
+ }
+ })
+ }
+}
+
+func TestInspectPartitionsParallelManifestReadFailure(t *testing.T) {
+ tbl, tracked := inspectWithTrackingIO(t,
inspectParallelPartitionManifestTable(t, 8))
+ tracked.failPath = "inspect-3.avro"
+
+ reader, err := tbl.Inspect().Partitions(t.Context())
+ require.ErrorContains(t, err, "injected manifest read error")
+ require.Nil(t, reader)
+ require.EqualValues(t, 0, tracked.active.Load())
+}
+
+func TestInspectPartitionsParallelCancelledContext(t *testing.T) {
+ tbl, tracked := inspectWithTrackingIO(t,
inspectParallelPartitionManifestTable(t, 8))
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ reader, err := tbl.Inspect().Partitions(ctx)
+ require.ErrorIs(t, err, context.Canceled)
+ require.Nil(t, reader)
+ require.Zero(t, tracked.maxActive.Load())
+}
+
+func TestMergeInspectPartitionAggregateTimestampTies(t *testing.T) {
+ const timestamp = int64(123000)
+ firstSnapshot := int64(10)
+ secondSnapshot := int64(20)
+ dst := &inspectPartitionAggregate{
+ dataRecordCount: 3, dataFileCount: 1, specID: 1,
+ lastUpdatedAt: new(timestamp), lastUpdatedSnapshot:
&firstSnapshot,
+ }
+ sameTime := &inspectPartitionAggregate{
+ dataRecordCount: 5, dataFileCount: 2, specID: 2,
+ lastUpdatedAt: new(timestamp), lastUpdatedSnapshot:
&secondSnapshot,
+ }
+ mergeInspectPartitionAggregate(dst, sameTime)
+
+ // The earlier manifest wins a timestamp tie, but counts still add up.
+ require.EqualValues(t, 8, dst.dataRecordCount)
+ require.EqualValues(t, 3, dst.dataFileCount)
+ require.EqualValues(t, 1, dst.specID)
+ require.Equal(t, firstSnapshot, *dst.lastUpdatedSnapshot)
+
+ newerTime := timestamp + 1
+ newer := &inspectPartitionAggregate{
+ specID: 3, lastUpdatedAt: &newerTime, lastUpdatedSnapshot:
&secondSnapshot,
+ }
+ mergeInspectPartitionAggregate(dst, newer)
+ require.EqualValues(t, 3, dst.specID)
+ require.Equal(t, newerTime, *dst.lastUpdatedAt)
+ require.Equal(t, secondSnapshot, *dst.lastUpdatedSnapshot)
+}
