laskoviymishka commented on code in PR #1856:
URL: https://github.com/apache/iceberg-go/pull/1856#discussion_r3850560460
##########
table/row_delta_test.go:
##########
@@ -1427,3 +1427,164 @@ func assertRowCount(t *testing.T, tbl *table.Table,
expected int64) {
assert.Equal(t, expected, total, "unexpected row count")
}
+
+func unregisteredSpec() iceberg.PartitionSpec {
+ return iceberg.NewPartitionSpecID(99, iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id_identity",
Transform: iceberg.IdentityTransform{},
+ })
+}
+
+func buildPosDeleteFileForSpec(t *testing.T, spec iceberg.PartitionSpec, path
string, partition map[int]any) iceberg.DataFile {
+ t.Helper()
+
+ b, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes,
+ path, iceberg.ParquetFile, partition, nil, nil, 5, 512)
+ require.NoError(t, err)
+
+ return b.Build()
+}
+
+func buildDataFileForSpec(t *testing.T, spec iceberg.PartitionSpec, path
string, partition map[int]any) iceberg.DataFile {
+ t.Helper()
+
+ b, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentData,
+ path, iceberg.ParquetFile, partition, nil, nil, 10, 1024)
+ require.NoError(t, err)
+
+ return b.Build()
+}
+
+func TestRowDeltaRejectsDeleteFileWithUnregisteredSpec(t *testing.T) {
+ tbl := newRowDeltaCommitTestTable(t)
+
+ tx := tbl.NewTransaction()
+ rd := tx.NewRowDelta(nil)
+ rd.AddDeletes(buildPosDeleteFileForSpec(t, unregisteredSpec(),
+ "s3://bucket/data/pos-del.parquet", map[int]any{1000:
int64(7)}))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrPartitionSpecNotFound)
+ assert.ErrorContains(t, err, "99")
+
+ result, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Nil(t, result.CurrentSnapshot(), "a rejected row delta must not
produce a snapshot")
+}
+
+func TestRowDeltaRejectsDataFileWithUnregisteredSpec(t *testing.T) {
+ tbl := newRowDeltaCommitTestTable(t)
+
+ tx := tbl.NewTransaction()
+ rd := tx.NewRowDelta(nil)
+ rd.AddRows(buildDataFileForSpec(t, unregisteredSpec(),
+ "s3://bucket/data/insert.parquet", map[int]any{1000: int64(7)}))
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrPartitionSpecNotFound)
+ assert.ErrorContains(t, err, "99")
+}
+
+func TestRowDeltaWritesRegisteredNonDefaultSpec(t *testing.T) {
+ tbl := newRowDeltaCommitTestTable(t)
+
+ evolveTx := tbl.NewTransaction()
+ require.NoError(t, evolveTx.UpdateSpec(false).
+ AddField("id", iceberg.IdentityTransform{}, "id_identity").
+ Commit())
+ evolved, err := evolveTx.Commit(t.Context())
+ require.NoError(t, err)
+
+ spec := evolved.Metadata().PartitionSpec()
+ require.NotEqual(t, 0, spec.ID(), "spec evolution must register a new
spec id")
+ require.Equal(t, 1, spec.NumFields())
+ fieldID := spec.Field(0).FieldID
+ partition := map[int]any{fieldID: int64(7)}
+
+ tx := evolved.NewTransaction()
+ rd := tx.NewRowDelta(nil)
+ rd.AddRows(buildDataFileForSpec(t, spec,
"s3://bucket/data/insert.parquet", partition))
+ rd.AddDeletes(buildPosDeleteFileForSpec(t, spec,
"s3://bucket/data/pos-del.parquet", partition))
+ require.NoError(t, rd.Commit(t.Context()))
+
+ result, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ snap := result.CurrentSnapshot()
+ require.NotNil(t, snap)
+
+ fs := iceio.LocalFS{}
+ manifests, err := snap.Manifests(fs)
+ require.NoError(t, err)
+ require.Len(t, manifests, 2)
+
+ for _, m := range manifests {
+ assert.Equal(t, int32(spec.ID()), m.PartitionSpecID(),
+ "manifest must declare the spec its entries were
written under")
+ entries := 0
+ for e, err := range m.Entries(fs, true) {
+ require.NoError(t, err)
+ assert.Equal(t, int32(spec.ID()), e.DataFile().SpecID())
+ assert.Equal(t, int64(7),
e.DataFile().Partition()[fieldID])
+ entries++
+ }
+ assert.Equal(t, 1, entries)
+ }
+}
+
+func TestRowDeltaMixedSpecCommitLeavesNoManifests(t *testing.T) {
+ validData := func() iceberg.DataFile { return buildDataFile(t,
"s3://bucket/data/insert.parquet") }
+ validDelete := func() iceberg.DataFile { return buildPosDeleteFile(t,
"s3://bucket/data/pos-del.parquet") }
+ orphanData := func() iceberg.DataFile {
+ return buildDataFileForSpec(t, unregisteredSpec(),
"s3://bucket/data/orphan.parquet", map[int]any{1000: int64(7)})
+ }
+ orphanDelete := func() iceberg.DataFile {
+ return buildPosDeleteFileForSpec(t, unregisteredSpec(),
"s3://bucket/data/orphan-del.parquet", map[int]any{1000: int64(7)})
+ }
+
+ tests := []struct {
+ name string
+ rows []iceberg.DataFile
+ deletes []iceberg.DataFile
+ }{
+ {"valid data with unregistered delete",
[]iceberg.DataFile{validData()}, []iceberg.DataFile{orphanDelete()}},
+ {"valid delete with unregistered data",
[]iceberg.DataFile{orphanData()}, []iceberg.DataFile{validDelete()}},
+ {"valid data group before unregistered one",
[]iceberg.DataFile{validData(), orphanData()}, nil},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTable(t)
Review Comment:
This is the test that would catch the ordering gap I flagged in
`snapshot_producers.go`, but it can't as written — `newRowDeltaCommitTestTable`
gives a fresh table, so `deletedEntries()` is empty and no tombstones get
written before `validateAddedSpecs` fires. Every subtest here rejects during
the added-file phase with nothing on disk yet.
`TestParentDependentManifestsMixedSpecWritesNoManifests` has the same
limitation — it drives `parentDependentManifests` in isolation, never the full
`buildManifests` where the parent-dependent phase succeeds and the added-spec
phase then fails.
Once the validation moves up, I'd add a case that commits a file first, then
runs an overwrite with a valid deletion plus an unregistered added-file spec,
and asserts no new manifests. That's the path that's currently unguarded.
##########
table/snapshot_producers.go:
##########
@@ -626,12 +632,16 @@ func createSnapshotProducer(op Operation, txn
*Transaction, fs iceio.WriteFileIO
}
}
-func (sp *snapshotProducer) spec(id int) iceberg.PartitionSpec {
- if spec, _ := sp.txn.meta.GetSpecByID(id); spec != nil {
- return *spec
+func (sp *snapshotProducer) spec(id int) (iceberg.PartitionSpec, error) {
+ spec, err := sp.txn.meta.GetSpecByID(id)
+ if err == nil && spec == nil {
+ err = ErrPartitionSpecNotFound
+ }
+ if err != nil {
Review Comment:
I'd reserve the "unregistered" wording for the actual not-found case. As
written, any error out of `GetSpecByID` (a future decode/IO failure, say) comes
back to the caller labeled `"unregistered partition spec id N"`, which wouldn't
be true.
The `if err == nil && spec == nil` guard also looks dead — `GetSpecByID`
returns either `(*spec, nil)` or `(nil, err)`, never `(nil, nil)` — so the
not-found signal is already the error it hands back. Something like:
```go
spec, err := sp.txn.meta.GetSpecByID(id)
if errors.Is(err, ErrPartitionSpecNotFound) {
return iceberg.PartitionSpec{}, fmt.Errorf("unregistered partition spec
id %d: %w", id, err)
}
if err != nil {
return iceberg.PartitionSpec{}, err
}
return *spec, nil
```
Minor while we're here: the id lands in the message twice (`"unregistered
partition spec id 99: partition spec not found: id 99"`) — worth trimming one.
##########
table/row_delta_test.go:
##########
@@ -1427,3 +1427,164 @@ func assertRowCount(t *testing.T, tbl *table.Table,
expected int64) {
assert.Equal(t, expected, total, "unexpected row count")
}
+
+func unregisteredSpec() iceberg.PartitionSpec {
+ return iceberg.NewPartitionSpecID(99, iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id_identity",
Transform: iceberg.IdentityTransform{},
Review Comment:
Small one: this and `unregisteredSpec()` in `snapshot_producers_test.go`
(around line 2033) are near-duplicates both hard-coding id 99, but with
different `Name` values — `"id_identity"` here, `"id"` there. Different
packages so they can't be shared, but the mismatch reads as accidental, and
anyone who later registers spec 99 or evolves past it breaks both files
confusingly. I'd at least align the `Name`.
##########
table/snapshot_producers.go:
##########
@@ -788,6 +798,10 @@ func (sp *snapshotProducer) buildManifests(ctx
context.Context, parent *Snapshot
// written once and reused verbatim across OCC retries (rewriting them would
// churn manifest paths and orphan object-store files on every attempt).
func (sp *snapshotProducer) addedContentManifests() ([]iceberg.ManifestFile,
error) {
+ if err := sp.validateAddedSpecs(); err != nil {
Review Comment:
Looking at this validation path on its own, I think it closes the
sibling-goroutine race but not the cross-phase one. `buildManifests` calls
`parentDependentManifests` first, and that can write tombstone manifests for
deleted entries before we ever reach `addedContentManifests`, where
`validateAddedSpecs` actually runs.
So an overwrite with valid parent-dependent deletions plus an added file on
an unregistered spec writes the tombstones, then rejects, and those manifests
are orphaned. Same shape on the summary side: the new `removeFile` closure in
`accumulateSummaryDelta` errors on an unregistered deleted-file spec, but that
runs after `buildManifests` has already written everything.
I'd lift a single validation to the very top of `buildManifests`, before
either phase, and widen it to cover the to-be-deleted spec ids too
(`deletedFiles` / `deletedDeleteFiles` / `deletedDVsByRef`), not just
`addedFiles` — so nothing touches object storage until every spec id in the
commit is known-good. I left a note on
`TestRowDeltaMixedSpecCommitLeavesNoManifests` about the test that would guard
this.
--
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]
