You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2025/03/09 21:41:46 UTC
(iceberg-go) branch main updated: feat(internal): Adding BinPacker (#321)
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 16e6b69 feat(internal): Adding BinPacker (#321)
16e6b69 is described below
commit 16e6b69db76b25146e3c286c6c95be2410d2cc89
Author: Matt Topol <zo...@gmail.com>
AuthorDate: Sun Mar 9 17:41:39 2025 -0400
feat(internal): Adding BinPacker (#321)
Needed for adding manifest and data file writers
---
internal/utils.go | 97 ++++++++++++++++++++++++++++++++++++++++++++
internal/utils_test.go | 106 ++++++++++++++++++++++++++++++++++++++++++++++++
table/internal/utils.go | 4 +-
3 files changed, 206 insertions(+), 1 deletion(-)
diff --git a/internal/utils.go b/internal/utils.go
index ced65cb..c7eb231 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -17,6 +17,11 @@
package internal
+import (
+ "cmp"
+ "slices"
+)
+
// Helper function to find the difference between two slices (a - b).
func Difference(a, b []string) []string {
m := make(map[string]bool)
@@ -33,3 +38,95 @@ func Difference(a, b []string) []string {
return diff
}
+
+type Bin[T any] struct {
+ binWeight int64
+ targetWeight int64
+ items []T
+}
+
+func (b *Bin[T]) Weight() int64 { return b.binWeight }
+func (b *Bin[T]) CanAdd(weight int64) bool { return b.binWeight+weight <= b.targetWeight }
+func (b *Bin[T]) Add(item T, weight int64) {
+ b.binWeight += weight
+ b.items = append(b.items, item)
+}
+
+type SlicePacker[T any] struct {
+ TargetWeight int64
+ Lookback int
+ LargestBinFirst bool
+}
+
+func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
+ bins := make([]Bin[T], 0)
+ findBin := func(weight int64) *Bin[T] {
+ for i := range bins {
+ if bins[i].CanAdd(weight) {
+ return &bins[i]
+ }
+ }
+
+ return nil
+ }
+
+ removeBin := func() Bin[T] {
+ if s.LargestBinFirst {
+ maxBin := slices.MaxFunc(bins, func(a, b Bin[T]) int {
+ return cmp.Compare(a.Weight(), b.Weight())
+ })
+ i := slices.IndexFunc(bins, func(e Bin[T]) bool {
+ return e.Weight() == maxBin.Weight()
+ })
+
+ bins = slices.Delete(bins, i, i+1)
+
+ return maxBin
+ }
+
+ var out Bin[T]
+ out, bins = bins[0], bins[1:]
+
+ return out
+ }
+
+ return slices.Collect(func(yield func([]T) bool) {
+ for _, item := range items {
+ w := weightFunc(item)
+ bin := findBin(w)
+ if bin != nil {
+ bin.Add(item, w)
+ } else {
+ bin := Bin[T]{targetWeight: s.TargetWeight}
+ bin.Add(item, w)
+ bins = append(bins, bin)
+
+ if len(bins) > s.Lookback {
+ if !yield(removeBin().items) {
+ return
+ }
+ }
+ }
+ }
+
+ for len(bins) > 0 {
+ if !yield(removeBin().items) {
+ return
+ }
+ }
+ })
+}
+
+func (s *SlicePacker[T]) PackEnd(items []T, weightFunc func(T) int64) [][]T {
+ slices.Reverse(items)
+ packed := s.Pack(items, weightFunc)
+ slices.Reverse(packed)
+
+ result := make([][]T, 0, len(packed))
+ for _, items := range packed {
+ slices.Reverse(items)
+ result = append(result, items)
+ }
+
+ return result
+}
diff --git a/internal/utils_test.go b/internal/utils_test.go
index a90b3eb..236c5b1 100644
--- a/internal/utils_test.go
+++ b/internal/utils_test.go
@@ -18,12 +18,118 @@
package internal_test
import (
+ "math"
"testing"
"github.com/apache/iceberg-go/internal"
+ "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+func TestBinPacking(t *testing.T) {
+ tests := []struct {
+ splits []int
+ tgtWeight int64
+ lookback int
+ largestBinFirst bool
+ expected [][]int
+ }{
+ {
+ []int{36, 36, 36, 36, 73, 110, 128},
+ 128, 2, true,
+ [][]int{{110}, {128}, {36, 73}, {36, 36, 36}},
+ },
+ {
+ []int{36, 36, 36, 36, 73, 110, 128},
+ 128, 2, false,
+ [][]int{{36, 36, 36}, {36, 73}, {110}, {128}},
+ },
+ {
+ []int{64, 64, 128, 32, 32, 32, 32},
+ 128, 1, true,
+ [][]int{{64, 64}, {128}, {32, 32, 32, 32}},
+ },
+ {
+ []int{64, 64, 128, 32, 32, 32, 32},
+ 128, 1, false,
+ [][]int{{64, 64}, {128}, {32, 32, 32, 32}},
+ },
+ }
+
+ weightFn := func(i int) int64 {
+ return int64(i)
+ }
+
+ for _, tt := range tests {
+ t.Run("", func(t *testing.T) {
+ p := &internal.SlicePacker[int]{
+ TargetWeight: tt.tgtWeight,
+ Lookback: tt.lookback,
+ LargestBinFirst: tt.largestBinFirst,
+ }
+
+ bins := p.Pack(tt.splits, weightFn)
+ assert.Equal(t, tt.expected, bins)
+ })
+ }
+}
+
+func TestReverseBinPackingLookback(t *testing.T) {
+ intmax := math.MaxInt32
+
+ tests := []struct {
+ splits []int
+ tgtWeight int64
+ lookback int
+ largestBinFirst bool
+ expected [][]int
+ }{
+ // single lookback tests
+ {[]int{1, 2, 3, 4, 5}, 3, 1, false, [][]int{{1, 2}, {3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 4, 1, false, [][]int{{1, 2}, {3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 5, 1, false, [][]int{{1}, {2, 3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 6, 1, false, [][]int{{1, 2, 3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 7, 1, false, [][]int{{1, 2}, {3, 4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 8, 1, false, [][]int{{1, 2}, {3, 4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 9, 1, false, [][]int{{1, 2, 3}, {4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 11, 1, false, [][]int{{1, 2, 3}, {4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 12, 1, false, [][]int{{1, 2}, {3, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 14, 1, false, [][]int{{1}, {2, 3, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 15, 1, false, [][]int{{1, 2, 3, 4, 5}}},
+ // unlimited lookback tests
+ {[]int{1, 2, 3, 4, 5}, 3, intmax, false, [][]int{{1, 2}, {3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 4, intmax, false, [][]int{{2}, {1, 3}, {4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 5, intmax, false, [][]int{{2, 3}, {1, 4}, {5}}},
+ {[]int{1, 2, 3, 4, 5}, 6, intmax, false, [][]int{{3}, {2, 4}, {1, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 7, intmax, false, [][]int{{1}, {3, 4}, {2, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 8, intmax, false, [][]int{{1, 2, 4}, {3, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 9, intmax, false, [][]int{{1, 2, 3}, {4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 10, intmax, false, [][]int{{2, 3}, {1, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 11, intmax, false, [][]int{{1, 3}, {2, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 12, intmax, false, [][]int{{1, 2}, {3, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 13, intmax, false, [][]int{{2}, {1, 3, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 14, intmax, false, [][]int{{1}, {2, 3, 4, 5}}},
+ {[]int{1, 2, 3, 4, 5}, 15, intmax, false, [][]int{{1, 2, 3, 4, 5}}},
+ }
+
+ weightFn := func(i int) int64 {
+ return int64(i)
+ }
+
+ for _, tt := range tests {
+ t.Run("", func(t *testing.T) {
+ p := &internal.SlicePacker[int]{
+ TargetWeight: tt.tgtWeight,
+ Lookback: tt.lookback,
+ LargestBinFirst: tt.largestBinFirst,
+ }
+
+ bins := p.PackEnd(tt.splits, weightFn)
+ assert.Equal(t, tt.expected, bins)
+ })
+ }
+}
+
func TestDifference(t *testing.T) {
tests := []struct {
name string
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 2a1c261..7b5f7d8 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -17,7 +17,9 @@
package internal
-import "container/heap"
+import (
+ "container/heap"
+)
// Enumerated is a quick way to represent a sequenced value that can
// be processed in parallel and then needs to be reordered.