You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "lliangyu-lin (via GitHub)" <gi...@apache.org> on 2025/06/24 21:02:22 UTC
[PR] [WIP] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
lliangyu-lin opened a new pull request, #467:
URL: https://github.com/apache/iceberg-go/pull/467
### Description
* Add support for update spec in transaction API
* Reference: https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/update/spec.py
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195720556
##########
partitions.go:
##########
@@ -28,7 +28,7 @@ import (
)
const (
- partitionDataIDStart = 1000
+ PartitionDataIDStart = 1000
InitialPartitionSpecID = 0
)
Review Comment:
It's used here to when trying to get the last assigned partition field id across all partition specs and that could be nil. And this is different than last assigned id in the current spec, otherwise we could of use https://github.com/apache/iceberg-go/blob/main/partitions.go#L200
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195803110
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
Review Comment:
I agree with you. I realized Go will not allow chaining when you return Error as part of the result when I started writing the unit tests...
For `Change all the methods to return only error`, this still won't allow chaining right? Or do you mean returning only `*UpdateSpec`.
For Builder approach, how should handle the errors if one of the chained methods failed? This applies to any approach for chaining I guess, I think we might need to maintain an error message field within the struct?
What do you think?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195720556
##########
partitions.go:
##########
@@ -28,7 +28,7 @@ import (
)
const (
- partitionDataIDStart = 1000
+ PartitionDataIDStart = 1000
InitialPartitionSpecID = 0
)
Review Comment:
It's used [here](https://github.com/apache/iceberg-go/pull/467/files#diff-eb806166d1b9d570a36b8506d2b9ada56126a32ca94b1eb12d6de03973e292c2R54) to when trying to get the last assigned partition field id across all partition specs and that could be nil. And this is different than last assigned id in the current spec, otherwise we could of use https://github.com/apache/iceberg-go/blob/main/partitions.go#L200
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195814663
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
Review Comment:
> For Change all the methods to return only error, this still won't allow chaining right?
Correct, that would not allow chaining. If we're going to return the error individually from the methods then we should simplify it and just return the error since we can't do chaining that way anyways.
To do chaining with the Builder approach, the methods wouldn't actually *perform* the operations, they would essentially create a stack of operations to perform that we delay executing until `Build()` is called. `Build` can then just go through the list of chained operations until either they all complete or one of them returns an error.
For instance something maybe like:
```go
func (b *UpdateSpec) AddField(...) *UpdateSpec {
b.operations = append(b.operations, b.addField(...))
return b
}
...
func (b *UpdateSpec) BuildUpdates() ([]Update, []Requirement, error) {
for _, op := range b.operations {
if err := op(); err != nil {
return nil, nil, err
}
}
// build updates and requirements and return them
}
```
Does that kinda show what I mean?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2205319520
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
Review Comment:
`CommitUpdates` and `BuildChanges()` seems like performing the same logic, so I only kept the `BuildUpdates` with the error in the signature.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #467:
URL: https://github.com/apache/iceberg-go/pull/467
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2178445663
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
Review Comment:
The whole benefit to returning `*UpdateSpec` is so that you can chain these calls. If we're also returning the error then we, by definition, are preventing the chaining unless someone creates a `must` function which seems bad.
So either we should change the signatures to just return only an `error` instead of `*UpdateSpec` or we should change this to be more of a Builder pattern where the actual function calls just populate the list of operations (AddField, RemoveField, Rename, etc.) but don't actually attempt to *perform* the functions until something like `UpdateSpec.Build()` is called, at which point it returns the error.
Basically, I'm proposing we should modify this to one of these two patterns:
1. Change all the methods to return *only* `error` instead of `(*UpdateSpec, error)`
2. Change the structure of the `UpdateSpec` struct to facilitate something like: `err := (&UpdateSpec{}).AddField(...).RemoveField(...).RenameField(...).Build()`.
I think both of those would be cleaner than this current function signature.
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
+ newSpec := us.Apply()
+ updates := make([]Update, 0)
+ requirements := make([]Requirement, 0)
+
+ if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.isNewPartitionSpec(newSpec.ID()) {
+ updates = append(updates, NewAddPartitionSpecUpdate(newSpec, false))
+ updates = append(updates, NewSetDefaultSpecUpdate(-1))
+ } else {
+ updates = append(updates, NewSetDefaultSpecUpdate(newSpec.ID()))
+ }
+ requiredLastAssignedPartitionId := us.txn.tbl.Metadata().LastPartitionSpecID()
+ requirements = append(requirements, AssertLastAssignedPartitionID(*requiredLastAssignedPartitionId))
+ }
+
+ return updates, requirements, nil
+}
+
+func (us *UpdateSpec) partitionField(key transformKey, name string) (iceberg.PartitionField, error) {
+ if us.txn.tbl.Metadata().Version() == 2 {
+ sourceId, transform := key.SourceId, key.Transform
+ historicalFields := make([]iceberg.PartitionField, 0)
+ for _, spec := range us.txn.tbl.Metadata().PartitionSpecs() {
+ for field := range spec.Fields() {
+ historicalFields = append(historicalFields, field)
+ }
Review Comment:
```suggestion
historicalFields = slices.AppendSeq(historicalFields, spec.Fields())
```
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
+ newSpec := us.Apply()
+ updates := make([]Update, 0)
+ requirements := make([]Requirement, 0)
+
+ if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.isNewPartitionSpec(newSpec.ID()) {
+ updates = append(updates, NewAddPartitionSpecUpdate(newSpec, false))
+ updates = append(updates, NewSetDefaultSpecUpdate(-1))
+ } else {
+ updates = append(updates, NewSetDefaultSpecUpdate(newSpec.ID()))
+ }
+ requiredLastAssignedPartitionId := us.txn.tbl.Metadata().LastPartitionSpecID()
+ requirements = append(requirements, AssertLastAssignedPartitionID(*requiredLastAssignedPartitionId))
+ }
+
+ return updates, requirements, nil
+}
+
+func (us *UpdateSpec) partitionField(key transformKey, name string) (iceberg.PartitionField, error) {
+ if us.txn.tbl.Metadata().Version() == 2 {
+ sourceId, transform := key.SourceId, key.Transform
+ historicalFields := make([]iceberg.PartitionField, 0)
+ for _, spec := range us.txn.tbl.Metadata().PartitionSpecs() {
+ for field := range spec.Fields() {
+ historicalFields = append(historicalFields, field)
+ }
+ }
+ for _, fields := range historicalFields {
+ if fields.SourceID == sourceId && fields.Transform.String() == transform {
+ return iceberg.PartitionField{
+ SourceID: sourceId,
+ FieldID: fields.FieldID,
+ Name: name,
+ Transform: fields.Transform,
+ }, nil
+ }
+ }
+ }
+ newFieldId := us.newFieldId()
+ transform, _ := iceberg.ParseTransform(key.Transform)
+ if name == "" {
+ tmp_field := iceberg.PartitionField{
+ SourceID: key.SourceId,
+ FieldID: newFieldId,
+ Name: "",
+ Transform: transform,
+ }
+ var err error
+ name, err = iceberg.GeneratePartitionFieldName(us.txn.tbl.Schema(), tmp_field)
+ if err != nil {
+ return iceberg.PartitionField{}, err
+ }
+ }
+
+ return iceberg.PartitionField{
+ SourceID: key.SourceId,
+ FieldID: newFieldId,
+ Name: name,
+ Transform: transform,
+ }, nil
+}
+
+func (us *UpdateSpec) newFieldId() int {
+ us.lastAssignedFieldId += 1
+
+ return us.lastAssignedFieldId
+}
+
+func (us *UpdateSpec) isDuplicatePartition(transform iceberg.Transform, partitionField iceberg.PartitionField) bool {
+ _, exists := us.deletes[partitionField.FieldID]
+
+ return !exists && transform == partitionField.Transform
+}
+
+func (us *UpdateSpec) checkAndAddPartitionName(schema *iceberg.Schema, name string, sourceId int, partitionNames map[string]bool) error {
+ field, found := schema.FindFieldByName(name)
+ if found && field.ID != sourceId {
+ return fmt.Errorf("cannot create partition from name that exists in schema %s", name)
+ }
+ if _, exists := partitionNames[name]; exists {
+ return fmt.Errorf("partition name has to be unique: %s", name)
+ }
+ partitionNames[name] = true
+
+ return nil
+}
+
+func (us *UpdateSpec) addNewField(schema *iceberg.Schema, sourceId int, fieldId int, name string, transform iceberg.Transform, partitionNames map[string]bool) (iceberg.PartitionField, error) {
+ err := us.checkAndAddPartitionName(schema, name, sourceId, partitionNames)
+ if err != nil {
+ return iceberg.PartitionField{}, err
+ }
+
+ return iceberg.PartitionField{
+ SourceID: sourceId,
+ FieldID: fieldId,
+ Name: name,
+ Transform: transform,
+ }, nil
+}
+
+func (us *UpdateSpec) isNewPartitionSpec(newSpecId int) bool {
+ for _, spec := range us.txn.tbl.Metadata().PartitionSpecs() {
+ if spec.ID() == newSpecId {
+ return false
+ }
+ }
+
+ return true
Review Comment:
maybe
```go
return !slices.ContainsFunc(us.txn.tbl.Metadata().PartitionSpecs(), func(s iceberg.PartitionSpec) bool {
return s.ID() == newSpecId
})
```
##########
partitions.go:
##########
@@ -28,7 +28,7 @@ import (
)
const (
- partitionDataIDStart = 1000
+ PartitionDataIDStart = 1000
InitialPartitionSpecID = 0
)
Review Comment:
Is there a particular reason we need to expose this constant?
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
Review Comment:
same comment as above, no need to explicitly call `.String()`
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
+ newSpec := us.Apply()
+ updates := make([]Update, 0)
+ requirements := make([]Requirement, 0)
+
+ if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.isNewPartitionSpec(newSpec.ID()) {
+ updates = append(updates, NewAddPartitionSpecUpdate(newSpec, false))
+ updates = append(updates, NewSetDefaultSpecUpdate(-1))
+ } else {
+ updates = append(updates, NewSetDefaultSpecUpdate(newSpec.ID()))
+ }
+ requiredLastAssignedPartitionId := us.txn.tbl.Metadata().LastPartitionSpecID()
+ requirements = append(requirements, AssertLastAssignedPartitionID(*requiredLastAssignedPartitionId))
+ }
+
+ return updates, requirements, nil
+}
+
+func (us *UpdateSpec) partitionField(key transformKey, name string) (iceberg.PartitionField, error) {
+ if us.txn.tbl.Metadata().Version() == 2 {
+ sourceId, transform := key.SourceId, key.Transform
+ historicalFields := make([]iceberg.PartitionField, 0)
+ for _, spec := range us.txn.tbl.Metadata().PartitionSpecs() {
+ for field := range spec.Fields() {
+ historicalFields = append(historicalFields, field)
+ }
+ }
+ for _, fields := range historicalFields {
+ if fields.SourceID == sourceId && fields.Transform.String() == transform {
+ return iceberg.PartitionField{
+ SourceID: sourceId,
+ FieldID: fields.FieldID,
+ Name: name,
+ Transform: fields.Transform,
+ }, nil
+ }
+ }
+ }
+ newFieldId := us.newFieldId()
+ transform, _ := iceberg.ParseTransform(key.Transform)
+ if name == "" {
+ tmp_field := iceberg.PartitionField{
+ SourceID: key.SourceId,
+ FieldID: newFieldId,
+ Name: "",
+ Transform: transform,
+ }
+ var err error
+ name, err = iceberg.GeneratePartitionFieldName(us.txn.tbl.Schema(), tmp_field)
+ if err != nil {
+ return iceberg.PartitionField{}, err
+ }
+ }
+
+ return iceberg.PartitionField{
+ SourceID: key.SourceId,
+ FieldID: newFieldId,
+ Name: name,
+ Transform: transform,
+ }, nil
+}
+
+func (us *UpdateSpec) newFieldId() int {
+ us.lastAssignedFieldId += 1
+
+ return us.lastAssignedFieldId
+}
+
+func (us *UpdateSpec) isDuplicatePartition(transform iceberg.Transform, partitionField iceberg.PartitionField) bool {
+ _, exists := us.deletes[partitionField.FieldID]
+
+ return !exists && transform == partitionField.Transform
Review Comment:
should use `transform.Equals(partitionField.Transform)`
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
Review Comment:
You don't need to explicitly call `.String()` when passing things to `fmt.Errorf` since you're using `%s` it will automatically use the `.String()` to create the string. So you can simplify this a bit and remove those explicit calls
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
Review Comment:
it doesn't look like there's anything that can return an error here, can we remove the `error` from the signature?
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
Review Comment:
```suggestion
partitionFields = append(partitionFields, us.adds...)
```
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
Review Comment:
Is there a good reason to return a `*iceberg.PartitionSpec` vs just `iceberg.PartitionSpec`?
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
+ newSpec := us.Apply()
+ updates := make([]Update, 0)
+ requirements := make([]Requirement, 0)
+
+ if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.isNewPartitionSpec(newSpec.ID()) {
+ updates = append(updates, NewAddPartitionSpecUpdate(newSpec, false))
+ updates = append(updates, NewSetDefaultSpecUpdate(-1))
+ } else {
+ updates = append(updates, NewSetDefaultSpecUpdate(newSpec.ID()))
+ }
+ requiredLastAssignedPartitionId := us.txn.tbl.Metadata().LastPartitionSpecID()
+ requirements = append(requirements, AssertLastAssignedPartitionID(*requiredLastAssignedPartitionId))
+ }
+
+ return updates, requirements, nil
+}
+
+func (us *UpdateSpec) partitionField(key transformKey, name string) (iceberg.PartitionField, error) {
+ if us.txn.tbl.Metadata().Version() == 2 {
+ sourceId, transform := key.SourceId, key.Transform
+ historicalFields := make([]iceberg.PartitionField, 0)
+ for _, spec := range us.txn.tbl.Metadata().PartitionSpecs() {
+ for field := range spec.Fields() {
+ historicalFields = append(historicalFields, field)
+ }
+ }
+ for _, fields := range historicalFields {
+ if fields.SourceID == sourceId && fields.Transform.String() == transform {
+ return iceberg.PartitionField{
Review Comment:
i think this also needs to check `if len(name) > 0 || field.Name == name` before choosing to return the field.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195720556
##########
partitions.go:
##########
@@ -28,7 +28,7 @@ import (
)
const (
- partitionDataIDStart = 1000
+ PartitionDataIDStart = 1000
InitialPartitionSpecID = 0
)
Review Comment:
It's used [here](https://github.com/apache/iceberg-go/pull/467/files#diff-eb806166d1b9d570a36b8506d2b9ada56126a32ca94b1eb12d6de03973e292c2R54) to when trying to get the last assigned partition field id across all partition specs and that could be nil. And this is different than last assigned id in the current spec, otherwise we could of use https://github.com/apache/iceberg-go/blob/main/partitions.go#L200
Reference in python: https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/update/spec.py#L71
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2195907682
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
Review Comment:
I see. Thanks for the clarifications!
I like the builder approach in this case. Will try to update this locally and do some testings.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2205319520
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
+ partitionFields := make([]iceberg.PartitionField, 0)
+ partitionNames := make(map[string]bool)
+ spec := us.txn.tbl.Metadata().PartitionSpec()
+ for field := range spec.Fields() {
+ var newField iceberg.PartitionField
+ var err error
+ if _, deleted := us.deletes[field.FieldID]; !deleted {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, field.Transform, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, field.Transform, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ } else if us.txn.tbl.Metadata().Version() == 1 {
+ if rename, renamed := us.renames[field.Name]; renamed {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, rename, iceberg.VoidTransform{}, partitionNames)
+ } else {
+ newField, err = us.addNewField(us.txn.tbl.Schema(), field.SourceID, field.FieldID, field.Name, iceberg.VoidTransform{}, partitionNames)
+ }
+ if err != nil {
+ return nil
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+ }
+
+ for _, field := range us.adds {
+ newField := iceberg.PartitionField{
+ SourceID: field.SourceID,
+ FieldID: field.FieldID,
+ Name: field.Name,
+ Transform: field.Transform,
+ }
+ partitionFields = append(partitionFields, newField)
+ }
+
+ newSpec := iceberg.NewPartitionSpec(partitionFields...)
+ newSpecId := iceberg.InitialPartitionSpecID
+ for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ if newSpec.CompatibleWith(&spec) {
+ newSpecId = spec.ID()
+
+ break
+ } else if newSpecId <= spec.ID() {
+ newSpecId = spec.ID() + 1
+ }
+ }
+ newSpec = iceberg.NewPartitionSpecID(newSpecId, partitionFields...)
+
+ return &newSpec
+}
+
+func (us *UpdateSpec) Commit() error {
+ updates, requirements, err := us.CommitUpdates()
+ if err != nil {
+ return err
+ }
+
+ if len(updates) == 0 {
+ return nil
+ }
+
+ return us.txn.apply(updates, requirements)
+}
+
+func (us *UpdateSpec) CommitUpdates() ([]Update, []Requirement, error) {
Review Comment:
`CommitUpdates` and `BuildChanges()` seems like performing the same logic expect `BuildChanges` goes through the operation queue, so I only kept the `BuildUpdates` with the error in the signature.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "lliangyu-lin (via GitHub)" <gi...@apache.org>.
lliangyu-lin commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2205306733
##########
table/update_spec.go:
##########
@@ -0,0 +1,367 @@
+// 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 (
+ "errors"
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
+ txn *Transaction
+ nameToField map[string]iceberg.PartitionField
+ nameToAddedField map[string]iceberg.PartitionField
+ transformToField map[transformKey]iceberg.PartitionField
+ transformToAddedField map[transformKey]iceberg.PartitionField
+ renames map[string]string
+ addedTimeFields map[int]iceberg.PartitionField
+ caseSensitive bool
+ adds []iceberg.PartitionField
+ deletes map[int]bool
+ lastAssignedFieldId int
+}
+
+func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
+ transformToField := make(map[transformKey]iceberg.PartitionField)
+ nameToField := make(map[string]iceberg.PartitionField)
+ partitionSpec := t.tbl.Metadata().PartitionSpec()
+ for partitionField := range partitionSpec.Fields() {
+ transformToField[transformKey{
+ SourceId: partitionField.SourceID,
+ Transform: partitionField.Transform.String(),
+ }] = partitionField
+ nameToField[partitionField.Name] = partitionField
+ }
+ lastAssignedFieldId := t.tbl.Metadata().LastPartitionSpecID()
+ if lastAssignedFieldId == nil {
+ v := iceberg.PartitionDataIDStart - 1
+ lastAssignedFieldId = &v
+ }
+
+ return &UpdateSpec{
+ txn: t,
+ nameToField: nameToField,
+ nameToAddedField: make(map[string]iceberg.PartitionField),
+ transformToField: transformToField,
+ transformToAddedField: make(map[transformKey]iceberg.PartitionField),
+ renames: make(map[string]string),
+ addedTimeFields: make(map[int]iceberg.PartitionField),
+ caseSensitive: caseSensitive,
+ adds: make([]iceberg.PartitionField, 0),
+ deletes: make(map[int]bool),
+ lastAssignedFieldId: *lastAssignedFieldId,
+ }
+}
+
+type transformKey struct {
+ SourceId int
+ Transform string
+}
+
+func (us *UpdateSpec) AddField(sourceColName string, transform iceberg.Transform, partitionFieldName string) (*UpdateSpec, error) {
+ // Finds the column in the schema and binds it with case sensitivity.
+ ref := iceberg.Reference(sourceColName)
+ boundTerm, err := ref.Bind(us.txn.tbl.Schema(), us.caseSensitive)
+ if err != nil {
+ return nil, err
+ }
+
+ // Validate the transform
+ outputType := boundTerm.Type()
+ if !transform.CanTransform(outputType) {
+ return nil, fmt.Errorf("%s cannot transform %s values from %s", transform.String(), outputType.String(), boundTerm.Ref().Field().Name)
+ }
+
+ // Check for duplicate transform on same column
+ key := transformKey{
+ SourceId: boundTerm.Ref().Field().ID,
+ Transform: transform.String(),
+ }
+ existingPartitionField, exists := us.transformToField[key]
+ if exists && us.isDuplicatePartition(transform, existingPartitionField) {
+ return nil, fmt.Errorf("duplicate partition field for %s=%v, %v already exists", ref.String(), ref, existingPartitionField)
+ }
+
+ // Check if this transform was already added
+ added, exists := us.transformToAddedField[key]
+ if exists {
+ return nil, fmt.Errorf("already added partition: %s ", added.Name)
+ }
+
+ // Create the new partition field and Check for name collisions
+ // with existing fields
+ newField, err := us.partitionField(key, partitionFieldName)
+ if err != nil {
+ return nil, err
+ }
+ if _, exists = us.nameToAddedField[newField.Name]; exists {
+ return nil, fmt.Errorf("already added partition field with name: %s", newField.Name)
+ }
+
+ // Handle special case for time transforms
+ if _, isTimeTransform := newField.Transform.(iceberg.TimeTransform); isTimeTransform {
+ if existingTimeField, exists := us.addedTimeFields[newField.SourceID]; exists {
+ return nil, fmt.Errorf("cannot add time partition field: %s conflicts with %s", newField.Name, existingTimeField.Name)
+ }
+ us.addedTimeFields[newField.SourceID] = newField
+ }
+ us.transformToAddedField[key] = newField
+
+ // If name matches an existing field, rename it (if VOID)
+ existingPartitionField, exists = us.nameToField[newField.Name]
+ if _, inDelete := us.deletes[existingPartitionField.FieldID]; exists && !inDelete {
+ if _, isVoidTransform := existingPartitionField.Transform.(iceberg.VoidTransform); isVoidTransform {
+ _, err = us.RenameField(existingPartitionField.Name, fmt.Sprintf("%s_%d", existingPartitionField.Name, existingPartitionField.FieldID))
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ return nil, fmt.Errorf("cannot add duplicate partition field name: %s", existingPartitionField.Name)
+ }
+ }
+
+ // Register the new field
+ us.nameToAddedField[newField.Name] = newField
+ us.adds = append(us.adds, newField)
+
+ return us, nil
+}
+
+func (us *UpdateSpec) AddIdentity(sourceColName string) (*UpdateSpec, error) {
+ return us.AddField(sourceColName, iceberg.IdentityTransform{}, "")
+}
+
+func (us *UpdateSpec) RemoveField(name string) (*UpdateSpec, error) {
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, fmt.Errorf("cannot remove newly added field %s", name)
+ }
+ if _, renamed := us.renames[name]; renamed {
+ return nil, fmt.Errorf("cannot rename and delete field %s", name)
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field %s", name)
+ }
+ us.deletes[field.FieldID] = true
+
+ return us, nil
+}
+
+func (us *UpdateSpec) RenameField(name string, newName string) (*UpdateSpec, error) {
+ existingField, exists := us.nameToField[newName]
+ _, isVoidTransform := existingField.Transform.(iceberg.VoidTransform)
+ if exists && isVoidTransform {
+ return us.RenameField(name, fmt.Sprintf("%s_%d", name, existingField.FieldID))
+ }
+ if _, added := us.nameToAddedField[name]; added {
+ return nil, errors.New("cannot rename recently added partitions")
+ }
+ field, exists := us.nameToField[name]
+ if !exists {
+ return nil, fmt.Errorf("cannot find partition field: %s", name)
+ }
+ if _, deleted := us.deletes[field.FieldID]; deleted {
+ return nil, fmt.Errorf("cannot delete and rename partition field: %s", name)
+ }
+ us.renames[name] = newName
+
+ return us, nil
+}
+
+func (us *UpdateSpec) Apply() *iceberg.PartitionSpec {
Review Comment:
Not really, I checked again and seems like `iceberg.PartitionSpec` makes more sense since we unlikely to mutate the spec with additional changes.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(transaction): Add initial support for update spec in transaction API [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #467:
URL: https://github.com/apache/iceberg-go/pull/467#discussion_r2205363999
##########
table/update_spec.go:
##########
@@ -0,0 +1,394 @@
+// 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 (
+ "errors"
+ "fmt"
+ "slices"
+
+ "github.com/apache/iceberg-go"
+)
+
+type UpdateSpec struct {
Review Comment:
Needs a docstring
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org