Skip to content

fix(table): allocate partition IDs from history #1988

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mattfaltyn wants to merge 4 commits into
base: main
Choose a base branch
from
+95 −11
Open

fix(table): allocate partition IDs from history #1988

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
abc35da
fix(table): normalize stale last partition ID
mattfaltyn Sep 3, 2026
ebf0a16
fix(table): preserve unassigned partition counters
mattfaltyn Sep 3, 2026
e01dbb2
test(table): cover partition counter normalization
mattfaltyn Sep 6, 2026
c9bae41
fix(table): allocate partition IDs from history
mattfaltyn Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions table/metadata.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,9 +252,9 @@ type Metadata interface {
// DefaultPartitionSpec is the ID of the current spec that writerFactory should
// use by default.
DefaultPartitionSpec() int
// LastPartitionSpecID is the highest assigned partition field ID across
// all partition specs for the table. This is used to ensure partition
// fields are always assigned an unused ID when evolving specs.
// LastPartitionSpecID returns the persisted last assigned partition field ID.
// Allocation also scans partition spec history because metadata written by
// another client may contain a stale counter.
LastPartitionSpecID() *int
// Snapshots returns the list of valid snapshots. Valid snapshots are
// snapshots for which all data files exist in the file system. A data
Expand Down Expand Up @@ -700,14 +700,27 @@ func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) error {
return nil
}

func partitionFieldIDFloor(lastPartitionID *int, specs []iceberg.PartitionSpec) int {
floor := partitionFieldStartID - 1
if lastPartitionID != nil {
floor = max(floor, *lastPartitionID)
}
for _, spec := range specs {
floor = max(floor, spec.LastAssignedFieldID())
}

return floor
}

