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

Merged
laskoviymishka merged 5 commits into from
Sep 11, 2026
+130 −16
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 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
96d47a6
fix(table): repair partition counter write-back
mattfaltyn Sep 11, 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
28 changes: 19 additions & 9 deletions table/metadata.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,9 +252,8 @@ 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,
// which may be stale relative to partition spec history.
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 +699,29 @@ func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) error {
return nil
}

// partitionFieldIDFloor returns an allocation floor that accounts for the
// persisted counter and all partition field IDs in spec history.
func partitionFieldIDFloor(lastPartitionID *int, specs []iceberg.PartitionSpec) int {
floor := iceberg.PartitionDataIDStart - 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)

@laskoviymishka laskoviymishka Sep 11, 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.

The floor made it into BindToSchema, but the b.lastPartitionID write-back a few lines down still uses max(maxFieldID, prev) with prev = *b.lastPartitionID. For a non-empty spec that's fine, maxFieldID heals it. For an empty spec (unpartition) maxFieldID is 0 and prev is the stale 999, so the counter we serialize stays stale, and a reader that trusts it without scanning specs can still allocate 1000 and collide, the exact case this PR is closing on the NewUpdateSpec path.

fieldIDFloor is already in scope and always >= prev, so:

lastPartitionID := max(maxFieldID, fieldIDFloor)

The prev block and its nil-guard fall out as dead code, and the assertion value is unaffected since that's drawn from the base Metadata. wdyt?

@mattfaltyn mattfaltyn Sep 11, 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.

“fieldIDFloor is already in scope and always >= prev.”

Updated. Write-back now uses max(maxFieldID, fieldIDFloor), with an empty-spec regression test. Thanks!

if err != nil {
return err
}
Expand Down Expand Up @@ -736,11 +750,7 @@ func (b *MetadataBuilder) AddPartitionSpec(spec *iceberg.PartitionSpec, initial

}

prev := partitionFieldStartID - 1
if b.lastPartitionID != nil {
prev = *b.lastPartitionID
}
lastPartitionID := max(maxFieldID, prev)
lastPartitionID := max(maxFieldID, fieldIDFloor)

var specs []iceberg.PartitionSpec
if initial {
Expand Down
60 changes: 60 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,66 @@ 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)

for _, tt := range []struct {
name string
dropCounter bool
}{
{name: "stale counter"},
{name: "nil counter", dropCounter: true},
} {
t.Run(tt.name, func(t *testing.T) {
builder, err := MetadataBuilderFromBase(metadata, "")
require.NoError(t, err)
if tt.dropCounter {
builder.lastPartitionID = nil
}

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 TestAddEmptyPartitionSpecRepairsStaleLastPartitionID(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)

emptySpec := iceberg.NewPartitionSpecID(1)
require.NoError(t, builder.AddPartitionSpec(&emptySpec, false))

rebuilt, err := builder.Build()
require.NoError(t, err)
require.NotNil(t, rebuilt.LastPartitionSpecID())
assert.Equal(t, 1000, *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