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 5623f812f perf(table): make bulk schema field moves linear (#2277)
5623f812f is described below
commit 5623f812f654253d7f544421d01681e58c1d9de7
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 18:56:28 2026 +0200
perf(table): make bulk schema field moves linear (#2277)
* perf(table): index bulk schema field moves
Signed-off-by: Minh Vu <[email protected]>
* style(table): satisfy lint in bulk schema moves
* style(table): satisfy lint in bulk schema moves
* style(table): satisfy lint in bulk schema moves
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/update_schema.go | 97 ++++++++++++++++++++
table/update_schema_moves_bench_test.go | 81 +++++++++++++++++
table/update_schema_moves_indexed_test.go | 143 ++++++++++++++++++++++++++++++
3 files changed, 321 insertions(+)
diff --git a/table/update_schema.go b/table/update_schema.go
index dea77812f..e4e5c95dd 100644
--- a/table/update_schema.go
+++ b/table/update_schema.go
@@ -1347,7 +1347,104 @@ func addFields(fields []iceberg.NestedField, adds
[]iceberg.NestedField) []icebe
return append(fields, adds...)
}
+// fieldMoveLink stores stable field positions while moves relink their order.
+// Unlike slice insertions, each move changes only the adjacent links.
+type fieldMoveLink struct {
+ prev int
+ next int
+}
+
+// moveFields avoids the repeated linear searches and slice copies involved
+// in bulk reordering wide schemas. Small move sets keep the simpler linear
+// path to avoid allocating an index for the common one- or two-move case.
func moveFields(fields []iceberg.NestedField, moves []move)
[]iceberg.NestedField {
+ if len(fields) < 64 || len(moves) < 8 {
+ return moveFieldsLinear(fields, moves)
+ }
+
+ for _, m := range moves {
+ // Preserve the existing handling of unexpected move operations.
+ if m.Op != MoveOpFirst && m.Op != MoveOpBefore && m.Op !=
MoveOpAfter {
+ return moveFieldsLinear(fields, moves)
+ }
+ }
+
+ links := make([]fieldMoveLink, len(fields))
+ positions := make(map[int]int, len(fields))
+ for i, field := range fields {
+ // Duplicate IDs are invalid for a normal schema, but preserve
the
+ // original first-match behavior if they reach this helper.
+ if _, exists := positions[field.ID]; exists {
+ return moveFieldsLinear(fields, moves)
+ }
+ positions[field.ID] = i
+ links[i] = fieldMoveLink{prev: i - 1, next: i + 1}
+ }
+ links[len(links)-1].next = -1
+
+ head := 0
+ for _, m := range moves {
+ idx, ok := positions[m.FieldID]
+ if !ok {
+ panic(fmt.Errorf("cannot move field %d: field not
found", m.FieldID))
+ }
+
+ relativeIdx := -1
+ if m.Op == MoveOpBefore || m.Op == MoveOpAfter {
+ var found bool
+ relativeIdx, found = positions[m.RelativeTo]
+ // The original implementation removes the source
before finding
+ // the target, so a self-relative move must still fail.
+ if !found || relativeIdx == idx {
+ panic(fmt.Errorf("cannot move field %d: target
field %d not found", m.FieldID, m.RelativeTo))
+ }
+ }
+
+ prev, next := links[idx].prev, links[idx].next
+ if prev >= 0 {
+ links[prev].next = next
+ } else {
+ head = next
+ }
+ if next >= 0 {
+ links[next].prev = prev
+ }
+
+ switch m.Op {
+ case MoveOpFirst:
+ links[idx] = fieldMoveLink{prev: -1, next: head}
+ if head >= 0 {
+ links[head].prev = idx
+ }
+ head = idx
+ case MoveOpBefore:
+ prev = links[relativeIdx].prev
+ links[idx] = fieldMoveLink{prev: prev, next:
relativeIdx}
+ links[relativeIdx].prev = idx
+ if prev >= 0 {
+ links[prev].next = idx
+ } else {
+ head = idx
+ }
+ case MoveOpAfter:
+ next = links[relativeIdx].next
+ links[idx] = fieldMoveLink{prev: relativeIdx, next:
next}
+ links[relativeIdx].next = idx
+ if next >= 0 {
+ links[next].prev = idx
+ }
+ }
+ }
+
+ reordered := make([]iceberg.NestedField, 0, len(fields))
+ for idx := head; idx >= 0; idx = links[idx].next {
+ reordered = append(reordered, fields[idx])
+ }
+
+ return reordered
+}
+
+func moveFieldsLinear(fields []iceberg.NestedField, moves []move)
[]iceberg.NestedField {
reordered := slices.Clone(fields)
for _, move := range moves {
var fieldToMove iceberg.NestedField
diff --git a/table/update_schema_moves_bench_test.go
b/table/update_schema_moves_bench_test.go
new file mode 100644
index 000000000..1ba2cfe54
--- /dev/null
+++ b/table/update_schema_moves_bench_test.go
@@ -0,0 +1,81 @@
+// 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 (
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+var moveFieldsBenchmarkSink []iceberg.NestedField
+
+func BenchmarkMoveFieldsBulkReorder(b *testing.B) {
+ cases := []struct {
+ fieldCount int
+ moveCount int
+ mixed bool
+ }{
+ {16, 2, false},
+ {64, 8, false},
+ {256, 64, false},
+ {1024, 1024, false},
+ {4096, 4096, false},
+ {256, 64, true},
+ {1024, 1024, true},
+ }
+ for _, tc := range cases {
+ fields := makeMoveTestFields(tc.fieldCount)
+ moves := make([]move, tc.moveCount)
+ for i := range moves {
+ moves[i] = move{FieldID: tc.fieldCount -
i%tc.fieldCount, Op: MoveOpFirst}
+ if tc.mixed {
+ moves[i].FieldID = 1 + i%tc.fieldCount
+ moves[i].RelativeTo = 1 + (i*7+3)%tc.fieldCount
+ if moves[i].FieldID == moves[i].RelativeTo {
+ moves[i].RelativeTo = 1 +
moves[i].RelativeTo%tc.fieldCount
+ }
+ switch i % 3 {
+ case 1:
+ moves[i].Op = MoveOpBefore
+ case 2:
+ moves[i].Op = MoveOpAfter
+ }
+ }
+ }
+ shape := "first"
+ if tc.mixed {
+ shape = "mixed"
+ }
+ name := fmt.Sprintf("%s/%d_fields_%d_moves", shape,
tc.fieldCount, tc.moveCount)
+ b.Run(name, func(b *testing.B) {
+ for _, impl := range []struct {
+ name string
+ fn func([]iceberg.NestedField, []move)
[]iceberg.NestedField
+ }{{"linear", moveFieldsLinear}, {"indexed",
moveFields}} {
+ b.Run(impl.name, func(b *testing.B) {
+ b.ReportAllocs()
+ for range b.N {
+ moveFieldsBenchmarkSink =
impl.fn(fields, moves)
+ }
+ })
+ }
+ })
+ }
+}
diff --git a/table/update_schema_moves_indexed_test.go
b/table/update_schema_moves_indexed_test.go
new file mode 100644
index 000000000..6f03da76e
--- /dev/null
+++ b/table/update_schema_moves_indexed_test.go
@@ -0,0 +1,143 @@
+// 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 (
+ "fmt"
+ "math/rand"
+ "reflect"
+ "slices"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+func makeMoveTestFields(count int) []iceberg.NestedField {
+ fields := make([]iceberg.NestedField, count)
+ for i := range fields {
+ fields[i] = iceberg.NestedField{ID: i + 1, Name:
fmt.Sprintf("col_%d", i)}
+ }
+
+ return fields
+}
+
+func TestMoveFieldsIndexedMatchesLinear(t *testing.T) {
+ for _, count := range []int{1, 16, 63, 64, 128, 256, 1024} {
+ for _, moveCount := range []int{0, 1, 7, 8, 64, 256} {
+ t.Run(fmt.Sprintf("%d_fields_%d_moves", count,
moveCount), func(t *testing.T) {
+ fields := makeMoveTestFields(count)
+ original := slices.Clone(fields)
+ for seed := range int64(10) {
+ rng := rand.New(rand.NewSource(seed))
+ moves := make([]move, moveCount)
+ for i := range moves {
+ moves[i].FieldID = 1 +
rng.Intn(count)
+ moves[i].RelativeTo = 1 +
rng.Intn(count)
+ if moves[i].RelativeTo ==
moves[i].FieldID {
+ moves[i].RelativeTo = 1
+ moves[i].RelativeTo%count
+ }
+ switch rng.Intn(3) {
+ case 0:
+ moves[i].Op =
MoveOpFirst
+ case 1:
+ moves[i].Op =
MoveOpBefore
+ case 2:
+ moves[i].Op =
MoveOpAfter
+ }
+ if count == 1 {
+ moves[i].Op =
MoveOpFirst
+ }
+ }
+ want := moveFieldsLinear(fields, moves)
+ got := moveFields(fields, moves)
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("seed %d: move order
mismatch: got %v, want %v", seed, got, want)
+ }
+ if !reflect.DeepEqual(fields, original)
{
+ t.Fatal("moveFields mutated
input fields")
+ }
+ }
+ })
+ }
+ }
+}
+
+func TestMoveFieldsIndexedFallbacks(t *testing.T) {
+ fields := makeMoveTestFields(128)
+ t.Run("missing source", func(t *testing.T) {
+ moves := make([]move, 8)
+ for i := range moves {
+ moves[i] = move{FieldID: 99999, Op: MoveOpFirst}
+ }
+ assertSameMovePanic(t, fields, moves)
+ })
+ t.Run("missing target", func(t *testing.T) {
+ moves := make([]move, 8)
+ for i := range moves {
+ moves[i] = move{FieldID: i + 1, RelativeTo: 99999, Op:
MoveOpAfter}
+ }
+ assertSameMovePanic(t, fields, moves)
+ })
+ t.Run("self relative", func(t *testing.T) {
+ moves := make([]move, 8)
+ for i := range moves {
+ moves[i] = move{FieldID: i + 1, RelativeTo: i + 1, Op:
MoveOpBefore}
+ }
+ assertSameMovePanic(t, fields, moves)
+ })
+ t.Run("duplicate IDs", func(t *testing.T) {
+ duplicate := slices.Clone(fields)
+ duplicate[1].ID = duplicate[0].ID
+ moves := []move{
+ {FieldID: 1, Op: MoveOpFirst},
+ {FieldID: 3, Op: MoveOpFirst},
+ {FieldID: 4, Op: MoveOpFirst},
+ {FieldID: 5, Op: MoveOpFirst},
+ {FieldID: 6, Op: MoveOpFirst},
+ {FieldID: 7, Op: MoveOpFirst},
+ {FieldID: 8, Op: MoveOpFirst},
+ {FieldID: 9, Op: MoveOpFirst},
+ }
+ if got, want := moveFields(duplicate, moves),
moveFieldsLinear(duplicate, moves); !reflect.DeepEqual(got, want) {
+ t.Fatalf("duplicate field IDs: got %v, want %v", got,
want)
+ }
+ })
+ t.Run("unknown operation", func(t *testing.T) {
+ moves := make([]move, 8)
+ for i := range moves {
+ moves[i] = move{FieldID: i + 1, Op: MoveOp("unknown")}
+ }
+ if got, want := moveFields(fields, moves),
moveFieldsLinear(fields, moves); !reflect.DeepEqual(got, want) {
+ t.Fatalf("unknown operation: got %v, want %v", got,
want)
+ }
+ })
+}
+
+func assertSameMovePanic(t *testing.T, fields []iceberg.NestedField, moves
[]move) {
+ t.Helper()
+ getPanic := func(fn func([]iceberg.NestedField, []move)
[]iceberg.NestedField) (panicValue any) {
+ defer func() { panicValue = recover() }()
+ fn(fields, moves)
+
+ return nil
+ }
+ got, want := getPanic(moveFields), getPanic(moveFieldsLinear)
+ if got == nil || fmt.Sprint(got) != fmt.Sprint(want) {
+ t.Fatalf("panic mismatch: got %v, want %v", got, want)
+ }
+}
