laskoviymishka commented on code in PR #2005:
URL: https://github.com/apache/iceberg-go/pull/2005#discussion_r4007846878
##########
table/snapshot_producers.go:
##########
@@ -912,6 +1090,24 @@ func (sp *snapshotProducer) assembleManifests(ctx
context.Context, parent *Snaps
// files are inherited and the removed files are actually dropped — grafting
the
// attempt-0 result onto a fresh parent would resurrect the removed files.
func (sp *snapshotProducer) parentDependentManifests(ctx context.Context,
parent *Snapshot) (_ []iceberg.ManifestFile, err error) {
+ if of, ok := sp.producerImpl.(*overwriteFiles); ok {
+ parentManifests, err := of.filterAndRewriteParentManifests(ctx,
parent)
+ if err != nil {
+ return nil, err
+ }
+
+ deletedManifests, err :=
sp.writeDeletedEntries(parentManifests.deleted)
+ if err != nil {
+ if cleanupErr :=
sp.cleanupGeneratedManifests(parentManifests.rewritten); cleanupErr != nil {
Review Comment:
This second-stage cleanup has no test: `filterAndRewriteParentManifests`
succeeds and writes new manifests, then `writeDeletedEntries` fails and we
clean up `parentManifests.rewritten`.
`TestOverwriteExistingManifestsCleansCompletedRewritesOnError` covers a failure
inside `filterAndRewriteParentManifests`, but this branch is structurally
different, and if `rewritten` is ever mispopulated we silently leak the
rewritten manifests.
I'd add a test that injects a write failure into `writeDeletedEntries` after
a successful rewrite pass and asserts no files containing `sp.commitUuid`
survive.
##########
table/snapshot_producers.go:
##########
@@ -912,6 +1090,24 @@ func (sp *snapshotProducer) assembleManifests(ctx
context.Context, parent *Snaps
// files are inherited and the removed files are actually dropped — grafting
the
// attempt-0 result onto a fresh parent would resurrect the removed files.
func (sp *snapshotProducer) parentDependentManifests(ctx context.Context,
parent *Snapshot) (_ []iceberg.ManifestFile, err error) {
+ if of, ok := sp.producerImpl.(*overwriteFiles); ok {
Review Comment:
I think this dispatch is the thing I'd most want to resolve before merge.
The type assertion here routes every overwrite commit into
`filterAndRewriteParentManifests`, so `overwriteFiles.existingManifests` and
`deletedEntries` are never reached in production anymore, including on OCC
retries, since `assembleManifests` comes back through here. That leaves two
implementations of the same filtering logic (the concurrent one in
`filterAndRewriteParentManifests`, and the one behind the interface methods)
kept in sync by hand, with nothing at the compiler level enforcing it. A future
`scanManifest` fix has to be consciously mirrored, and any producer embedding
`overwriteFiles` silently gets the old two-scan path.
A couple of ways to handle it, no strong preference:
Fold both into a single interface method that returns survivors and
tombstones together (basically expose `overwriteParentManifests` through
`producerImpl`), so the concurrent path is the only path and the interface is
what production dispatches on.
Or, if the type-switch is meant to be the canonical routing, delete the
now-dead `existingManifests`/`deletedEntries` bodies on `overwriteFiles` and
leave a comment here saying overwrite is handled via
`filterAndRewriteParentManifests`. Right now the interface doc on
`existingManifests` still says it's re-evaluated on every OCC retry and must
read parent, which is misleading for a method production never calls.
Either is fine, but I'd like the two collapsed to one before this lands so
we're not maintaining parallel filtering logic. wdyt?
##########
table/snapshot_producers_test.go:
##########
@@ -1269,13 +1269,165 @@ func
TestOverwriteExistingManifestsClosesUnderlyingFile(t *testing.T) {
trackIO.writers = make(map[string]*trackingWriteCloser)
- _, err = sp.existingManifests(&snap)
+ _, err = sp.existingManifests(context.Background(), &snap)
require.NoError(t, err, "existingManifests should succeed")
unclosed := trackIO.GetUnclosedWriters()
require.Empty(t, unclosed, "all file writerFactory should be closed
after existingManifests, but these are still open: %v", unclosed)
}
+func TestOverwriteExistingManifestsLimitsConcurrencyAndPreservesOrder(t
*testing.T) {
+ const (
+ concurrency = 2
+ manifestCount = 4
+ )
+
+ spec := partitionedSpec()
+ schema := simpleSchema()
+ blockingIO := newBlockingCreateIO(1<<20, nil, concurrency)
+ defer blockingIO.Release()
+ txn := createTestTransaction(t, blockingIO, spec)
+ sp := newOverwriteFilesProducer(OpOverwrite, txn, blockingIO, nil, nil)
+ of := sp.producerImpl.(*overwriteFiles)
+ of.manifestConcurrency = concurrency
+
+ snapshotID := int64(100)
+ sequenceNumber := int64(-1)
+ manifestSequenceNumber := int64(42)
+ manifests := make([]iceberg.ManifestFile, 0, manifestCount)
+ for i := range manifestCount {
+ deletedFile := newTestDataFile(t, spec,
"file://deleted-"+strconv.Itoa(i)+".parquet", nil)
+ keptFile := newTestDataFile(t, spec,
"file://kept-"+strconv.Itoa(i)+".parquet", nil)
+ sp.deleteDataFile(deletedFile)
+
+ entries := []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
&snapshotID, &manifestSequenceNumber, nil, deletedFile),
+ iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
&snapshotID, &manifestSequenceNumber, nil, keptFile),
+ }
+ path := "table-location/metadata/source-" + strconv.Itoa(i) +
".avro"
+ manifests = append(manifests, writeTestManifestWithEntries(t,
blockingIO, spec, schema, snapshotID, path, entries))
+ }
+
+ manifestListPath := "table-location/metadata/snap-1.avro"
+ var listBuf bytes.Buffer
+ err := iceberg.WriteManifestList(2, &listBuf, snapshotID, nil,
&sequenceNumber, 0, manifests)
+ require.NoError(t, err, "write manifest list")
+ require.NoError(t, blockingIO.WriteFile(manifestListPath,
listBuf.Bytes()))
+
+ snap := Snapshot{
+ SnapshotID: snapshotID,
+ SequenceNumber: sequenceNumber,
+ ManifestList: manifestListPath,
+ }
+ txn.meta.snapshotList = []Snapshot{snap}
+ txn.meta.currentSnapshotID = &snapshotID
+
+ type result struct {
+ manifests []iceberg.ManifestFile
+ err error
+ }
+ done := make(chan result, 1)
+ go func() {
+ got, err := sp.existingManifests(context.Background(), &snap)
+ done <- result{manifests: got, err: err}
+ }()
+
+ select {
+ case <-blockingIO.reached:
+ case <-time.After(time.Second):
+ require.FailNow(t, "timed out waiting for overwrite manifest
workers")
+ }
+ time.Sleep(50 * time.Millisecond)
Review Comment:
The 50ms sleep makes this assertion racy. `blockingIO.reached` already fires
once `concurrency` workers are inside `Create`, and any extra goroutine blocks
on the semaphore before it can call `Create`, so `MaxActive()` is already
exactly `concurrency` the moment `reached` fires. On a slow CI box the sleep
can expire before anything extra gets scheduled and the test passes for the
wrong reason.
I'd drop the sleep, assert `MaxActive()` right after `reached`, then
`Release()`.
##########
table/snapshot_producers_test.go:
##########
@@ -1269,13 +1269,165 @@ func
TestOverwriteExistingManifestsClosesUnderlyingFile(t *testing.T) {
trackIO.writers = make(map[string]*trackingWriteCloser)
- _, err = sp.existingManifests(&snap)
+ _, err = sp.existingManifests(context.Background(), &snap)
require.NoError(t, err, "existingManifests should succeed")
unclosed := trackIO.GetUnclosedWriters()
require.Empty(t, unclosed, "all file writerFactory should be closed
after existingManifests, but these are still open: %v", unclosed)
}
+func TestOverwriteExistingManifestsLimitsConcurrencyAndPreservesOrder(t
*testing.T) {
+ const (
+ concurrency = 2
+ manifestCount = 4
+ )
+
+ spec := partitionedSpec()
+ schema := simpleSchema()
+ blockingIO := newBlockingCreateIO(1<<20, nil, concurrency)
+ defer blockingIO.Release()
+ txn := createTestTransaction(t, blockingIO, spec)
+ sp := newOverwriteFilesProducer(OpOverwrite, txn, blockingIO, nil, nil)
+ of := sp.producerImpl.(*overwriteFiles)
+ of.manifestConcurrency = concurrency
+
+ snapshotID := int64(100)
+ sequenceNumber := int64(-1)
+ manifestSequenceNumber := int64(42)
+ manifests := make([]iceberg.ManifestFile, 0, manifestCount)
+ for i := range manifestCount {
+ deletedFile := newTestDataFile(t, spec,
"file://deleted-"+strconv.Itoa(i)+".parquet", nil)
+ keptFile := newTestDataFile(t, spec,
"file://kept-"+strconv.Itoa(i)+".parquet", nil)
+ sp.deleteDataFile(deletedFile)
+
+ entries := []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
&snapshotID, &manifestSequenceNumber, nil, deletedFile),
+ iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
&snapshotID, &manifestSequenceNumber, nil, keptFile),
+ }
+ path := "table-location/metadata/source-" + strconv.Itoa(i) +
".avro"
+ manifests = append(manifests, writeTestManifestWithEntries(t,
blockingIO, spec, schema, snapshotID, path, entries))
+ }
+
+ manifestListPath := "table-location/metadata/snap-1.avro"
+ var listBuf bytes.Buffer
+ err := iceberg.WriteManifestList(2, &listBuf, snapshotID, nil,
&sequenceNumber, 0, manifests)
+ require.NoError(t, err, "write manifest list")
+ require.NoError(t, blockingIO.WriteFile(manifestListPath,
listBuf.Bytes()))
+
+ snap := Snapshot{
+ SnapshotID: snapshotID,
+ SequenceNumber: sequenceNumber,
+ ManifestList: manifestListPath,
+ }
+ txn.meta.snapshotList = []Snapshot{snap}
+ txn.meta.currentSnapshotID = &snapshotID
+
+ type result struct {
+ manifests []iceberg.ManifestFile
+ err error
+ }
+ done := make(chan result, 1)
+ go func() {
+ got, err := sp.existingManifests(context.Background(), &snap)
Review Comment:
The new correctness and ordering coverage all drives `sp.existingManifests`
directly, but production takes the `filterAndRewriteParentManifests` path from
`parentDependentManifests`, so these tests exercise a branch overwrite commits
never hit. The only thing touching the real path is the benchmark, and it just
asserts no error.
I'd add a test that calls `sp.parentDependentManifests` (or runs a full
overwrite/delete commit) on an `*overwriteFiles` producer and asserts the
actual output: surviving manifests hold only the kept entries, a delete
manifest is written for the removed files, the delete manifests sort ahead of
the existing ones, and the tombstone entries carry the right snapshot id and
sequence numbers. That's the path that ships.
##########
table/transaction.go:
##########
@@ -2450,6 +2450,7 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx
context.Context, operation
commitUUID := uuid.New()
updater := t.updateSnapshot(wfs, snapshotProps,
operation).mergeOverwrite(&commitUUID, filter)
+ updater.producerImpl.(*overwriteFiles).manifestConcurrency = concurrency
Review Comment:
This assertion (and the matching one in `performMergeOnReadDeletion` at line
2501) panics if `mergeOverwrite` ever returns a different producer type, and it
reaches past the constructor to set a field directly.
The two existing `producerImpl.(*overwriteFiles)` sites carry a `//
mergeOverwrite guarantees an *overwriteFiles producerImpl.` comment, so I'd at
least add that here too. Better still, expose `manifestConcurrency` as an
unexported option or setter on `newOverwriteFilesProducer` and go through that.
##########
table/snapshot_producers.go:
##########
@@ -912,6 +1090,24 @@ func (sp *snapshotProducer) assembleManifests(ctx
context.Context, parent *Snaps
// files are inherited and the removed files are actually dropped — grafting
the
// attempt-0 result onto a fresh parent would resurrect the removed files.
func (sp *snapshotProducer) parentDependentManifests(ctx context.Context,
parent *Snapshot) (_ []iceberg.ManifestFile, err error) {
+ if of, ok := sp.producerImpl.(*overwriteFiles); ok {
+ parentManifests, err := of.filterAndRewriteParentManifests(ctx,
parent)
+ if err != nil {
+ return nil, err
+ }
+
+ deletedManifests, err :=
sp.writeDeletedEntries(parentManifests.deleted)
Review Comment:
Not a blocker, and I think it's fine for the common case, but worth
flagging: previously `writeDeletedEntries` ran concurrently with the
existing-manifest scan in its own goroutine, and here it's serialized after
`filterAndRewriteParentManifests` returns. For a table with few manifests but
many deleted entries, the tombstone write is now on the critical path where it
used to overlap.
Probably not worth pipelining now. Could we drop a comment noting the
trade-off so it reads as a deliberate choice? wdyt?
##########
table/snapshot_producers.go:
##########
@@ -924,92 +1120,115 @@ func (sp *snapshotProducer)
parentDependentManifests(ctx context.Context, parent
if len(deleted) > 0 {
g.Go(func() error {
- // Group deleted entries by (specID, contentType) to
ensure data and
- // delete file entries are written to separate
manifests with the
- // correct ManifestContent.
- type groupKey struct {
- specID int
- content iceberg.ManifestContent
- }
- groups := map[groupKey][]iceberg.ManifestEntry{}
- for _, entry := range deleted {
- content := iceberg.ManifestContentData
- if entry.DataFile().ContentType() !=
iceberg.EntryContentData {
- content = iceberg.ManifestContentDeletes
- }
- key := groupKey{specID:
int(entry.DataFile().SpecID()), content: content}
- groups[key] = append(groups[key], entry)
- }
+ var err error
+ deletedFilesManifests, err =
sp.writeDeletedEntries(deleted)
- writeGroup := func(key groupKey, entries
[]iceberg.ManifestEntry) (_ iceberg.ManifestFile, retErr error) {
- spec, err := sp.spec(key.specID)
- if err != nil {
- return nil, err
- }
+ return err
+ })
+ }
- wr, path, counter, out, err :=
sp.newManifestWriter(spec,
-
iceberg.WithManifestWriterContent(key.content))
- if err != nil {
- return nil, err
- }
- defer internal.CheckedClose(out, &retErr)
+ g.Go(func() error {
+ m, err := sp.existingManifests(ctx, parent)
+ if err != nil {
+ return err
+ }
+ existingManifests = m
- writerClosed := false
- defer func() {
- if !writerClosed {
- internal.CheckedClose(wr,
&retErr)
- }
- }()
+ return nil
+ })
- for _, entry := range entries {
- if err := wr.Delete(entry); err != nil {
- return nil, err
- }
- }
+ if err := g.Wait(); err != nil {
+ return nil, err
+ }
- writerClosed = true
- if err := wr.Close(); err != nil {
- return nil, err
- }
+ return slices.Concat(deletedFilesManifests, existingManifests), nil
+}
- return wr.ToManifestFile(path, counter.Count,
iceberg.WithManifestFileContent(key.content))
- }
+func (sp *snapshotProducer) writeDeletedEntries(deleted
[]iceberg.ManifestEntry) (_ []iceberg.ManifestFile, retErr error) {
Review Comment:
`writeDeletedEntries` does all its manifest I/O without a context, so once
we're in here the writes can't be cancelled. In the overwrite fast path it runs
after `filterAndRewriteParentManifests`, so a context that's already been
cancelled by then won't stop it, and every other I/O function in this PR takes
and checks `ctx`.
I'd thread `ctx` through here and into `writeGroup`, with a
`context.Cause(ctx)` check in the key loop, matching `rewriteManifest`.
--
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]
