badalprasadsingh commented on code in PR #2438: URL: https://github.com/apache/iceberg-go/pull/2438#discussion_r4491164721
########## table/replace_sort_order_builder.go: ########## @@ -0,0 +1,192 @@ +// 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" + "strings" + + "github.com/apache/iceberg-go" +) + +// ReplaceSortOrderBuilder builds a replacement for the table's default sort order. +// Replacement starts with no fields, rather than copying the existing order. +// Calling Commit stages the change in the transaction; Transaction.Commit +// persists it to the catalog. +type ReplaceSortOrderBuilder struct { + txn *Transaction + caseSensitive bool + fields []sortOrderColumn +} + +type sortOrderColumn struct { + name string + transform iceberg.Transform + direction SortDirection + nullOrder NullOrder +} + +// NewReplaceSortOrder starts a sort order replacement in this transaction. +// Names are resolved using the transaction's current schema. An empty +// replacement resets the default sort order to unsorted (order ID 0). +func (t *Transaction) NewReplaceSortOrder(caseSensitive bool) *ReplaceSortOrderBuilder { + return NewReplaceSortOrderBuilder(t, caseSensitive) +} + +// NewReplaceSortOrderBuilder creates a builder for replacing the default sort order. +func NewReplaceSortOrderBuilder(t *Transaction, caseSensitive bool) *ReplaceSortOrderBuilder { + return &ReplaceSortOrderBuilder{txn: t, caseSensitive: caseSensitive} +} + +// Asc adds a column sorted in ascending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Asc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortASC, nullOrder) +} + +// Desc adds a column sorted in descending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Desc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortDESC, nullOrder) +} + +// AddField adds a transformed sort field. A nil or incompatible transform is +// rejected when the replacement is applied or committed. +func (u *ReplaceSortOrderBuilder) AddField(column string, transform iceberg.Transform, direction SortDirection, nullOrder NullOrder) *ReplaceSortOrderBuilder { + u.fields = append(u.fields, sortOrderColumn{ + name: column, transform: transform, + direction: direction, nullOrder: nullOrder, + }) + + return u +} + +// Apply returns the replacement sort order, including its proposed order ID, +// without modifying the transaction. An equivalent historical order reuses +// its existing ID. +func (u *ReplaceSortOrderBuilder) Apply() (SortOrder, error) { + meta, err := u.stagedMetadata() + if err != nil { + return SortOrder{}, err + } + + return u.buildOrder(meta) +} + +func (u *ReplaceSortOrderBuilder) stagedMetadata() (Metadata, error) { + if u == nil { + return nil, fmt.Errorf("%w: sort order replacement builder is nil", ErrInvalidMetadata) + } + + builder, err := u.txn.txnMeta() + if err != nil { + return nil, err + } + + return builder.Build() +} + +func (u *ReplaceSortOrderBuilder) buildOrder(meta Metadata) (SortOrder, error) { + if len(u.fields) == 0 { + return UnsortedSortOrder, nil + } + + schema := meta.CurrentSchema() + var lowerNameToID map[string]int + if !u.caseSensitive { + // Java's case-insensitive schema index rejects names that differ only + // in case. The generic Go lookup currently overwrites such entries in + // map iteration order, so do not use it for a metadata-changing API. Review Comment: `UpdateSpec` also changes metadata, and it still picks a random field when names collide. Better: make `lazyNameToIDLower` reject collisions, then call `FindFieldByNameCaseInsensitive` and drop this copy. If that's out of scope, please file an issue and cite it in this comment. ########## table/transaction.go: ########## @@ -640,10 +640,20 @@ func (t *Transaction) ReplaceSortOrder(order SortOrder) error { return nil } - return t.apply( - []Update{NewAddSortOrderUpdate(&order), NewSetDefaultSortOrderUpdate(-1)}, - []Requirement{AssertDefaultSortOrderID(t.tbl.Metadata().DefaultSortOrder())}, - ) + reqs := []Requirement{AssertDefaultSortOrderID(t.tbl.Metadata().DefaultSortOrder())} + for _, previous := range meta.sortOrderList { + if sameSortFields(previous, order) { + if err := order.CheckCompatibility(meta.CurrentSchema()); err != nil { Review Comment: the payload changes here, please add it in the release notes add it in pr description as well ########## table/replace_sort_order_builder_test.go: ########## @@ -0,0 +1,341 @@ +// 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_test + +import ( + "encoding/json" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestReplaceSortOrderBuilderReplaceAndReuse(t *testing.T) { + txn := testNonPartitionedTable.NewTransaction() + + builder := txn.NewReplaceSortOrder(true).Asc("id", table.NullsLast).Desc("name", table.NullsFirst) + preview, err := builder.Apply() + require.NoError(t, err) + require.Equal(t, 1, preview.OrderID()) + require.Equal(t, 2, preview.Len()) + + unchanged, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 0, unchanged.Metadata().DefaultSortOrder(), "Apply must not stage updates") + + require.NoError(t, builder.Commit()) + firstCommit, err := txn.TableCommit() + require.NoError(t, err) + require.Len(t, firstCommit.Updates, 2) + require.Len(t, firstCommit.Requirements, 3) + require.Equal(t, "assert-default-sort-order-id", firstCommit.Requirements[0].GetType()) + require.NoError(t, firstCommit.Requirements[0].Validate(unchanged.Metadata())) + + raw, err := json.Marshal(firstCommit.Updates) + require.NoError(t, err) + require.Contains(t, string(raw), "\"action\":\"add-sort-order\"") + require.Contains(t, string(raw), "\"action\":\"set-default-sort-order\"") + require.Contains(t, string(raw), "\"sort-order-id\":-1") + first, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 1, first.Metadata().DefaultSortOrder()) + require.Equal(t, 2, first.Metadata().SortOrder().Len()) + require.Len(t, first.Metadata().SortOrders(), 2) + + fields := make([]table.SortField, 0, 2) + for _, f := range first.Metadata().SortOrder().Fields() { + fields = append(fields, f) + } + require.Equal(t, []int{1}, fields[0].SourceIDs) + require.Equal(t, table.SortASC, fields[0].Direction) + require.Equal(t, table.NullsLast, fields[0].NullOrder) + require.Equal(t, []int{2}, fields[1].SourceIDs) + require.Equal(t, table.SortDESC, fields[1].Direction) + require.Equal(t, table.NullsFirst, fields[1].NullOrder) + + // A concurrent change to the default sort order must invalidate the + // optimistic-concurrency requirement captured from the original table. + require.Error(t, firstCommit.Requirements[0].Validate(first.Metadata())) + + // Reapplying the same sort order must not create a new order or update. + require.NoError(t, builder.Commit()) + same, err := txn.StagedTable() + require.NoError(t, err) + require.Len(t, same.Metadata().SortOrders(), 2) + + // A different definition gets a new ID, retaining previous orders. + require.NoError(t, txn.NewReplaceSortOrder(true).Desc("id", table.NullsFirst).Commit()) + second, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 2, second.Metadata().DefaultSortOrder()) + require.Len(t, second.Metadata().SortOrders(), 3) + + // Returning to a historical definition only changes the default ID. + restore := txn.NewReplaceSortOrder(true).Asc("id", table.NullsLast).Desc("name", table.NullsFirst) + reused, err := restore.Apply() + require.NoError(t, err) + require.Equal(t, 1, reused.OrderID()) + + beforeRestore, err := txn.TableCommit() + require.NoError(t, err) + require.NoError(t, restore.Commit()) + afterRestore, err := txn.TableCommit() + require.NoError(t, err) + require.Len(t, afterRestore.Updates, len(beforeRestore.Updates)+1) + raw, err = json.Marshal(afterRestore.Updates[len(afterRestore.Updates)-1]) + require.NoError(t, err) + require.Contains(t, string(raw), "\"sort-order-id\":1") + restored, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 1, restored.Metadata().DefaultSortOrder()) + require.Len(t, restored.Metadata().SortOrders(), 3) + + // Undoing sorting reuses order 0 and preserves historical definitions. + unsort := txn.NewReplaceSortOrder(true) + require.NoError(t, unsort.Commit()) + last, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 0, last.Metadata().DefaultSortOrder()) + require.True(t, last.Metadata().SortOrder().IsUnsorted()) + require.Len(t, last.Metadata().SortOrders(), 3) + + commits, err := txn.TableCommit() + require.NoError(t, err) + require.Len(t, commits.Requirements, 3, "deduplicate requirements across staged replacements") + require.Equal(t, "assert-default-sort-order-id", commits.Requirements[0].GetType()) + require.Equal(t, "assert-ref-snapshot-id", commits.Requirements[1].GetType()) + require.Equal(t, "assert-table-uuid", commits.Requirements[2].GetType()) +} + +func TestReplaceSortOrderBuilderNoopAndUnsortedAddition(t *testing.T) { + txn := testNonPartitionedTable.NewTransaction() + empty := txn.NewReplaceSortOrder(true) + preview, err := empty.Apply() + require.NoError(t, err) + require.Equal(t, 0, preview.OrderID()) + require.NoError(t, empty.Commit()) + commit, err := txn.TableCommit() + require.NoError(t, err) + require.Empty(t, commit.Updates) + + // A table initially created with a sorted order need not list order 0. + sorted, err := table.NewSortOrder(1, []table.SortField{{ + SourceIDs: []int{1}, Transform: iceberg.IdentityTransform{}, + Direction: table.SortASC, NullOrder: table.NullsFirst, + }}) + require.NoError(t, err) + meta, err := table.NewMetadata(testSchema, iceberg.UnpartitionedSpec, sorted, "file:///tmp/sorted", nil) + require.NoError(t, err) + require.Equal(t, 1, meta.DefaultSortOrder()) + + tbl := table.New([]string{"sorted"}, meta, "", nil, nil) + t2 := tbl.NewTransaction() + require.NoError(t, t2.NewReplaceSortOrder(true).Commit()) + result, err := t2.StagedTable() + require.NoError(t, err) + require.Equal(t, 0, result.Metadata().DefaultSortOrder()) + committed, err := t2.TableCommit() + require.NoError(t, err) + require.Len(t, committed.Updates, 2, "unsorted order must be registered before becoming default") + require.Len(t, result.Metadata().SortOrders(), 2) + require.Equal(t, 1, result.Metadata().SortOrders()[0].OrderID()) + require.Equal(t, 0, result.Metadata().SortOrders()[1].OrderID()) +} + +func TestReplaceSortOrderBuilderTransformsAndNameBinding(t *testing.T) { + txn := testNonPartitionedTable.NewTransaction() + builder := txn.NewReplaceSortOrder(false). + AddField("TS", iceberg.MonthTransform{}, table.SortDESC, table.NullsLast). + Asc("NaMe", table.NullsFirst) + require.NoError(t, builder.Commit()) + staged, err := txn.StagedTable() + require.NoError(t, err) + require.Equal(t, 1, staged.Metadata().DefaultSortOrder()) + + fields := make([]table.SortField, 0, 2) + for _, f := range staged.Metadata().SortOrder().Fields() { + fields = append(fields, f) + } + require.Equal(t, []int{3}, fields[0].SourceIDs) + require.Equal(t, iceberg.MonthTransform{}, fields[0].Transform) + require.Equal(t, table.SortDESC, fields[0].Direction) + require.Equal(t, table.NullsLast, fields[0].NullOrder) + require.Equal(t, []int{2}, fields[1].SourceIDs) + require.Equal(t, iceberg.IdentityTransform{}, fields[1].Transform) +} + +func TestReplaceSortOrderBuilderRejectsInvalidFieldsWithoutStaging(t *testing.T) { + tests := []struct { + name string + build func(*table.Transaction) *table.ReplaceSortOrderBuilder + errorText string + }{ + { + name: "case sensitive field missing", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).Asc("ID", table.NullsFirst) + }, + errorText: "cannot bind sort field", + }, + { + name: "unknown field", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(false).Desc("missing", table.NullsLast) + }, + errorText: "cannot bind sort field", + }, + { + name: "non-primitive field", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).Asc("address", table.NullsLast) + }, + errorText: "cannot sort by non-primitive", + }, + { + name: "incompatible transform", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).AddField("name", iceberg.YearTransform{}, table.SortASC, table.NullsLast) + }, + errorText: "invalid source type", + }, + { + name: "nil transform", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).AddField("id", nil, table.SortASC, table.NullsLast) + }, + errorText: "has no transform", + }, + { + name: "invalid direction", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).AddField("id", iceberg.IdentityTransform{}, "sideways", table.NullsLast) + }, + errorText: "invalid sort direction", + }, + { + name: "invalid null placement", + build: func(t *table.Transaction) *table.ReplaceSortOrderBuilder { + return t.NewReplaceSortOrder(true).AddField("id", iceberg.IdentityTransform{}, table.SortDESC, "random") + }, + errorText: "invalid null order", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + txn := testNonPartitionedTable.NewTransaction() + err := tt.build(txn).Commit() + require.ErrorContains(t, err, tt.errorText) Review Comment: `Commit` checks again in `AddSortOrder`, so no test catches Apply returning a bad order, or a wrong ID for a second new order. I'd call Apply in this loop too, and add a parity case for a second new order. ########## table/replace_sort_order_builder_parity_test.go: ########## @@ -0,0 +1,165 @@ +// 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_test + +import ( + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestReplaceSortOrderBuilderMatchesDirectReplacement(t *testing.T) { + byID := table.SortField{ + SourceIDs: []int{1}, Transform: iceberg.IdentityTransform{}, + Direction: table.SortASC, NullOrder: table.NullsFirst, + } + byName := table.SortField{ + SourceIDs: []int{2}, Transform: iceberg.IdentityTransform{}, + Direction: table.SortDESC, NullOrder: table.NullsLast, + } + byMonth := table.SortField{ + SourceIDs: []int{3}, Transform: iceberg.MonthTransform{}, + Direction: table.SortDESC, NullOrder: table.NullsLast, + } + + cases := []struct { + name string + prepare func(*testing.T, *table.Transaction) + build func(*table.Transaction) *table.ReplaceSortOrderBuilder + wantOrderID int + wantOrderCount int + wantUpdates int + }{ + { + name: "new order", + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true).Asc("id", table.NullsFirst) + }, + wantOrderID: 1, wantOrderCount: 2, wantUpdates: 2, + }, + { + name: "current order is a no-op", + prepare: func(t *testing.T, txn *table.Transaction) { + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byID))) + }, + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true).Asc("id", table.NullsFirst) + }, + wantOrderID: 1, wantOrderCount: 2, wantUpdates: 2, + }, + { + name: "reuse historical order", + prepare: func(t *testing.T, txn *table.Transaction) { + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byID))) + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byName))) + }, + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true).Asc("id", table.NullsFirst) + }, + wantOrderID: 1, wantOrderCount: 3, wantUpdates: 5, + }, + { + name: "reuse historical transformed order", + prepare: func(t *testing.T, txn *table.Transaction) { + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byMonth))) + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byID))) + }, + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true).AddField("ts", iceberg.MonthTransform{}, table.SortDESC, table.NullsLast) + }, + wantOrderID: 1, wantOrderCount: 3, wantUpdates: 5, + }, + { + name: "reset to unsorted", + prepare: func(t *testing.T, txn *table.Transaction) { + require.NoError(t, txn.ReplaceSortOrder(sortOrderOf(t, byID))) + }, + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true) + }, + wantOrderID: 0, wantOrderCount: 2, wantUpdates: 3, + }, + { + name: "bind against staged schema", + prepare: func(t *testing.T, txn *table.Transaction) { + require.NoError(t, txn.UpdateSchema(true, false). + AddColumn([]string{"priority"}, iceberg.PrimitiveTypes.Int64, "", false, nil).Commit()) + }, + build: func(txn *table.Transaction) *table.ReplaceSortOrderBuilder { + return txn.NewReplaceSortOrder(true).Desc("priority", table.NullsLast) + }, + wantOrderID: 1, wantOrderCount: 2, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + direct := testNonPartitionedTable.NewTransaction() + withBuilder := testNonPartitionedTable.NewTransaction() + if tc.prepare != nil { + tc.prepare(t, direct) + tc.prepare(t, withBuilder) + } + + builder := tc.build(withBuilder) + preview, err := builder.Apply() + require.NoError(t, err) + require.Equal(t, tc.wantOrderID, preview.OrderID()) + + require.NoError(t, direct.ReplaceSortOrder(preview)) + require.NoError(t, builder.Commit()) + + directCommit, err := direct.TableCommit() + require.NoError(t, err) + builderCommit, err := withBuilder.TableCommit() + require.NoError(t, err) + require.Equal(t, directCommit.Updates, builderCommit.Updates) Review Comment: P1: this fails on every run, which is the reason why CI is red. `require.Equal` also compares hidden lookup caches inside each schema, and those differ between the two txns. marshal both update lists to JSON and compare them with require.JSONEq. ########## table/replace_sort_order_builder.go: ########## @@ -0,0 +1,192 @@ +// 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" + "strings" + + "github.com/apache/iceberg-go" +) + +// ReplaceSortOrderBuilder builds a replacement for the table's default sort order. +// Replacement starts with no fields, rather than copying the existing order. +// Calling Commit stages the change in the transaction; Transaction.Commit +// persists it to the catalog. +type ReplaceSortOrderBuilder struct { + txn *Transaction + caseSensitive bool + fields []sortOrderColumn +} + +type sortOrderColumn struct { + name string + transform iceberg.Transform + direction SortDirection + nullOrder NullOrder +} + +// NewReplaceSortOrder starts a sort order replacement in this transaction. +// Names are resolved using the transaction's current schema. An empty +// replacement resets the default sort order to unsorted (order ID 0). +func (t *Transaction) NewReplaceSortOrder(caseSensitive bool) *ReplaceSortOrderBuilder { + return NewReplaceSortOrderBuilder(t, caseSensitive) +} + +// NewReplaceSortOrderBuilder creates a builder for replacing the default sort order. +func NewReplaceSortOrderBuilder(t *Transaction, caseSensitive bool) *ReplaceSortOrderBuilder { + return &ReplaceSortOrderBuilder{txn: t, caseSensitive: caseSensitive} +} + +// Asc adds a column sorted in ascending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Asc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortASC, nullOrder) +} + +// Desc adds a column sorted in descending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Desc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortDESC, nullOrder) +} + +// AddField adds a transformed sort field. A nil or incompatible transform is +// rejected when the replacement is applied or committed. +func (u *ReplaceSortOrderBuilder) AddField(column string, transform iceberg.Transform, direction SortDirection, nullOrder NullOrder) *ReplaceSortOrderBuilder { + u.fields = append(u.fields, sortOrderColumn{ + name: column, transform: transform, + direction: direction, nullOrder: nullOrder, + }) + + return u +} + +// Apply returns the replacement sort order, including its proposed order ID, +// without modifying the transaction. An equivalent historical order reuses +// its existing ID. +func (u *ReplaceSortOrderBuilder) Apply() (SortOrder, error) { + meta, err := u.stagedMetadata() + if err != nil { + return SortOrder{}, err + } + + return u.buildOrder(meta) +} + +func (u *ReplaceSortOrderBuilder) stagedMetadata() (Metadata, error) { + if u == nil { + return nil, fmt.Errorf("%w: sort order replacement builder is nil", ErrInvalidMetadata) + } + + builder, err := u.txn.txnMeta() + if err != nil { + return nil, err + } + + return builder.Build() +} + +func (u *ReplaceSortOrderBuilder) buildOrder(meta Metadata) (SortOrder, error) { + if len(u.fields) == 0 { + return UnsortedSortOrder, nil + } + + schema := meta.CurrentSchema() + var lowerNameToID map[string]int + if !u.caseSensitive { + // Java's case-insensitive schema index rejects names that differ only + // in case. The generic Go lookup currently overwrites such entries in + // map iteration order, so do not use it for a metadata-changing API. + byName, err := iceberg.IndexByName(schema) + if err != nil { + return SortOrder{}, err + } + lowerNameToID = make(map[string]int, len(byName)) + for name, id := range byName { + lower := strings.ToLower(name) + if previous, exists := lowerNameToID[lower]; exists && previous != id { + return SortOrder{}, fmt.Errorf("%w: ambiguous case-insensitive sort column %q", iceberg.ErrInvalidSchema, lower) + } + lowerNameToID[lower] = id + } + } + fields := make([]SortField, 0, len(u.fields)) + for _, col := range u.fields { + if col.transform == nil { + return SortOrder{}, fmt.Errorf("%w: sort field %q has no transform", ErrInvalidTransform, col.name) + } + + switch col.transform.(type) { + case iceberg.UnknownTransform: + return SortOrder{}, fmt.Errorf("%w: cannot add sort field %q with unknown transform %s", + ErrInvalidTransform, col.name, col.transform) + } + + var field iceberg.NestedField + var found bool + if u.caseSensitive { + field, found = schema.FindFieldByName(col.name) + } else if id, ok := lowerNameToID[strings.ToLower(col.name)]; ok { + field, found = schema.FindFieldByID(id) + } + if !found { + return SortOrder{}, fmt.Errorf("cannot bind sort field %q: %w", col.name, iceberg.ErrInvalidSchema) + } + // Expression binding panics on complex fields. Reject them before + // constructing an expression, as Iceberg only sorts primitive values. + if _, ok := field.Type.(iceberg.PrimitiveType); !ok { + return SortOrder{}, fmt.Errorf("cannot sort by non-primitive source field: %s", field.Type.Type()) + } + + fields = append(fields, SortField{ + SourceIDs: []int{field.ID}, + Transform: col.transform, + Direction: col.direction, + NullOrder: col.nullOrder, + }) + } + + // Validate the field definitions and transform/type compatibility before + // deciding whether to reuse an existing sort order. + proposed, err := NewSortOrder(InitialSortOrderID, fields) + if err != nil { + return SortOrder{}, err + } + if err := proposed.CheckCompatibility(schema); err != nil { + return SortOrder{}, err + } + + nextID := InitialSortOrderID + for _, previous := range meta.SortOrders() { + if sameSortFields(previous, proposed) { + return previous, nil + } + nextID = max(nextID, previous.OrderID()+1) + } Review Comment: the reuse rule now lives here, in `ReplaceSortOrder`, and in `MetadataBuilder.reuseOrCreateNewSortOrderID`. Better: call that helper in both places, read the builder from `txnMeta()` here, and drop both loops, `stagedMetadata` and the `Build()` call. ########## table/transaction.go: ########## @@ -640,10 +640,20 @@ func (t *Transaction) ReplaceSortOrder(order SortOrder) error { return nil } - return t.apply( - []Update{NewAddSortOrderUpdate(&order), NewSetDefaultSortOrderUpdate(-1)}, - []Requirement{AssertDefaultSortOrderID(t.tbl.Metadata().DefaultSortOrder())}, - ) + reqs := []Requirement{AssertDefaultSortOrderID(t.tbl.Metadata().DefaultSortOrder())} + for _, previous := range meta.sortOrderList { + if sameSortFields(previous, order) { + if err := order.CheckCompatibility(meta.CurrentSchema()); err != nil { + return err + } Review Comment: no test reaches this check, I'd reuse an order whose column was dropped earlier in the transaction, and expect `ReplaceSortOrder` to fail right away with nothing staged. ########## table/replace_sort_order_builder.go: ########## @@ -0,0 +1,192 @@ +// 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" + "strings" + + "github.com/apache/iceberg-go" +) + +// ReplaceSortOrderBuilder builds a replacement for the table's default sort order. +// Replacement starts with no fields, rather than copying the existing order. +// Calling Commit stages the change in the transaction; Transaction.Commit +// persists it to the catalog. +type ReplaceSortOrderBuilder struct { + txn *Transaction + caseSensitive bool + fields []sortOrderColumn +} + +type sortOrderColumn struct { + name string + transform iceberg.Transform + direction SortDirection + nullOrder NullOrder +} + +// NewReplaceSortOrder starts a sort order replacement in this transaction. +// Names are resolved using the transaction's current schema. An empty +// replacement resets the default sort order to unsorted (order ID 0). +func (t *Transaction) NewReplaceSortOrder(caseSensitive bool) *ReplaceSortOrderBuilder { + return NewReplaceSortOrderBuilder(t, caseSensitive) +} + +// NewReplaceSortOrderBuilder creates a builder for replacing the default sort order. +func NewReplaceSortOrderBuilder(t *Transaction, caseSensitive bool) *ReplaceSortOrderBuilder { + return &ReplaceSortOrderBuilder{txn: t, caseSensitive: caseSensitive} +} + +// Asc adds a column sorted in ascending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Asc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortASC, nullOrder) +} + +// Desc adds a column sorted in descending order with the requested null ordering. +func (u *ReplaceSortOrderBuilder) Desc(column string, nullOrder NullOrder) *ReplaceSortOrderBuilder { + return u.AddField(column, iceberg.IdentityTransform{}, SortDESC, nullOrder) +} + +// AddField adds a transformed sort field. A nil or incompatible transform is +// rejected when the replacement is applied or committed. Review Comment: A nested column like address.city is accepted, but right now our go writers sort only by top-level columns, so every later Append fails until the order is replaced. iceberg-java allows nested sorts, so I'd keep accepting them and document this limit on `Asc`, `Desc` and `AddField`, whats your take on this? ########## table/replace_sort_order_builder.go: ########## @@ -0,0 +1,192 @@ +// 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" + "strings" + + "github.com/apache/iceberg-go" +) + +// ReplaceSortOrderBuilder builds a replacement for the table's default sort order. +// Replacement starts with no fields, rather than copying the existing order. +// Calling Commit stages the change in the transaction; Transaction.Commit +// persists it to the catalog. +type ReplaceSortOrderBuilder struct { + txn *Transaction + caseSensitive bool + fields []sortOrderColumn +} + +type sortOrderColumn struct { + name string + transform iceberg.Transform + direction SortDirection + nullOrder NullOrder +} + +// NewReplaceSortOrder starts a sort order replacement in this transaction. +// Names are resolved using the transaction's current schema. An empty +// replacement resets the default sort order to unsorted (order ID 0). Review Comment: please add a "Sort order evolution" section to `website/src/api.md`, after "Partition evolution", showing this builder and when to call `ReplaceSortOrder` directly instead, neither is documented there yet. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
