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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
