You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "badalprasadsingh (via GitHub)" <gi...@apache.org> on 2026/08/05 05:48:48 UTC
[PR] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
badalprasadsingh opened a new pull request, #1651:
URL: https://github.com/apache/iceberg-go/pull/1651
## Description
Fixes #1642
Now, `UpdateSpec` reads table schema and metadata from the transaction's staged metadata snapshot (captured at construction) - similar to what we have in `PyIceberg` and `Iceberg-Java`.
This unblocks fundamental workflows such as adding a column and partitioning by it within a single transaction.
## Testing
Added the necessary tests for it.
--
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] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #1651:
URL: https://github.com/apache/iceberg-go/pull/1651#discussion_r3739094608
##########
table/update_spec.go:
##########
@@ -145,14 +154,14 @@ func (us *UpdateSpec) BuildUpdates() ([]Update, []Requirement, error) {
updates := make([]Update, 0)
requirements := make([]Requirement, 0)
- if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.meta.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()
+ requiredLastAssignedPartitionId := us.meta.LastPartitionSpecID()
requirements = append(requirements, AssertLastAssignedPartitionID(*requiredLastAssignedPartitionId))
Review Comment:
Separate from the baseline question: `LastPartitionSpecID()` returns `*int` and we dereference it unconditionally on the next line. The constructor guards exactly this case a few lines up (`if lastAssignedFieldId == nil` falling back to `PartitionDataIDStart - 1`), and the `MetadataBuilder` path can hand us a nil `last-partition-id`, so this is a latent panic. I'd mirror the constructor's guard here before the deref.
##########
table/update_spec_test.go:
##########
@@ -280,6 +280,110 @@ func TestUpdateSpecAddField(t *testing.T) {
})
}
+func TestUpdateSpecReadsStagedTransactionMetadata(t *testing.T) {
+ t.Run("end-to-end: partition by column added earlier in the same transaction", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ require.NoError(t, txn.UpdateSchema(false, false).
+ AddColumn([]string{"new_col"}, iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ require.NoError(t, txn.UpdateSpec(false).
+ AddField("new_col", iceberg.IdentityTransform{}, "new_col_identity").
+ Commit())
+
+ stagedTbl, err := txn.StagedTable()
+ require.NoError(t, err)
+
+ // The new column is assigned schema field id 8 (the existing schema
+ // occupies ids 1-7), so the partition field must reference source id 8.
+ spec := stagedTbl.Spec()
+ added := spec.FieldsBySourceID(8)
+ require.Len(t, added, 1)
+ assert.Equal(t, "new_col_identity", added[0].Name)
+ assert.Equal(t, iceberg.IdentityTransform{}, added[0].Transform)
+ assert.Equal(t, iceberg.PartitionDataIDStart, added[0].FieldID)
+ })
+
+ t.Run("auto-generated partition name resolves against the staged schema", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ require.NoError(t, txn.UpdateSchema(false, false).
+ AddColumn([]string{"new_col"}, iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ // An empty target name forces GeneratePartitionFieldName, which must
+ // resolve the source column against the staged schema.
+ specUpdate := txn.UpdateSpec(false)
+ _, _, err := specUpdate.
+ AddField("new_col", iceberg.IdentityTransform{}, "").
+ BuildUpdates()
+ require.NoError(t, err)
+
+ newSpec, err := specUpdate.Apply()
+ require.NoError(t, err)
+ added := newSpec.FieldsBySourceID(8)
+ require.Len(t, added, 1)
+ assert.Equal(t, "new_col", added[0].Name)
+ })
+
+ t.Run("end-to-end: chained UpdateSpec sees partition fields staged earlier", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ // Two independent UpdateSpec commits in the same transaction. The
+ // second must observe the field staged by the first.
+ require.NoError(t, txn.UpdateSpec(false).AddIdentity("id").Commit())
Review Comment:
This subtest stages two specs and asserts against `StagedTable()`, but it never calls `txn.Commit(ctx)`, so the contradictory-requirements bug is invisible; the test passes whether or not the requirement baseline is correct.
Same ask as zeroshade's inline: I'd add a real catalog commit (in-memory catalog is fine) after the two `UpdateSpec.Commit()` calls, assert it succeeds, and assert exactly one `assert-last-assigned-partition-id` survives with the original value. That's the assertion that actually guards this fix.
##########
table/update_spec.go:
##########
@@ -35,6 +35,7 @@ type UpdateSpec struct {
operations []updateSpecOp
txn *Transaction
+ meta Metadata
Review Comment:
One design thing on the new `meta` field: it's snapshotted once in `NewUpdateSpec` via `meta.Build()` and then frozen, but `BuildUpdates()`/`Apply()` run lazily later. If anything else stages a change on the transaction between constructing this `UpdateSpec` and committing it, the snapshot is stale, and there's no doc on the field saying it intentionally freezes at construction.
I'd consider building the snapshot at the start of `BuildUpdates()` instead, so it reflects transaction state at apply time the way `transaction.apply()` does. If we keep it at construction, I'd at least document that on the field. wdyt?
##########
table/update_spec.go:
##########
@@ -196,15 +205,15 @@ func (us *UpdateSpec) Apply() (iceberg.PartitionSpec, error) {
partitionFields = append(partitionFields, us.adds...)
opts := make([]iceberg.PartitionOption, len(partitionFields))
for i, field := range partitionFields {
- opts[i] = iceberg.AddPartitionFieldBySourceID(field.SourceID(), field.Name, field.Transform, us.txn.tbl.Schema(), &field.FieldID)
+ opts[i] = iceberg.AddPartitionFieldBySourceID(field.SourceID(), field.Name, field.Transform, us.meta.CurrentSchema(), &field.FieldID)
}
newSpec, err := iceberg.NewPartitionSpecOpts(opts...)
if err != nil {
return iceberg.PartitionSpec{}, err
}
newSpecId := iceberg.InitialPartitionSpecID
- for _, spec = range us.txn.tbl.Metadata().PartitionSpecs() {
+ for _, spec = range us.meta.PartitionSpecs() {
Review Comment:
This dedup loop and `isNewPartitionSpec()` (line 454) now read partition specs from the staged `us.meta`. After the first chained `Commit()`, a spec that's genuinely new relative to the committed catalog can already appear in `us.meta`, so `isNewPartitionSpec` returns false for a net-new spec and we'd suppress the `AddPartitionSpecUpdate` that has to reach the catalog, or select the wrong `newSpecId`.
Field resolution should keep using `us.meta`, but these existing-spec lookups want the committed snapshot. wdyt?
##########
table/update_spec.go:
##########
@@ -73,26 +74,34 @@ func NewUpdateSpec(t *Transaction, caseSensitive bool) *UpdateSpec {
return us
}
- // UpdateSpec reads exclusively from the committed table metadata
- // (t.tbl.Metadata() / t.tbl.Schema()) rather than the transaction's
- // metadata builder, but still routes its initialization check through the
- // canonical txnMeta accessor for consistency.
- _, us.err = t.txnMeta()
- if us.err != nil {
+ // Read table state from the transaction's staged metadata builder rather
+ // than the frozen table snapshot captured when the transaction began.
+ // So, columns and specs added earlier in the same transaction can now be observed.
Review Comment:
Small thing: the `So,` opener reads a little awkwardly for a doc comment. The earlier single-sentence phrasing was cleaner, maybe `...captured when the transaction began, so that columns and specs added earlier in the same transaction are visible immediately.`
##########
table/update_spec.go:
##########
@@ -145,14 +154,14 @@ func (us *UpdateSpec) BuildUpdates() ([]Update, []Requirement, error) {
updates := make([]Update, 0)
requirements := make([]Requirement, 0)
- if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.meta.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()
+ requiredLastAssignedPartitionId := us.meta.LastPartitionSpecID()
Review Comment:
This is the root of the conflict zeroshade flagged. Sourcing the requirement from `us.meta` means a second chained `UpdateSpec.Commit()` asserts the staged `last-partition-id` that the first call already advanced, so we emit two `AssertLastAssignedPartitionID` requirements with different values and both survive dedup and reach the catalog.
The concurrency baseline has to be the value the catalog actually holds, so I'd read this one from `us.txn.tbl.Metadata().LastPartitionSpecID()` (the pre-transaction snapshot) even though everything else in here correctly moved to `us.meta`. Staged state is right for field resolution, wrong for the assertion.
##########
table/update_spec_test.go:
##########
@@ -280,6 +280,110 @@ func TestUpdateSpecAddField(t *testing.T) {
})
}
+func TestUpdateSpecReadsStagedTransactionMetadata(t *testing.T) {
+ t.Run("end-to-end: partition by column added earlier in the same transaction", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ require.NoError(t, txn.UpdateSchema(false, false).
+ AddColumn([]string{"new_col"}, iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ require.NoError(t, txn.UpdateSpec(false).
+ AddField("new_col", iceberg.IdentityTransform{}, "new_col_identity").
+ Commit())
+
+ stagedTbl, err := txn.StagedTable()
+ require.NoError(t, err)
+
+ // The new column is assigned schema field id 8 (the existing schema
+ // occupies ids 1-7), so the partition field must reference source id 8.
+ spec := stagedTbl.Spec()
+ added := spec.FieldsBySourceID(8)
Review Comment:
The literal `8` here (and at line 325) leans on `testNonPartitionedTable`'s schema occupying ids 1-7, so if that schema ever changes this silently returns an empty slice and fails on the `Len` with a confusing message. I'd resolve the id from the staged schema instead, e.g. `f, _ := stagedTbl.Schema().FindFieldByName("new_col"); spec.FieldsBySourceID(f.ID)`.
--
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] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1651:
URL: https://github.com/apache/iceberg-go/pull/1651#discussion_r3732110893
##########
table/update_spec.go:
##########
@@ -145,14 +154,14 @@ func (us *UpdateSpec) BuildUpdates() ([]Update, []Requirement, error) {
updates := make([]Update, 0)
requirements := make([]Requirement, 0)
- if us.txn.tbl.Metadata().DefaultPartitionSpec() != newSpec.ID() {
+ if us.meta.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()
+ requiredLastAssignedPartitionId := us.meta.LastPartitionSpecID()
Review Comment:
On the second staged spec update, `us.meta` contains the first update's last partition-field ID, while the first update already staged a requirement for the original value. Both requirements survive deduplication and are validated against the same catalog snapshot, so they contradict each other and cannot be repaired by retry. **Suggested fix:** derive this requirement from `us.txn.tbl.Metadata()` so every staged update asserts the same original/base value, or make this a singleton requirement that retains the first value.
##########
table/update_spec_test.go:
##########
@@ -280,6 +280,110 @@ func TestUpdateSpecAddField(t *testing.T) {
})
}
+func TestUpdateSpecReadsStagedTransactionMetadata(t *testing.T) {
+ t.Run("end-to-end: partition by column added earlier in the same transaction", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ require.NoError(t, txn.UpdateSchema(false, false).
+ AddColumn([]string{"new_col"}, iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ require.NoError(t, txn.UpdateSpec(false).
+ AddField("new_col", iceberg.IdentityTransform{}, "new_col_identity").
+ Commit())
+
+ stagedTbl, err := txn.StagedTable()
+ require.NoError(t, err)
+
+ // The new column is assigned schema field id 8 (the existing schema
+ // occupies ids 1-7), so the partition field must reference source id 8.
+ spec := stagedTbl.Spec()
+ added := spec.FieldsBySourceID(8)
+ require.Len(t, added, 1)
+ assert.Equal(t, "new_col_identity", added[0].Name)
+ assert.Equal(t, iceberg.IdentityTransform{}, added[0].Transform)
+ assert.Equal(t, iceberg.PartitionDataIDStart, added[0].FieldID)
+ })
+
+ t.Run("auto-generated partition name resolves against the staged schema", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ require.NoError(t, txn.UpdateSchema(false, false).
+ AddColumn([]string{"new_col"}, iceberg.PrimitiveTypes.String, "", false, nil).
+ Commit())
+
+ // An empty target name forces GeneratePartitionFieldName, which must
+ // resolve the source column against the staged schema.
+ specUpdate := txn.UpdateSpec(false)
+ _, _, err := specUpdate.
+ AddField("new_col", iceberg.IdentityTransform{}, "").
+ BuildUpdates()
+ require.NoError(t, err)
+
+ newSpec, err := specUpdate.Apply()
+ require.NoError(t, err)
+ added := newSpec.FieldsBySourceID(8)
+ require.Len(t, added, 1)
+ assert.Equal(t, "new_col", added[0].Name)
+ })
+
+ t.Run("end-to-end: chained UpdateSpec sees partition fields staged earlier", func(t *testing.T) {
+ txn := testNonPartitionedTable.NewTransaction()
+
+ // Two independent UpdateSpec commits in the same transaction. The
+ // second must observe the field staged by the first.
+ require.NoError(t, txn.UpdateSpec(false).AddIdentity("id").Commit())
+ require.NoError(t, txn.UpdateSpec(false).AddIdentity("name").Commit())
+
+ stagedTbl, err := txn.StagedTable()
Review Comment:
This stops at `StagedTable()`, before the contradictory requirements are sent to the catalog. **Suggested fix:** follow the two staged updates with `txn.Commit(ctx)`, assert that both fields are committed with exactly one base `last-assigned-partition-id` requirement, and cover the retry path as well.
--
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] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #1651:
URL: https://github.com/apache/iceberg-go/pull/1651#issuecomment-5279503664
@badalprasadsingh ready for merge, once conflicts resolved
--
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] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #1651:
URL: https://github.com/apache/iceberg-go/pull/1651
--
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] fix(table): update spec reading stale metadata in chained transactions [iceberg-go]
Posted by "badalprasadsingh (via GitHub)" <gi...@apache.org>.
badalprasadsingh commented on PR #1651:
URL: https://github.com/apache/iceberg-go/pull/1651#issuecomment-5280590572
Done @laskoviymishka :)
--
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