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 7a2acb510 perf(compaction): index FIFO first-fit bins for large plans
(#2212)
7a2acb510 is described below
commit 7a2acb510141258044b45e5fcff7693cc66f6d73
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 19 22:00:46 2026 +0200
perf(compaction): index FIFO first-fit bins for large plans (#2212)
* perf(compaction): index FIFO first-fit bins for large plans
* style: address Go lint for indexed FIFO packing
* style: address Go lint for indexed FIFO packing
---
internal/packing_indexed.go | 183 +++++++++++++++++++++++++++++++++++++++
internal/packing_indexed_test.go | 129 +++++++++++++++++++++++++++
internal/utils.go | 18 +++-
3 files changed, 328 insertions(+), 2 deletions(-)
diff --git a/internal/packing_indexed.go b/internal/packing_indexed.go
new file mode 100644
index 000000000..0cac74057
--- /dev/null
+++ b/internal/packing_indexed.go
@@ -0,0 +1,183 @@
+// 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 internal
+
+import (
+ "iter"
+ "math"
+)
+
+// Large FIFO plans spend most of their time searching the open bins.
+// Index only when there are enough items to amortize the tree and ring buffer.
+// Smaller plans, largest-bin-first packing, and potentially overflowing
+// weights keep the existing linear implementation.
+func (s *SlicePacker[T]) shouldUseIndexedFIFO(itemCount int) bool {
+ return !s.LargestBinFirst &&
+ s.Lookback >= 32 && s.Lookback <= 8192 &&
+ itemCount >= 2*s.Lookback &&
+ s.TargetWeight >= 0 && s.TargetWeight <= math.MaxInt64/2
+}
+
+// packingWeightIndex tracks the minimum weight in each subtree. Finding the
+// first bin that can accept an item then takes O(log lookback) instead of
+// scanning up to lookback open bins.
+type packingWeightIndex struct {
+ size int
+ weights []int64
+}
+
+func newPackingWeightIndex(capacity int) packingWeightIndex {
+ size := 1
+ for size < capacity {
+ size *= 2
+ }
+
+ weights := make([]int64, 2*size)
+ for i := range weights {
+ weights[i] = math.MaxInt64
+ }
+
+ return packingWeightIndex{size: size, weights: weights}
+}
+
+func (idx *packingWeightIndex) update(bin int, weight int64) {
+ node := idx.size + bin
+ idx.weights[node] = weight
+
+ for node /= 2; node > 0; node /= 2 {
+ idx.weights[node] = min(idx.weights[node*2],
idx.weights[node*2+1])
+ }
+}
+
+func (idx *packingWeightIndex) firstFit(start, end int, maxWeight int64) int {
+ return idx.firstFitNode(1, 0, idx.size, start, end, maxWeight)
+}
+
+func (idx *packingWeightIndex) firstFitNode(node, lo, hi, start, end int,
maxWeight int64) int {
+ if lo >= end || hi <= start || idx.weights[node] > maxWeight {
+ return -1
+ }
+ if hi-lo == 1 {
+ return lo
+ }
+
+ mid := (lo + hi) / 2
+ if bin := idx.firstFitNode(node*2, lo, mid, start, end, maxWeight); bin
>= 0 {
+ return bin
+ }
+
+ return idx.firstFitNode(node*2+1, mid, hi, start, end, maxWeight)
+}
+
+// indexedFIFOPackingIterator preserves first-fit selection and FIFO eviction.
+// The ring keeps bin order stable when the oldest bin is emitted. The index is
+// built lazily once 32 bins are open so easy-to-pack inputs remain
inexpensive.
+func indexedFIFOPackingIterator[T any](
+ input iter.Seq[T], targetWeight int64, lookback int, weightFunc func(T)
int64,
+) iter.Seq[[]T] {
+ capacity := lookback + 1
+ bins := make([]Bin[T], capacity)
+
+ var weights packingWeightIndex
+ head, openBins := 0, 0
+ safeToIndex, indexed := true, false
+
+ return func(yield func([]T) bool) {
+ for item := range input {
+ weight := weightFunc(item)
+ if weight < 0 || weight > targetWeight {
+ // Preserve Bin.CanAdd's signed arithmetic for
unusual inputs.
+ // The indexed path assumes non-negative,
bounded weights.
+ safeToIndex = false
+ indexed = false
+ }
+
+ bin := -1
+ if indexed {
+ maxBinWeight := targetWeight - weight
+ bin = weights.firstFit(head, capacity,
maxBinWeight)
+ if bin < 0 && head > 0 {
+ bin = weights.firstFit(0, head,
maxBinWeight)
+ }
+ } else {
+ for i := range openBins {
+ j := (head + i) % capacity
+ if bins[j].CanAdd(weight) {
+ bin = j
+
+ break
+ }
+ }
+ }
+
+ if bin >= 0 {
+ bins[bin].Add(item, weight)
+ if indexed {
+ weights.update(bin, bins[bin].Weight())
+ }
+
+ continue
+ }
+
+ slot := (head + openBins) % capacity
+ bins[slot] = Bin[T]{targetWeight: targetWeight}
+ bins[slot].Add(item, weight)
+ if indexed {
+ weights.update(slot, weight)
+ }
+ openBins++
+
+ if safeToIndex && !indexed && openBins == 32 {
+ weights = newPackingWeightIndex(capacity)
+ for i := range openBins {
+ j := (head + i) % capacity
+ weights.update(j, bins[j].Weight())
+ }
+ indexed = true
+ }
+
+ if openBins > lookback {
+ out := bins[head].items
+ bins[head] = Bin[T]{}
+ if indexed {
+ weights.update(head, math.MaxInt64)
+ }
+ head = (head + 1) % capacity
+ openBins--
+
+ if !yield(out) {
+ return
+ }
+ }
+ }
+
+ for openBins > 0 {
+ out := bins[head].items
+ bins[head] = Bin[T]{}
+ if indexed {
+ weights.update(head, math.MaxInt64)
+ }
+ head = (head + 1) % capacity
+ openBins--
+
+ if !yield(out) {
+ return
+ }
+ }
+ }
+}
diff --git a/internal/packing_indexed_test.go b/internal/packing_indexed_test.go
new file mode 100644
index 000000000..c4487f24d
--- /dev/null
+++ b/internal/packing_indexed_test.go
@@ -0,0 +1,129 @@
+// 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 internal_test
+
+import (
+ "math"
+ "math/rand"
+ "reflect"
+ "slices"
+ "strconv"
+ "testing"
+
+ "github.com/apache/iceberg-go/internal"
+)
+
+func TestIndexedFIFOPackingMatchesLinear(t *testing.T) {
+ rng := rand.New(rand.NewSource(42))
+
+ for _, lookback := range []int{31, 32, 64, 128, 512} {
+ for trial := range 36 {
+ target := []int64{0, 10, 128, math.MaxInt64 / 2,
math.MaxInt64}[trial%5]
+ items := make([]int, 2*lookback+32)
+ weights := make([]int64, len(items))
+ for i := range items {
+ items[i] = i
+ switch trial % 3 {
+ case 0:
+ weights[i] = int64(1 + rng.Intn(128))
+ case 1:
+ weights[i] = int64(65 + rng.Intn(64))
+ default:
+ weights[i] = int64(rng.Intn(161) - 20)
+ }
+ }
+ switch trial % 11 {
+ case 0:
+ weights[len(weights)/2] = math.MaxInt64
+ case 1:
+ weights[len(weights)/2] = math.MinInt64
+ }
+
+ weight := func(index int) int64 { return weights[index]
}
+ packer := internal.SlicePacker[int]{
+ TargetWeight: target,
+ Lookback: lookback,
+ }
+
+ linear := slices.Collect(internal.PackingIterator(
+ slices.Values(items), target, lookback, weight,
false))
+ if got := packer.Pack(items, weight);
!reflect.DeepEqual(got, linear) {
+ t.Fatalf("Pack lookback=%d trial=%d: got %v,
want %v",
+ lookback, trial, got, linear)
+ }
+
+ reversed := func(yield func(int) bool) {
+ for _, item := range slices.Backward(items) {
+ if !yield(item) {
+ return
+ }
+ }
+ }
+ linearEnd := slices.Collect(internal.PackingIterator(
+ reversed, target, lookback, weight, false))
+ slices.Reverse(linearEnd)
+ for _, bin := range linearEnd {
+ slices.Reverse(bin)
+ }
+ if got := packer.PackEnd(items, weight);
!reflect.DeepEqual(got, linearEnd) {
+ t.Fatalf("PackEnd lookback=%d trial=%d: got %v,
want %v",
+ lookback, trial, got, linearEnd)
+ }
+ }
+ }
+}
+
+var indexedFIFOBenchmarkSink [][]int64
+
+func BenchmarkIndexedFIFOSlicePacking(b *testing.B) {
+ for _, lookback := range []int{128, 512} {
+ for _, workload := range []string{"mixed", "large"} {
+ items := make([]int64, 16_384)
+ for i := range items {
+ switch workload {
+ case "mixed":
+ items[i] = int64(1 + (i*73)%128)
+ case "large":
+ items[i] = int64(65 + i%64)
+ }
+ }
+
+ weight := func(value int64) int64 { return value }
+ packer := internal.SlicePacker[int64]{
+ TargetWeight: 128,
+ Lookback: lookback,
+ }
+
+ for _, mode := range []string{"linear", "indexed"} {
+
b.Run("lookback="+strconv.Itoa(lookback)+"/"+workload+"/"+mode, func(b
*testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for range b.N {
+ if mode == "linear" {
+
indexedFIFOBenchmarkSink = slices.Collect(internal.PackingIterator(
+
slices.Values(items), 128, lookback, weight, false))
+ } else {
+
indexedFIFOBenchmarkSink = packer.Pack(items, weight)
+ }
+ }
+ })
+ }
+ }
+ }
+}
diff --git a/internal/utils.go b/internal/utils.go
index 53fa0e220..ca9108f8b 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -189,18 +189,32 @@ type SlicePacker[T any] struct {
}
func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
+ if s.shouldUseIndexedFIFO(len(items)) {
+ return
slices.Collect(indexedFIFOPackingIterator(slices.Values(items),
+ s.TargetWeight, s.Lookback, weightFunc))
+ }
+
return slices.Collect(PackingIterator(slices.Values(items),
s.TargetWeight,
s.Lookback, weightFunc, s.LargestBinFirst))
}
func (s *SlicePacker[T]) PackEnd(items []T, weightFunc func(T) int64) [][]T {
- packed := slices.Collect(PackingIterator(func(yield func(T) bool) {
+ reversed := func(yield func(T) bool) {
for i := len(items); i > 0; i-- {
if !yield(items[i-1]) {
return
}
}
- }, s.TargetWeight, s.Lookback, weightFunc, s.LargestBinFirst))
+ }
+
+ var packed [][]T
+ if s.shouldUseIndexedFIFO(len(items)) {
+ packed = slices.Collect(indexedFIFOPackingIterator(reversed,
+ s.TargetWeight, s.Lookback, weightFunc))
+ } else {
+ packed = slices.Collect(PackingIterator(reversed,
+ s.TargetWeight, s.Lookback, weightFunc,
s.LargestBinFirst))
+ }
slices.Reverse(packed)
for _, items := range packed {
