This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 569a427a4 refactor(partitions,sorting)!: make first-source access
explicit (#2304)
569a427a4 is described below
commit 569a427a472be4fb555375240cd56b44c0e5cd08
Author: Minh Vu <[email protected]>
AuthorDate: Sun Oct 25 10:10:53 2026 +0100
refactor(partitions,sorting)!: make first-source access explicit (#2304)
---
catalog/rest/scan_task_decoder.go | 2 +-
cmd/iceberg/utils_test.go | 4 ++--
data_file_codec.go | 2 +-
partitions.go | 24 +++++++++++-----------
partitions_test.go | 37 +++++++++++++++++++++++++---------
table/conflict_validation.go | 4 ++--
table/equality_delete_writer.go | 8 ++++----
table/internal/utils.go | 10 ++++-----
table/metadata.go | 4 ++--
table/metadata_internal_test.go | 8 ++++----
table/partition_conflict_bench_test.go | 4 ++--
table/partitioned_fanout_writer.go | 10 ++++-----
table/sort_keys.go | 4 ++--
table/sort_keys_bench_test.go | 4 ++--
table/sorting.go | 15 +++++++-------
table/sorting_test.go | 23 ++++++++++++++++++---
table/update_spec.go | 14 ++++++-------
table/update_spec_test.go | 20 +++++++++---------
website/src/releases.md | 2 ++
19 files changed, 118 insertions(+), 81 deletions(-)
diff --git a/catalog/rest/scan_task_decoder.go
b/catalog/rest/scan_task_decoder.go
index f61c667d3..0d852837d 100644
--- a/catalog/rest/scan_task_decoder.go
+++ b/catalog/rest/scan_task_decoder.go
@@ -608,7 +608,7 @@ func newPartitionDecodePlan(spec *iceberg.PartitionSpec,
metadata table.ScanPlan
fieldPlan := partitionDecodeField{
fieldID: field.FieldID,
fieldName: field.Name,
- sourceID: field.SourceID(),
+ sourceID: field.FirstSourceID(),
}
sourceType, ok := findFieldType(fieldPlan.sourceID, metadata)
fieldPlan.sourceTypeFound = ok
diff --git a/cmd/iceberg/utils_test.go b/cmd/iceberg/utils_test.go
index 69f7c4929..0d935f9dd 100644
--- a/cmd/iceberg/utils_test.go
+++ b/cmd/iceberg/utils_test.go
@@ -344,8 +344,8 @@ func TestParseSortOrder(t *testing.T) {
// Validate sort directions and null orders
i := 0
for _, field := range got.Fields() {
- if i < len(tt.expectedSourceIDs) &&
field.SourceID() != tt.expectedSourceIDs[i] {
- t.Errorf("parseSortOrder()
field %d source ID = %v, expected %v", i, field.SourceID(),
tt.expectedSourceIDs[i])
+ if i < len(tt.expectedSourceIDs) &&
field.FirstSourceID() != tt.expectedSourceIDs[i] {
+ t.Errorf("parseSortOrder()
field %d source ID = %v, expected %v", i, field.FirstSourceID(),
tt.expectedSourceIDs[i])
}
if i < len(tt.expectedDirections) &&
field.Direction != tt.expectedDirections[i] {
t.Errorf("parseSortOrder()
field %d direction = %v, expected %v", i, field.Direction,
tt.expectedDirections[i])
diff --git a/data_file_codec.go b/data_file_codec.go
index b11fec574..f9897c030 100644
--- a/data_file_codec.go
+++ b/data_file_codec.go
@@ -333,7 +333,7 @@ func partitionSchemaFingerprint(spec PartitionSpec, schema
*Schema) (string, err
sourceType := Type(UnknownType{})
// ResultType only inspects the source type here; borrow it to
avoid
// cloning nested source types on every schema-cache hit.
- if sourceField, ok := schema.FindFieldByIDRef(field.SourceID(),
internal.SchemaRef{}); ok {
+ if sourceField, ok :=
schema.FindFieldByIDRef(field.FirstSourceID(), internal.SchemaRef{}); ok {
sourceType = sourceField.Type
}
resultType := field.Transform.ResultType(sourceType)
diff --git a/partitions.go b/partitions.go
index 38867e267..396b5e16f 100644
--- a/partitions.go
+++ b/partitions.go
@@ -69,10 +69,10 @@ type PartitionField struct {
escapedName string
}
-// SourceID returns the first source column id. For single-argument transforms
-// this is the only source column. For multi-argument transforms this is the
-// first source column.
-func (p PartitionField) SourceID() int {
+// FirstSourceID returns the first source column id, or 0 when SourceIDs is
+// empty. For single-argument transforms this is the only source column; for
+// multi-argument transforms the other ids are only in SourceIDs.
+func (p PartitionField) FirstSourceID() int {
if len(p.SourceIDs) == 0 {
return 0
}
@@ -114,7 +114,7 @@ func (p PartitionField) MarshalJSON() ([]byte, error) {
FieldID int `json:"field-id"`
Name string `json:"name"`
Transform Transform `json:"transform"`
- }{p.SourceID(), p.FieldID, p.Name, p.Transform})
+ }{p.FirstSourceID(), p.FieldID, p.Name, p.Transform})
}
func (p PartitionField) Equals(other PartitionField) bool {
@@ -129,7 +129,7 @@ func (p *PartitionField) String() string {
return fmt.Sprintf("%d: %s: %s(%v)", p.FieldID, p.Name,
p.Transform, p.SourceIDs)
}
- return fmt.Sprintf("%d: %s: %s(%d)", p.FieldID, p.Name, p.Transform,
p.SourceID())
+ return fmt.Sprintf("%d: %s: %s(%d)", p.FieldID, p.Name, p.Transform,
p.FirstSourceID())
}
func (p *PartitionField) UnmarshalJSON(b []byte) error {
@@ -313,7 +313,7 @@ func (p *PartitionSpec) BindToSchema(schema *Schema,
lastPartitionID *int, newSp
continue
}
}
- opts = append(opts,
AddPartitionFieldBySourceID(field.SourceID(), field.Name, field.Transform,
schema, &field.FieldID))
+ opts = append(opts,
AddPartitionFieldBySourceID(field.FirstSourceID(), field.Name, field.Transform,
schema, &field.FieldID))
}
// Replay, not authoring: the spec may come from another client, so it
only
@@ -760,7 +760,7 @@ func (ps *PartitionSpec) initialize() {
for i := range ps.fields {
ps.fields[i].escapedName = url.QueryEscape(ps.fields[i].Name)
- ps.sourceIdToFields[ps.fields[i].SourceID()] =
append(ps.sourceIdToFields[ps.fields[i].SourceID()], ps.fields[i])
+ ps.sourceIdToFields[ps.fields[i].FirstSourceID()] =
append(ps.sourceIdToFields[ps.fields[i].FirstSourceID()], ps.fields[i])
}
}
@@ -867,7 +867,7 @@ func (ps *PartitionSpec) resolvedPartitionFields(schema
*Schema) []resolvedParti
fields := make([]resolvedPartitionField, 0, len(ps.fields))
for _, field := range ps.fields {
sourceType := Type(UnknownType{})
- if sourceField, ok := schema.FindFieldByIDRef(field.SourceID(),
internal.SchemaRef{}); ok {
+ if sourceField, ok :=
schema.FindFieldByIDRef(field.FirstSourceID(), internal.SchemaRef{}); ok {
sourceType = sourceField.Type
}
@@ -899,7 +899,7 @@ func (ps *PartitionSpec) PartitionType(schema *Schema)
*StructType {
nestedFields := make([]NestedField, len(ps.fields))
for i, field := range ps.fields {
sourceType := Type(UnknownType{})
- if typ, ok := schema.FindTypeByID(field.SourceID()); ok {
+ if typ, ok := schema.FindTypeByID(field.FirstSourceID()); ok {
sourceType = typ
}
@@ -963,9 +963,9 @@ func GeneratePartitionFieldName(schema *Schema, field
PartitionField) (string, e
return field.Name, nil
}
- sourceName, exists := schema.FindColumnName(field.SourceID())
+ sourceName, exists := schema.FindColumnName(field.FirstSourceID())
if !exists {
- return "", fmt.Errorf("could not find field with id %d",
field.SourceID())
+ return "", fmt.Errorf("could not find field with id %d",
field.FirstSourceID())
}
transform := field.Transform
diff --git a/partitions_test.go b/partitions_test.go
index 542b3e87d..274dfe042 100644
--- a/partitions_test.go
+++ b/partitions_test.go
@@ -772,7 +772,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
var field iceberg.PartitionField
err := json.Unmarshal([]byte(jsonData), &field)
require.NoError(t, err)
- assert.Equal(t, 1, field.SourceID())
+ assert.Equal(t, 1, field.FirstSourceID())
assert.Equal(t, 1000, field.FieldID)
assert.Equal(t, "str_truncate", field.Name)
assert.Equal(t, iceberg.TruncateTransform{Width: 19},
field.Transform)
@@ -789,7 +789,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
var field iceberg.PartitionField
err := json.Unmarshal([]byte(jsonData), &field)
require.NoError(t, err)
- assert.Equal(t, 2, field.SourceID())
+ assert.Equal(t, 2, field.FirstSourceID())
assert.Equal(t, 1001, field.FieldID)
assert.Equal(t, "int_bucket", field.Name)
assert.Equal(t, iceberg.BucketTransform{NumBuckets: 25},
field.Transform)
@@ -806,7 +806,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
var field iceberg.PartitionField
err := json.Unmarshal([]byte(jsonData), &field)
require.NoError(t, err)
- assert.Equal(t, 2, field.SourceID(), "SourceID should be first
element")
+ assert.Equal(t, 2, field.FirstSourceID(), "FirstSourceID should
return the first element")
assert.Equal(t, []int{2, 3}, field.SourceIDs, "SourceIDs should
contain all elements")
assert.Equal(t, 1001, field.FieldID)
assert.Equal(t, "int_bucket", field.Name)
@@ -827,7 +827,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
var decoded iceberg.PartitionField
err = json.Unmarshal(data, &decoded)
require.NoError(t, err)
- assert.Equal(t, 2, decoded.SourceID())
+ assert.Equal(t, 2, decoded.FirstSourceID())
assert.Equal(t, []int{2, 3}, decoded.SourceIDs)
})
@@ -886,7 +886,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
}`
var field iceberg.PartitionField
require.NoError(t, json.Unmarshal([]byte(jsonData), &field))
- assert.Equal(t, 0, field.SourceID())
+ assert.Equal(t, 0, field.FirstSourceID())
assert.Equal(t, []int{0}, field.SourceIDs)
})
@@ -895,7 +895,7 @@ func TestPartitionFieldUnmarshalJSON(t *testing.T) {
require.NoError(t, json.Unmarshal([]byte(`{
"source-id": 1, "field-id": 1003, "transform": "void",
"name": "void"
}`), &field))
- assert.Equal(t, 1, field.SourceID())
+ assert.Equal(t, 1, field.FirstSourceID())
})
}
@@ -1172,7 +1172,7 @@ func
TestUnboundPartitionSpecUnmarshalAcceptsZeroSourceID(t *testing.T) {
var spec iceberg.UnboundPartitionSpec
require.NoError(t, json.Unmarshal([]byte(data), &spec))
- assert.Equal(t, 0, spec.Field(0).SourceID())
+ assert.Equal(t, 0, spec.Field(0).FirstSourceID())
roundTripped, err := json.Marshal(spec)
require.NoError(t, err)
@@ -1191,8 +1191,8 @@ func
TestUnboundPartitionSpecUnmarshalAcceptsRepeatedZeroSourceID(t *testing.T)
var spec iceberg.UnboundPartitionSpec
require.NoError(t, json.Unmarshal([]byte(data), &spec))
require.Equal(t, 2, spec.NumFields())
- assert.Equal(t, 0, spec.Field(0).SourceID())
- assert.Equal(t, 0, spec.Field(1).SourceID())
+ assert.Equal(t, 0, spec.Field(0).FirstSourceID())
+ assert.Equal(t, 0, spec.Field(1).FirstSourceID())
}
func TestUnboundPartitionSpecUnmarshalRejectsInvalidStructure(t *testing.T) {
@@ -1294,7 +1294,7 @@ func TestUnboundPartitionSpecBindToSchema(t *testing.T) {
bound, err := spec.BindToSchema(requestSchema, nil, nil)
require.NoError(t, err)
require.Equal(t, 1, bound.NumFields())
- source, ok := requestSchema.FindFieldByID(bound.Field(0).SourceID())
+ source, ok :=
requestSchema.FindFieldByID(bound.Field(0).FirstSourceID())
require.True(t, ok)
assert.Equal(t, "floats", source.Name)
@@ -1692,3 +1692,20 @@ func TestNewPartitionSpecOptsTimeTransformMembership(t
*testing.T) {
})
}
}
+
+func TestPartitionFieldFirstSourceID(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ sourceIDs []int
+ want int
+ }{
+ {name: "empty", sourceIDs: nil, want: 0},
+ {name: "single", sourceIDs: []int{7}, want: 7},
+ {name: "multiple", sourceIDs: []int{7, 9}, want: 7},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ field := iceberg.PartitionField{SourceIDs: tc.sourceIDs}
+ assert.Equal(t, tc.want, field.FirstSourceID())
+ })
+ }
+}
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index e8b56a159..c47eee777 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -974,12 +974,12 @@ func eqDeletePartitionsToFilter(files []iceberg.DataFile,
meta Metadata) (iceber
if currentSchema == nil {
currentSchema = meta.CurrentSchema()
}
- field, found :=
currentSchema.FindFieldByID(pf.SourceID())
+ field, found :=
currentSchema.FindFieldByID(pf.FirstSourceID())
sourceField =
equalityDeleteSourceFieldInfo{name: field.Name, found: found}
specInfo.sourceFields[partFieldID] = sourceField
}
if !sourceField.found {
- return nil, fmt.Errorf("source field ID %d
(partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ return nil, fmt.Errorf("source field ID %d
(partition field %q) not found in schema", pf.FirstSourceID(), pf.Name)
}
value := p[partFieldID]
diff --git a/table/equality_delete_writer.go b/table/equality_delete_writer.go
index d35fcf493..852149354 100644
--- a/table/equality_delete_writer.go
+++ b/table/equality_delete_writer.go
@@ -316,14 +316,14 @@ func equalityDeleteWriteSchema(tableSchema
*iceberg.Schema, equalityFieldIDs []i
// Partition source columns not already in the equality key.
for _, f := range spec.Fields() {
- if _, ok := seen[f.SourceID()]; ok {
+ if _, ok := seen[f.FirstSourceID()]; ok {
continue
}
- name, ok := tableSchema.FindColumnName(f.SourceID())
+ name, ok := tableSchema.FindColumnName(f.FirstSourceID())
if !ok {
- return nil, fmt.Errorf("%w: partition source field ID
%d not found in table schema", iceberg.ErrInvalidSchema, f.SourceID())
+ return nil, fmt.Errorf("%w: partition source field ID
%d not found in table schema", iceberg.ErrInvalidSchema, f.FirstSourceID())
}
- seen[f.SourceID()] = struct{}{}
+ seen[f.FirstSourceID()] = struct{}{}
names = append(names, name)
}
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 8fb7f1e22..ca52adf4e 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -177,7 +177,7 @@ func PartitionRecordValue(field iceberg.PartitionField, val
iceberg.Literal, sch
return ret, nil
}
- f, ok := schema.FindFieldByIDRef(field.SourceID(),
iceberginternal.SchemaRef{})
+ f, ok := schema.FindFieldByIDRef(field.FirstSourceID(),
iceberginternal.SchemaRef{})
if !ok {
return ret, fmt.Errorf("%w: could not find source field in
schema for %s",
iceberg.ErrInvalidSchema, field.String())
@@ -239,7 +239,7 @@ type DataFileStatistics struct {
}
func (d *DataFileStatistics) PartitionValue(field iceberg.PartitionField, sc
*iceberg.Schema) any {
- agg, ok := d.ColAggs[field.SourceID()]
+ agg, ok := d.ColAggs[field.FirstSourceID()]
if !ok {
return nil
}
@@ -248,7 +248,7 @@ func (d *DataFileStatistics) PartitionValue(field
iceberg.PartitionField, sc *ic
return nil
}
- sourceField, ok := sc.FindFieldByIDRef(field.SourceID(),
iceberginternal.SchemaRef{})
+ sourceField, ok := sc.FindFieldByIDRef(field.FirstSourceID(),
iceberginternal.SchemaRef{})
if !ok {
panic(fmt.Errorf("%w: could not find source field in schema for
%s",
iceberg.ErrInvalidSchema, field.String()))
@@ -312,7 +312,7 @@ func (d *DataFileStatistics) ToDataFile(opts DataFileOpts)
iceberg.DataFile {
fieldIDToPartitionData = make(map[int]any,
opts.Spec.NumFields())
for _, field := range opts.Spec.Fields() {
sourceField, sourceFieldFound :=
opts.Schema.FindFieldByIDRef(
- field.SourceID(), iceberginternal.SchemaRef{})
+ field.FirstSourceID(),
iceberginternal.SchemaRef{})
partitionVal := opts.PartitionValues[field.FieldID]
switch {
case partitionVal != nil:
@@ -320,7 +320,7 @@ func (d *DataFileStatistics) ToDataFile(opts DataFileOpts)
iceberg.DataFile {
fieldIDToPartitionData[field.FieldID] =
partitionVal
case field.Transform.PreservesOrder() &&
sourceFieldFound:
fieldIDToPartitionData[field.FieldID] =
d.partitionValueWithSourceType(
- field, d.ColAggs[field.SourceID()],
sourceField.Type)
+ field,
d.ColAggs[field.FirstSourceID()], sourceField.Type)
case field.Transform.PreservesOrder():
// Preserve the existing error path for an
invalid schema. This branch
// is not part of the valid hot path because
sourceFieldFound is false.
diff --git a/table/metadata.go b/table/metadata.go
index 3caf45d71..6c0843afc 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -3727,8 +3727,8 @@ func reassignIDs(sc *iceberg.Schema, partitions
*iceberg.PartitionSpec, sortOrde
for _, f := range partitions.Fields() {
var s string
var ok bool
- if s, ok = previousMapFn(f.SourceID()); !ok {
- return nil, fmt.Errorf("%w: partition source field %d
not found in schema", ErrInvalidMetadata, f.SourceID())
+ if s, ok = previousMapFn(f.FirstSourceID()); !ok {
+ return nil, fmt.Errorf("%w: partition source field %d
not found in schema", ErrInvalidMetadata, f.FirstSourceID())
}
opts = append(opts, iceberg.AddPartitionFieldByName(s, f.Name,
f.Transform, freshSc, nil))
}
diff --git a/table/metadata_internal_test.go b/table/metadata_internal_test.go
index 7e6de78ee..9c9e3d3c2 100644
--- a/table/metadata_internal_test.go
+++ b/table/metadata_internal_test.go
@@ -1168,14 +1168,14 @@ func TestNewMetadataFromOrdinalNumberedRequest(t
*testing.T) {
gotSpec := meta.PartitionSpec()
require.Equal(t, len(tt.wantSourceIDs),
gotSpec.NumFields())
for i, wantSourceID := range tt.wantSourceIDs {
- assert.Equal(t, wantSourceID,
gotSpec.Field(i).SourceID())
+ assert.Equal(t, wantSourceID,
gotSpec.Field(i).FirstSourceID())
assert.Equal(t, tt.wantNames[i],
gotSpec.Field(i).Name)
}
// The sort order references the first column, so it
remaps to 1.
require.Equal(t, 1, meta.SortOrder().Len())
for _, field := range meta.SortOrder().Fields() {
- assert.Equal(t, wantSortID, field.SourceID())
+ assert.Equal(t, wantSortID,
field.FirstSourceID())
}
// Written metadata carries bound IDs only, so it
parses back as
@@ -1186,7 +1186,7 @@ func TestNewMetadataFromOrdinalNumberedRequest(t
*testing.T) {
require.NoError(t, err)
reparsedSpec := reparsed.PartitionSpec()
for i, wantSourceID := range tt.wantSourceIDs {
- assert.Equal(t, wantSourceID,
reparsedSpec.Field(i).SourceID())
+ assert.Equal(t, wantSourceID,
reparsedSpec.Field(i).FirstSourceID())
}
})
}
@@ -2102,7 +2102,7 @@ func TestTableMetadataV1PartitionSpecsWithoutDefaultId(t
*testing.T) {
require.Equal(t, spec.NumFields(), 1)
require.Equal(t, spec.Field(0).Name, "y")
require.Equal(t, spec.Field(0).Transform, iceberg.IdentityTransform{})
- require.Equal(t, spec.Field(0).SourceID(), 2)
+ require.Equal(t, spec.Field(0).FirstSourceID(), 2)
}
func TestTableMetadataV2MissingPartitionSpecs(t *testing.T) {
diff --git a/table/partition_conflict_bench_test.go
b/table/partition_conflict_bench_test.go
index 9e5cc06c8..9ec32ff17 100644
--- a/table/partition_conflict_bench_test.go
+++ b/table/partition_conflict_bench_test.go
@@ -201,9 +201,9 @@ func eqDeletePartitionsToFilterBeforeDedup(
return nil, fmt.Errorf("partition field ID %d
not found in spec %d", partFieldID, f.SpecID())
}
- sourceField, ok :=
meta.CurrentSchema().FindFieldByID(pf.SourceID())
+ sourceField, ok :=
meta.CurrentSchema().FindFieldByID(pf.FirstSourceID())
if !ok {
- return nil, fmt.Errorf("source field ID %d
(partition field %q) not found in schema", pf.SourceID(), pf.Name)
+ return nil, fmt.Errorf("source field ID %d
(partition field %q) not found in schema", pf.FirstSourceID(), pf.Name)
}
value := p[partFieldID]
diff --git a/table/partitioned_fanout_writer.go
b/table/partitioned_fanout_writer.go
index 6583758a0..686786213 100644
--- a/table/partitioned_fanout_writer.go
+++ b/table/partitioned_fanout_writer.go
@@ -377,7 +377,7 @@ func newPartitionExtractionPlan(spec iceberg.PartitionSpec,
schema *iceberg.Sche
pathPlan.fields[i] = pathField
pathPlan.estimatedSize += len(pathField.escapedName) + 20
partitionFieldsInfo[i].fieldID = partitionField.ID
- colName, ok := schema.FindColumnName(sourceField.SourceID())
+ colName, ok :=
schema.FindColumnName(sourceField.FirstSourceID())
if !ok {
continue
}
@@ -385,9 +385,9 @@ func newPartitionExtractionPlan(spec iceberg.PartitionSpec,
schema *iceberg.Sche
if !ok {
return nil, fmt.Errorf("failed to find source column %q
in record schema", colName)
}
- sourceType, ok := schema.FindTypeByID(sourceField.SourceID())
+ sourceType, ok :=
schema.FindTypeByID(sourceField.FirstSourceID())
if !ok {
- return nil, fmt.Errorf("failed to find type for source
field ID %d in schema", sourceField.SourceID())
+ return nil, fmt.Errorf("failed to find type for source
field ID %d in schema", sourceField.FirstSourceID())
}
partitionFieldsInfo[i] = partitionFieldInfo{
sourceField: sourceField,
@@ -554,7 +554,7 @@ func (p *partitionExtractionPlan)
getRecordPartitions(record arrow.RecordBatch)
chain, err := resolveColumnChain(record,
fieldInfo.columnPath)
if err != nil {
return nil, fmt.Errorf("failed to resolve
source column %q (field ID %d): %w",
- fieldInfo.sourceName,
fieldInfo.sourceField.SourceID(), err)
+ fieldInfo.sourceName,
fieldInfo.sourceField.FirstSourceID(), err)
}
columnChains[i] = chain
}
@@ -589,7 +589,7 @@ func (p *partitionExtractionPlan)
getRecordPartitions(record arrow.RecordBatch)
return nil, fmt.Errorf(
"failed to convert source
column %q (field ID %d) from Arrow type %s to Iceberg type %s: %w",
fieldInfo.sourceName,
-
fieldInfo.sourceField.SourceID(),
+
fieldInfo.sourceField.FirstSourceID(),
col.DataType(),
fieldInfo.sourceType,
err,
diff --git a/table/sort_keys.go b/table/sort_keys.go
index 0088a93dc..16ee150f1 100644
--- a/table/sort_keys.go
+++ b/table/sort_keys.go
@@ -50,10 +50,10 @@ func resolveSortKeys(order SortOrder, fileSchema
*iceberg.Schema) ([]compute.Sor
fields := fileSchema.Fields()
keys := make([]compute.SortKey, 0, order.Len())
for _, field := range order.fields {
- idx, ok := topLevelFieldIndex(fields, field.SourceID())
+ idx, ok := topLevelFieldIndex(fields, field.FirstSourceID())
if !ok {
return nil, fmt.Errorf("sort order %d: source id %d is
not a top-level column in schema",
- order.OrderID(), field.SourceID())
+ order.OrderID(), field.FirstSourceID())
}
key := compute.SortKey{ColumnIndex: idx, Order:
compute.SortOrderAscending, NullPlacement: compute.SortNullsAtEnd}
diff --git a/table/sort_keys_bench_test.go b/table/sort_keys_bench_test.go
index 01d231048..785b5ae86 100644
--- a/table/sort_keys_bench_test.go
+++ b/table/sort_keys_bench_test.go
@@ -144,10 +144,10 @@ func resolveSortKeysBefore(order SortOrder, fileSchema
*iceberg.Schema) ([]compu
keys := make([]compute.SortKey, 0, order.Len())
for _, field := range order.fields {
- idx, ok := topLevelFieldIndexBefore(fileSchema,
field.SourceID())
+ idx, ok := topLevelFieldIndexBefore(fileSchema,
field.FirstSourceID())
if !ok {
return nil, fmt.Errorf("sort order %d: source id %d is
not a top-level column in schema",
- order.OrderID(), field.SourceID())
+ order.OrderID(), field.FirstSourceID())
}
key := compute.SortKey{ColumnIndex: idx, Order:
compute.SortOrderAscending, NullPlacement: compute.SortNullsAtEnd}
diff --git a/table/sorting.go b/table/sorting.go
index 80c288740..83e17b12f 100644
--- a/table/sorting.go
+++ b/table/sorting.go
@@ -76,8 +76,9 @@ type SortField struct {
NullOrder NullOrder `json:"null-order"`
}
-// SourceID returns the first source column id.
-func (s SortField) SourceID() int {
+// FirstSourceID returns the first source column id, or 0 when SourceIDs is
+// empty. For multi-argument transforms the other ids are only in SourceIDs.
+func (s SortField) FirstSourceID() int {
if len(s.SourceIDs) == 0 {
return 0
}
@@ -98,14 +99,14 @@ func (s *SortField) String() string {
return fmt.Sprintf("%v %s %s", s.SourceIDs,
s.Direction, s.NullOrder)
}
- return fmt.Sprintf("%d %s %s", s.SourceID(), s.Direction,
s.NullOrder)
+ return fmt.Sprintf("%d %s %s", s.FirstSourceID(), s.Direction,
s.NullOrder)
}
if len(s.SourceIDs) > 1 {
return fmt.Sprintf("%s(%v) %s %s", s.Transform, s.SourceIDs,
s.Direction, s.NullOrder)
}
- return fmt.Sprintf("%s(%d) %s %s", s.Transform, s.SourceID(),
s.Direction, s.NullOrder)
+ return fmt.Sprintf("%s(%d) %s %s", s.Transform, s.FirstSourceID(),
s.Direction, s.NullOrder)
}
func (s SortField) MarshalJSON() ([]byte, error) {
@@ -137,7 +138,7 @@ func (s SortField) MarshalJSON() ([]byte, error) {
Transform iceberg.Transform `json:"transform"`
Direction SortDirection `json:"direction"`
NullOrder NullOrder `json:"null-order"`
- }{s.SourceID(), s.Transform, direction, nullOrder})
+ }{s.FirstSourceID(), s.Transform, direction, nullOrder})
}
func (s *SortField) UnmarshalJSON(b []byte) error {
@@ -482,7 +483,7 @@ func (s *SortOrder) CheckCompatibility(schema
*iceberg.Schema) error {
for _, field := range s.fields {
if field.Transform == nil {
- return fmt.Errorf("%w: sort field with source id %d has
no transform", ErrInvalidTransform, field.SourceID())
+ return fmt.Errorf("%w: sort field with source id %d has
no transform", ErrInvalidTransform, field.FirstSourceID())
}
if err := validateSortSourceIDs(field.SourceIDs, boundOrder);
err != nil {
@@ -552,7 +553,7 @@ func AssignFreshSortOrderIDsWithID(sortOrder SortOrder,
old, fresh *iceberg.Sche
fields := make([]SortField, 0, len(sortOrder.fields))
for _, field := range sortOrder.fields {
- originalField, ok := old.FindColumnName(field.SourceID())
+ originalField, ok := old.FindColumnName(field.FirstSourceID())
if !ok {
return SortOrder{}, fmt.Errorf("cannot find source
column id %s in old schema", field.String())
}
diff --git a/table/sorting_test.go b/table/sorting_test.go
index 06b14ac6f..e8ff271f3 100644
--- a/table/sorting_test.go
+++ b/table/sorting_test.go
@@ -231,7 +231,7 @@ func TestUnboundSortOrderUnmarshalAcceptsZeroSourceID(t
*testing.T) {
require.NoError(t, json.Unmarshal([]byte(`{"order-id": 1, "fields":
[{"source-id": 0, "transform": "identity", "direction": "asc", "null-order":
"nulls-first"}]}`), &sortOrder))
require.Equal(t, 1, sortOrder.Len())
for _, field := range sortOrder.Fields() {
- assert.Equal(t, 0, field.SourceID())
+ assert.Equal(t, 0, field.FirstSourceID())
}
roundTripped, err := json.Marshal(sortOrder)
@@ -631,7 +631,7 @@ func TestSortFieldMultiArgSourceIDs(t *testing.T) {
var field table.SortField
err := json.Unmarshal([]byte(jsonData), &field)
require.NoError(t, err)
- assert.Equal(t, 2, field.SourceID())
+ assert.Equal(t, 2, field.FirstSourceID())
assert.Equal(t, []int{2, 3}, field.SourceIDs)
})
@@ -700,7 +700,7 @@ func TestSortFieldMultiArgSourceIDs(t *testing.T) {
var decoded table.SortField
err = json.Unmarshal(data, &decoded)
require.NoError(t, err)
- assert.Equal(t, 2, decoded.SourceID())
+ assert.Equal(t, 2, decoded.FirstSourceID())
assert.Equal(t, []int{2, 3}, decoded.SourceIDs)
})
@@ -747,3 +747,20 @@ func
TestSortFieldMarshalAppliesDefaultsWithoutMutatingReceiver(t *testing.T) {
assert.Empty(t, field.Direction)
assert.Empty(t, field.NullOrder)
}
+
+func TestSortFieldFirstSourceID(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ sourceIDs []int
+ want int
+ }{
+ {name: "empty", sourceIDs: nil, want: 0},
+ {name: "single", sourceIDs: []int{7}, want: 7},
+ {name: "multiple", sourceIDs: []int{7, 9}, want: 7},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ field := table.SortField{SourceIDs: tc.sourceIDs}
+ assert.Equal(t, tc.want, field.FirstSourceID())
+ })
+ }
+}
diff --git a/table/update_spec.go b/table/update_spec.go
index dae04d0e1..678a77f92 100644
--- a/table/update_spec.go
+++ b/table/update_spec.go
@@ -116,7 +116,7 @@ func NewUpdateSpec(t *Transaction, caseSensitive bool)
*UpdateSpec {
return us
}
transformToField[transformKey{
- SourceId: partitionField.SourceID(),
+ SourceId: partitionField.FirstSourceID(),
Transform: partitionField.Transform.String(),
}] = partitionField
nameToField[partitionField.Name] = partitionField
@@ -215,9 +215,9 @@ func (us *UpdateSpec) Apply() (iceberg.PartitionSpec,
error) {
var err error
if _, deleted := us.deletes[field.FieldID]; !deleted {
if rename, renamed := us.renames[field.Name]; renamed {
- newField, err = us.addNewField(us.schema,
field.SourceID(), field.FieldID, rename, field.Transform, partitionNames)
+ newField, err = us.addNewField(us.schema,
field.FirstSourceID(), field.FieldID, rename, field.Transform, partitionNames)
} else {
- newField, err = us.addNewField(us.schema,
field.SourceID(), field.FieldID, field.Name, field.Transform, partitionNames)
+ newField, err = us.addNewField(us.schema,
field.FirstSourceID(), field.FieldID, field.Name, field.Transform,
partitionNames)
}
if err != nil {
return iceberg.PartitionSpec{}, err
@@ -225,9 +225,9 @@ func (us *UpdateSpec) Apply() (iceberg.PartitionSpec,
error) {
partitionFields = append(partitionFields, newField)
} else if us.meta.Version() == 1 {
if rename, renamed := us.renames[field.Name]; renamed {
- newField, err = us.addNewField(us.schema,
field.SourceID(), field.FieldID, rename, iceberg.VoidTransform{},
partitionNames)
+ newField, err = us.addNewField(us.schema,
field.FirstSourceID(), field.FieldID, rename, iceberg.VoidTransform{},
partitionNames)
} else {
- newField, err = us.addNewField(us.schema,
field.SourceID(), field.FieldID, field.Name, iceberg.VoidTransform{},
partitionNames)
+ newField, err = us.addNewField(us.schema,
field.FirstSourceID(), field.FieldID, field.Name, iceberg.VoidTransform{},
partitionNames)
}
if err != nil {
return iceberg.PartitionSpec{}, err
@@ -275,7 +275,7 @@ func validateNoRedundantTimeFields(fields
[]iceberg.PartitionField) error {
if _, isTime := field.Transform.(iceberg.TimeTransform);
!isTime {
continue
}
- source := field.SourceID()
+ source := field.FirstSourceID()
if existing, exists := timeFields[source]; exists {
return fmt.Errorf("%w: redundant partition field for
source ID %d: %s (%s) conflicts with %s (%s); remove one of them in this
update",
iceberg.ErrInvalidPartitionSpec, source,
@@ -419,7 +419,7 @@ func (us *UpdateSpec) historicalPartitionFields(key
transformKey) []iceberg.Part
for _, spec := range us.meta.PartitionSpecs() {
for _, field := range spec.Fields() {
historicalKey := transformKey{
- SourceId: field.SourceID(),
+ SourceId: field.FirstSourceID(),
Transform: field.Transform.String(),
}
us.historicalFields[historicalKey] =
append(us.historicalFields[historicalKey], field)
diff --git a/table/update_spec_test.go b/table/update_spec_test.go
index 92727b0bf..65bae0be4 100644
--- a/table/update_spec_test.go
+++ b/table/update_spec_test.go
@@ -118,7 +118,7 @@ func TestUpdateSpecPreservesSourceLessVoidTombstone(t
*testing.T) {
assert.Equal(t, 1000, newSpec.Field(0).FieldID)
assert.Equal(t, "id", newSpec.Field(0).Name)
assert.Equal(t, iceberg.VoidTransform{}, newSpec.Field(0).Transform)
- assert.Equal(t, 2, newSpec.Field(1).SourceID())
+ assert.Equal(t, 2, newSpec.Field(1).FirstSourceID())
assert.Equal(t, "name_identity", newSpec.Field(1).Name)
}
@@ -169,7 +169,7 @@ func
TestUpdateSpecAllowsReplacingTimeTransformOnSameSource(t *testing.T) {
require.Equal(t, 1, newSpec.NumFields())
assert.Equal(t, "ts_month", newSpec.Field(0).Name)
assert.Equal(t, iceberg.MonthTransform{},
newSpec.Field(0).Transform)
- assert.Equal(t, 3, newSpec.Field(0).SourceID())
+ assert.Equal(t, 3, newSpec.Field(0).FirstSourceID())
}
t.Run("remove then add", func(t *testing.T) {
@@ -211,10 +211,10 @@ func
TestUpdateSpecReplacesTimeTransformOverV1VoidTombstone(t *testing.T) {
require.Equal(t, 2, newSpec.NumFields())
assert.Equal(t, "ts_year", newSpec.Field(0).Name)
assert.Equal(t, iceberg.VoidTransform{}, newSpec.Field(0).Transform)
- assert.Equal(t, 3, newSpec.Field(0).SourceID())
+ assert.Equal(t, 3, newSpec.Field(0).FirstSourceID())
assert.Equal(t, "ts_month", newSpec.Field(1).Name)
assert.Equal(t, iceberg.MonthTransform{}, newSpec.Field(1).Transform)
- assert.Equal(t, 3, newSpec.Field(1).SourceID())
+ assert.Equal(t, 3, newSpec.Field(1).FirstSourceID())
}
// Iceberg-Java writes this metadata today, so loading it has to keep working:
the
@@ -328,13 +328,13 @@ func TestUpdateSpecAddField(t *testing.T) {
assert.Equal(t, "street_void",
newSpec.FieldsBySourceID(5)[0].Name)
addedField := newSpec.FieldsBySourceID(3)[0]
- assert.Equal(t, 3, addedField.SourceID())
+ assert.Equal(t, 3, addedField.FirstSourceID())
assert.Equal(t, 1002, addedField.FieldID)
assert.Equal(t, "year_transform", addedField.Name)
assert.Equal(t, iceberg.YearTransform{}, addedField.Transform)
addedField = newSpec.FieldsBySourceID(7)[0]
- assert.Equal(t, 7, addedField.SourceID())
+ assert.Equal(t, 7, addedField.FirstSourceID())
assert.Equal(t, 1003, addedField.FieldID)
assert.Equal(t, "zipcode_bucket", addedField.Name)
assert.Equal(t, iceberg.BucketTransform{NumBuckets: 5},
addedField.Transform)
@@ -643,13 +643,13 @@ func TestUpdateSpecAddIdentityField(t *testing.T) {
assert.Equal(t, "street_void",
newSpec.FieldsBySourceID(5)[0].Name)
addedField := newSpec.FieldsBySourceID(3)[0]
- assert.Equal(t, 3, addedField.SourceID())
+ assert.Equal(t, 3, addedField.FirstSourceID())
assert.Equal(t, 1002, addedField.FieldID)
assert.Equal(t, "ts", addedField.Name)
assert.Equal(t, iceberg.IdentityTransform{},
addedField.Transform)
addedField = newSpec.FieldsBySourceID(2)[0]
- assert.Equal(t, 2, addedField.SourceID())
+ assert.Equal(t, 2, addedField.FirstSourceID())
assert.Equal(t, 1003, addedField.FieldID)
assert.Equal(t, "name", addedField.Name)
assert.Equal(t, iceberg.IdentityTransform{},
addedField.Transform)
@@ -910,13 +910,13 @@ func TestUpdateSpecCommit(t *testing.T) {
assert.Equal(t, []iceberg.PartitionField(nil),
currSpec.FieldsBySourceID(1))
addedField := currSpec.FieldsBySourceID(6)[0]
- assert.Equal(t, 6, addedField.SourceID())
+ assert.Equal(t, 6, addedField.FirstSourceID())
assert.Equal(t, 1002, addedField.FieldID)
assert.Equal(t, "address.city_trunc_3", addedField.Name)
assert.Equal(t, iceberg.TruncateTransform{Width: 3},
addedField.Transform)
addedIdentity := currSpec.FieldsBySourceID(7)[0]
- assert.Equal(t, 7, addedIdentity.SourceID())
+ assert.Equal(t, 7, addedIdentity.FirstSourceID())
assert.Equal(t, 1003, addedIdentity.FieldID)
assert.Equal(t, "address.zip_code", addedIdentity.Name)
assert.Equal(t, iceberg.IdentityTransform{},
addedIdentity.Transform)
diff --git a/website/src/releases.md b/website/src/releases.md
index ba5859370..61ae67689 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -61,6 +61,8 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
Relatedly, the previously unexported `isGCEnabled` is now
`table.IsGCEnabled`, so the new package can gate destructive operations on the
`gc.enabled` table property.
- `maintenance.PurgeFiles` (previously `Table.PurgeFiles`) now calls
`IO.Remove` concurrently, with up to 32 calls in flight, when the filesystem
does not implement `BulkRemovableIO`. Custom IO implementations, including
those registered with `io.Register`, must make `Remove` safe for concurrent use
and synchronize any shared mutable state. The bulk deletion path is unchanged.
+- **Partition and sort source accessors renamed.** Replace
`PartitionField.SourceID()` and `SortField.SourceID()` with `FirstSourceID()`.
Their behavior and serialized fields are unchanged.
+
## Verifying and producing releases
If you are validating an RC or cutting a new release, see:
