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 e7f50c3a2 perf(compaction): reduce split task normalization work
(#2377)
e7f50c3a2 is described below
commit e7f50c3a2d6fa9a4bdd8f03e85c91bc6e88c1787
Author: Minh Vu <[email protected]>
AuthorDate: Thu Nov 5 21:10:13 2026 +0100
perf(compaction): reduce split task normalization work (#2377)
* perf(compaction): reduce split task normalization work
Signed-off-by: Minh Vu <[email protected]>
* test(compaction): address split normalization review
Move the shared-delete-slice acceptance case alongside file-grain tests,
remove redundant test coverage, and clarify the slice shortcut comment.
Signed-off-by: Minh Vu <[email protected]>
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/compaction/compaction.go | 36 +++++++--
table/compaction/compaction_file_grain_test.go | 18 +++++
.../compaction_normalization_perf_test.go | 85 ++++++++++++++++++++++
3 files changed, 133 insertions(+), 6 deletions(-)
diff --git a/table/compaction/compaction.go b/table/compaction/compaction.go
index 1af817f3c..7f4411620 100644
--- a/table/compaction/compaction.go
+++ b/table/compaction/compaction.go
@@ -242,20 +242,22 @@ func (cfg Config) PlanCompaction(tasks
[]table.FileScanTask) (Plan, error) {
// selecting, or packing them. Keep the first task's delete and lineage
// metadata while replacing its range with the complete file range.
wholeFileTasks := make([]table.FileScanTask, 0, len(tasks))
- seenFiles := make(map[string]table.FileScanTask, len(tasks))
+ // Store indexes rather than task copies: a FileScanTask contains
several
+ // interfaces and slices, so a map value copy is expensive at scale.
+ seenFiles := make(map[string]int, len(tasks))
for _, task := range tasks {
if task.File == nil {
return Plan{}, fmt.Errorf("%w: compaction task has no
data file", table.ErrInvalidOperation)
}
path := task.File.FilePath()
- if first, ok := seenFiles[path]; ok {
- if !sameCompactionMetadata(first, task) {
+ if index, ok := seenFiles[path]; ok {
+ if !sameCompactionMetadata(wholeFileTasks[index], task)
{
return Plan{}, fmt.Errorf("%w: split tasks for
%q have different delete or lineage metadata", table.ErrInvalidOperation, path)
}
continue
}
- seenFiles[path] = task
+ seenFiles[path] = len(wholeFileTasks)
task.Start = 0
task.Length = task.File.FileSizeBytes()
task.Residual = nil
@@ -344,15 +346,37 @@ func (cfg Config) PlanCompaction(tasks
[]table.FileScanTask) (Plan, error) {
return plan, nil
}
+// sameCompactionOptionalInt64 compares the values, not pointer identities:
+// independently decoded scan tasks may have distinct pointers.
+func sameCompactionOptionalInt64(a, b *int64) bool {
+ if a == nil || b == nil {
+ return a == b
+ }
+
+ return *a == *b
+}
+
func sameCompactionMetadata(a, b table.FileScanTask) bool {
equalDataFiles := func(a, b []iceberg.DataFile) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ // Scan planning commonly reuses immutable delete-file slices
across
+ // splits. The same backing array and length prove identical
elements,
+ // regardless of capacity, without a deep comparison.
+ if len(a) == 0 || &a[0] == &b[0] {
+ return true
+ }
+
+ // Keep deep value equality for independently decoded delete
lists.
+ // DataFile is an interface and may contain an uncomparable
value.
return slices.EqualFunc(a, b, func(a, b iceberg.DataFile) bool {
return reflect.DeepEqual(a, b)
})
}
- return reflect.DeepEqual(a.FirstRowID, b.FirstRowID) &&
- reflect.DeepEqual(a.DataSequenceNumber, b.DataSequenceNumber) &&
+ return sameCompactionOptionalInt64(a.FirstRowID, b.FirstRowID) &&
+ sameCompactionOptionalInt64(a.DataSequenceNumber,
b.DataSequenceNumber) &&
equalDataFiles(a.DeleteFiles, b.DeleteFiles) &&
equalDataFiles(a.EqualityDeleteFiles, b.EqualityDeleteFiles) &&
equalDataFiles(a.DeletionVectorFiles, b.DeletionVectorFiles)
diff --git a/table/compaction/compaction_file_grain_test.go
b/table/compaction/compaction_file_grain_test.go
index 40a57193c..ff335646c 100644
--- a/table/compaction/compaction_file_grain_test.go
+++ b/table/compaction/compaction_file_grain_test.go
@@ -74,6 +74,24 @@ func TestPlanCompactionCountsWholeFiles(t *testing.T) {
require.Equal(t, original, tasks, "normalization must not mutate caller
tasks")
}
+func TestPlanCompactionAcceptsSharedSplitDeleteFiles(t *testing.T) {
+ file := newDataFile("shared.parquet", 10)
+ first := makeTask(file, 0, 0)
+ first.DeleteFiles = []iceberg.DataFile{newDeleteFile("a.pos"),
newDeleteFile("b.pos")}
+ first.Length = file.size / 2
+ second := first
+ second.Start, second.Length = first.Length, file.size-first.Length
+
+ cfg := compaction.DefaultConfig()
+ cfg.MinInputFiles = 1
+ plan, err := cfg.PlanCompaction([]table.FileScanTask{first, second})
+ require.NoError(t, err)
+ require.Equal(t, 1, plan.TotalInputFiles)
+ require.Len(t, plan.Groups, 1)
+ require.Len(t, plan.Groups[0].Tasks, 1)
+ require.Equal(t, 2, plan.Groups[0].DeleteFileCount)
+}
+
func TestPlanCompactionRejectsConflictingSplitMetadata(t *testing.T) {
for _, tc := range []struct {
name string
diff --git a/table/compaction/compaction_normalization_perf_test.go
b/table/compaction/compaction_normalization_perf_test.go
new file mode 100644
index 000000000..6d302e05d
--- /dev/null
+++ b/table/compaction/compaction_normalization_perf_test.go
@@ -0,0 +1,85 @@
+// 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 compaction_test
+
+import (
+ "fmt"
+ "slices"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/compaction"
+)
+
+// BenchmarkPlanCompactionSplitNormalization measures the full planner with
+// right-sized files. It excludes partition packing by configuring all files
+// as non-candidates, so the workload focuses on split normalization and
+// metadata consistency checks. Input construction is outside the timer.
+func BenchmarkPlanCompactionSplitNormalization(b *testing.B) {
+ for _, tc := range []struct {
+ files int
+ splits int
+ deletes int
+ copiedDeletes bool
+ }{
+ {files: 20_000, splits: 4},
+ {files: 5_000, splits: 8, deletes: 8},
+ {files: 3_000, splits: 8, deletes: 64},
+ {files: 5_000, splits: 8, deletes: 16, copiedDeletes: true},
+ } {
+ label := fmt.Sprintf("files=%d/splits=%d/deletes=%d/copied=%t",
tc.files, tc.splits, tc.deletes, tc.copiedDeletes)
+ b.Run(label, func(b *testing.B) {
+ tasks := make([]table.FileScanTask, 0,
tc.files*tc.splits)
+ for i := range tc.files {
+ file :=
newDataFile(fmt.Sprintf("file-%d.parquet", i), 500)
+ deletes := make([]iceberg.DataFile, tc.deletes)
+ for j := range deletes {
+ deletes[j] =
newDeleteFile(fmt.Sprintf("file-%d-delete-%d.pos", i, j))
+ }
+ original := table.FileScanTask{File: file,
Length: file.size, DeleteFiles: deletes}
+ for split := range tc.splits {
+ task := original
+ task.Start = int64(split) * file.size /
int64(tc.splits)
+ task.Length = file.size /
int64(tc.splits)
+ if split == tc.splits-1 {
+ task.Length = file.size -
task.Start
+ }
+ if tc.copiedDeletes {
+ task.DeleteFiles =
slices.Clone(deletes)
+ }
+ tasks = append(tasks, task)
+ }
+ }
+
+ cfg := compaction.DefaultConfig()
+ cfg.DeleteFileThreshold = 1_000
+ cfg.DeleteRatioThreshold = 0
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ plan, err := cfg.PlanCompaction(tasks)
+ if err != nil {
+ b.Fatal(err)
+ }
+ compactionPlanBenchmarkSink = plan
+ }
+ })
+ }
+}