func (b *MetadataBuilder) AddPartitionSpec(spec *iceberg.PartitionSpec, initial bool) error {
newSpecID := b.reuseOrCreateNewPartitionSpecID(*spec)
curSchema := b.CurrentSchema()
if curSchema == nil {
return errors.New("can't add sort order with no current schema")
}

freshSpec, err := spec.BindToSchema(curSchema, b.lastPartitionID, &newSpecID)
fieldIDFloor := partitionFieldIDFloor(b.lastPartitionID, b.specs)
freshSpec, err := spec.BindToSchema(curSchema, &fieldIDFloor, &newSpecID)
if err != nil {
return err
}
Expand Down
27 changes: 27 additions & 0 deletions table/metadata_builder_internal_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,33 @@ func TestAddRemovePartitionSpec(t *testing.T) {
require.ErrorContains(t, err, "id 1")
}

func TestAddPartitionSpecAllocatesAfterHistoricalFieldID(t *testing.T) {
data := strings.Replace(ExampleTableMetadataV2,
`"last-partition-id": 1000`, `"last-partition-id": 999`, 1)
require.Contains(t, data, `"last-partition-id": 999`)

metadata, err := ParseMetadataBytes([]byte(data))
require.NoError(t, err)
builder, err := MetadataBuilderFromBase(metadata, "")
require.NoError(t, err)

addedSpec, err := iceberg.NewPartitionSpecOpts(
iceberg.WithSpecID(1),
iceberg.AddPartitionFieldBySourceID(1, "x_bucket", iceberg.BucketTransform{NumBuckets: 16}, builder.CurrentSchema(), nil),
)
require.NoError(t, err)
require.NoError(t, builder.AddPartitionSpec(&addedSpec, false))

rebuilt, err := builder.Build()
require.NoError(t, err)
added := rebuilt.PartitionSpecByID(1)
require.NotNil(t, added)
require.Equal(t, 1, added.NumFields())
assert.Equal(t, 1001, added.Field(0).FieldID)
require.NotNil(t, rebuilt.LastPartitionSpecID())
assert.Equal(t, 1001, *rebuilt.LastPartitionSpecID())
}

func TestRemovePartitionSpecsNoMatchDoesNotUpdate(t *testing.T) {
for _, count := range []int{1, 2, 8, 9, 16, 17, 32, 33, 64} {
t.Run(fmt.Sprintf("requested=%d", count), func(t *testing.T) {
Expand Down
50 changes: 50 additions & 0 deletions table/metadata_preflight_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"strings"
"testing"

"github.com/apache/iceberg-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
Expand Down Expand Up @@ -85,6 +86,55 @@ func TestParseMetadataBytesAssignsMissingPartitionFieldIDs(t *testing.T) {
}
}

func TestParseMetadataBytesPreservesStaleLastPartitionIDForCommit(t *testing.T) {
data := strings.Replace(ExampleTableMetadataV2,

@zeroshade zeroshade Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor — New early-return condition adds three branches, only one is tested

The condition at metadata.go:2047 introduces distinct branches: counter below max field ID (tested), counter above max field ID (must stay untouched), counter below the 999 floor with no field IDs, and stale counter combined with a missing field-id. Only the first has a test. I verified the untested ones behave as follows -- add cases for them so the condition is pinned: counter-above-max stays untouched, and stale-counter-plus-missing-field-id assigns correctly.

@mattfaltyn mattfaltyn Sep 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for identifying the missing branches. Added coverage in ebf0a16 for sub-999 counters with no assigned fields, counters above the greatest assigned field ID, and stale counters combined with a missing field ID. The first two cases also verify the unchanged-byte fast path.

@zeroshade zeroshade Sep 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit — End-to-end allocation half of the issue's suggested regression coverage is not asserted

Issue #1987 asks for coverage that asserts both that the parsed counter becomes 1000 and that the next distinct partition field receives 1001. TestParseMetadataBytesNormalizesStaleLastPartitionID asserts only the former. I verified the latter holds today, so this is purely about locking in the user-visible symptom (the cross-spec ID collision) rather than only its parse-level cause.

@laskoviymishka laskoviymishka Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This only exercises the fix if the Replace actually lands. If the fixture spacing ever changes and the replace silently no-ops, last-partition-id stays 1000 and both assertions pass without touching the fix. I'd add a require.Contains right after, asserting the "last-partition-id": 999 substring is present in data, so a missed replace fails loudly.

@mattfaltyn mattfaltyn Sep 9, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add a require.Contains

Good call—added explicit fixture guards so silent replacement failures go red.

`"last-partition-id": 1000`, `"last-partition-id": 999`, 1)
data = strings.Replace(data,
`"default-spec-id": 0,`, `"default-spec-id": 1,`, 1)
data = strings.Replace(data,
`"partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}],`,
`"partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}, {"spec-id": 1, "fields": []}],`, 1)
require.Contains(t, data, `"last-partition-id": 999`)
require.Contains(t, data, `"spec-id": 1`)

parsed, err := ParseMetadataBytes([]byte(data))
require.NoError(t, err)
require.NotNil(t, parsed.LastPartitionSpecID())
assert.Equal(t, 999, *parsed.LastPartitionSpecID())

update := NewUpdateSpec(New(nil, parsed, "", nil, nil).NewTransaction(), false).
AddField("x", iceberg.BucketTransform{NumBuckets: 16}, "x_bucket")
_, requirements, err := update.BuildUpdates()
require.NoError(t, err)
assert.Equal(t, []int{999}, lastAssignedPartitionAssertions(requirements))
updated, err := update.Apply()
require.NoError(t, err)
require.Equal(t, 1, updated.NumFields())
assert.Equal(t, 1001, updated.Field(0).FieldID)
}

func TestAssignMissingPartitionFieldIDsPreservesConsistentMetadata(t *testing.T) {
for _, tt := range []struct {
name string
input string
}{
{
name: "counter below assignment floor with no fields",
input: `{"last-updated-ms":0,"last-partition-id":0,"partition-specs":[{"spec-id":0,"fields":[]}]}`,
},
{
name: "counter above greatest field ID",
input: `{"last-updated-ms":0,"last-partition-id":1001,"partition-specs":[{"spec-id":0,"fields":[{"field-id":1000}]}]}`,
},
} {
t.Run(tt.name, func(t *testing.T) {
normalized, err := assignMissingPartitionFieldIDs([]byte(tt.input))
require.NoError(t, err)
assert.Equal(t, tt.input, string(normalized))
})
}
}

func TestParseMetadataBytesRejectsCaseFoldedFormatVersionCollision(t *testing.T) {
data := strings.Replace(
ExampleTableMetadataV2,
Expand Down
8 changes: 1 addition & 7 deletions table/update_spec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,15 +122,9 @@ func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
nameToField[partitionField.Name] = partitionField
}
us.schema = stagedMeta.CurrentSchema()
lastAssignedFieldId := us.meta.LastPartitionSpecID()
if lastAssignedFieldId == nil {
v := iceberg.PartitionDataIDStart - 1
lastAssignedFieldId = &v
}

us.nameToField = nameToField
us.transformToField = transformToField
us.lastAssignedFieldId = *lastAssignedFieldId
us.lastAssignedFieldId = partitionFieldIDFloor(us.meta.LastPartitionSpecID(), us.meta.PartitionSpecs())

return us
}
Expand Down
Loading