zeroshade commented on code in PR #1973:
URL: https://github.com/apache/iceberg-go/pull/1973#discussion_r3937964158
##########
table/orphan_cleanup_test.go:
##########
@@ -1616,6 +1623,256 @@ func (m mockFileInfo) ModTime() time.Time { return
m.modTime }
func (m mockFileInfo) IsDir() bool { return m.mode.IsDir() }
func (m mockFileInfo) Sys() any { return nil }
+type purgeDeleteTrackingIO struct {
+ mockListableIO
+
+ targetActive int
+ reached chan struct{}
+ release chan struct{}
+
+ mu sync.Mutex
+ active int
+ maxActive int
+ removed []string
+ reachedOnce sync.Once
+ releaseOnce sync.Once
+}
+
+func (m *purgeDeleteTrackingIO) Remove(name string) error {
+ m.mu.Lock()
+ m.active++
+ if m.active > m.maxActive {
+ m.maxActive = m.active
+ }
+ if m.targetActive > 0 && m.active >= m.targetActive {
+ m.reachedOnce.Do(func() { close(m.reached) })
+ }
+ m.mu.Unlock()
+
+ if m.targetActive > 0 {
+ <-m.release
+ }
+
+ m.mu.Lock()
+ m.active--
+ m.removed = append(m.removed, name)
+ m.mu.Unlock()
+
+ return nil
+}
+
+func (m *purgeDeleteTrackingIO) MaxActive() int {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ return m.maxActive
+}
+
+func (m *purgeDeleteTrackingIO) Release() {
+ m.releaseOnce.Do(func() { close(m.release) })
+}
+
+func TestDeleteFilesParallelCollectsPurgeErrors(t *testing.T) {
+ const (
+ firstPath = "s3://bucket/table/first.parquet"
+ missingPath = "s3://bucket/table/missing.parquet"
+ slowPath = "s3://bucket/table/slow.parquet"
+ fastPath = "s3://bucket/table/fast.parquet"
+ )
+
+ slowErr := errors.New("slow removal failed")
+ fastErr := errors.New("fast removal failed")
+ files := []string{firstPath, missingPath, slowPath, fastPath}
+ var mu sync.Mutex
+ calls := make(map[string]int)
+
+ deleted, err := deleteFilesParallel(
+ context.Background(),
+ files,
+ 4,
+ func(path string) error {
+ mu.Lock()
+ calls[path]++
+ mu.Unlock()
+
+ var err error
+ switch path {
+ case missingPath:
+ err = stdfs.ErrNotExist
+ case slowPath:
+ time.Sleep(10 * time.Millisecond)
+
+ err = slowErr
+ case fastPath:
+ err = fastErr
+ }
+ if os.IsNotExist(err) {
+ return nil
+ }
+
+ return err
+ },
+ func(path string, err error) error {
+ return fmt.Errorf("failed to remove %s: %w", path, err)
+ },
+ )
+
+ assert.Equal(t, []string{firstPath, missingPath}, deleted)
+ require.ErrorIs(t, err, slowErr)
+ require.ErrorIs(t, err, fastErr)
+ assert.NotContains(t, err.Error(), missingPath)
+ assert.Less(t, strings.Index(err.Error(), slowPath),
strings.Index(err.Error(), fastPath))
+
+ mu.Lock()
+ assert.Equal(t, map[string]int{
+ firstPath: 1,
+ missingPath: 1,
+ slowPath: 1,
+ fastPath: 1,
+ }, calls)
+ mu.Unlock()
+}
+
+func TestDeleteFilesParallelPreservesErrorWithNilWrapper(t *testing.T) {
+ deleteErr := errors.New("delete failed")
+ files := []string{
+ "s3://bucket/table/ok.parquet",
+ "s3://bucket/table/failed.parquet",
+ }
+
+ deleted, err := deleteFilesParallel(
+ context.Background(),
+ files,
+ 2,
+ func(path string) error {
+ if path == files[1] {
+ return deleteErr
+ }
+
+ return nil
+ },
+ func(string, error) error { return nil },
+ )
+
+ assert.Equal(t, []string{files[0]}, deleted)
+ require.ErrorIs(t, err, deleteErr)
+}
+
+func TestDeleteFilesParallelStopsQueuedWorkOnCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ called := false
+ _, err := deleteFilesParallel(
+ ctx,
+ []string{"s3://bucket/table/file.parquet"},
+ 4,
+ func(string) error {
+ called = true
+
+ return nil
+ },
+ func(path string, err error) error {
+ return fmt.Errorf("failed to remove %s: %w", path, err)
+ },
+ )
+
+ require.ErrorIs(t, err, context.Canceled)
+ assert.False(t, called)
+}
+
+func TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation(t
*testing.T) {
+ const (
+ fileCount = 50
+ maxConcurrency = 4
+ )
+
+ files := make([]string, fileCount)
+ for i := range files {
+ files[i] = fmt.Sprintf("s3://bucket/table/file-%02d.parquet", i)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ var calls atomic.Int32
+ release := make(chan struct{})
+ deleted, err := deleteFilesParallel(
+ ctx,
+ files,
+ maxConcurrency,
+ func(string) error {
+ call := calls.Add(1)
+ if call == maxConcurrency {
+ cancel()
+ close(release)
+ } else if call < maxConcurrency {
+ <-release
+ }
+
+ return nil
+ },
+ func(path string, err error) error {
+ return fmt.Errorf("failed to remove %s: %w", path, err)
+ },
+ )
+
+ require.ErrorIs(t, err, context.Canceled)
+ assert.Equal(t, int32(maxConcurrency), calls.Load())
+ assert.Less(t, len(deleted), fileCount)
+}
+
+func TestPurgeFilesDeletesNonBulkFilesConcurrently(t *testing.T) {
+ const maxWorkers = defaultPurgeMaxConcurrency
+ const fileCount = maxWorkers * 2
+ entries := make([]mockWalkEntry, 0, fileCount)
+ for i := range fileCount {
+ entries = append(entries, mockWalkEntry{
+ path:
fmt.Sprintf("s3://bucket/table/data/file-%02d.parquet", i),
+ info: mockFileInfo{name:
fmt.Sprintf("file-%02d.parquet", i)},
+ })
+ }
+
+ fsys := &purgeDeleteTrackingIO{
+ mockListableIO: mockListableIO{entries: entries},
+ targetActive: 2,
+ reached: make(chan struct{}),
+ release: make(chan struct{}),
+ }
+
+ meta, err := NewMetadata(
+ iceberg.NewSchema(0),
+ iceberg.UnpartitionedSpec,
+ UnsortedSortOrder,
+ "s3://bucket/table",
+ iceberg.Properties{},
+ )
+ require.NoError(t, err)
+ tbl := New(
+ Identifier{"db", "tbl"},
+ meta,
+ "s3://bucket/table/metadata/v1.metadata.json",
+ testFSF(fsys),
+ nil,
+ )
+
+ done := make(chan error, 1)
+ go func() { done <- tbl.PurgeFiles(context.Background()) }()
+
+ select {
+ case <-fsys.reached:
+ case <-time.After(5 * time.Second):
+ fsys.Release()
+ <-done
+ t.Fatal("non-bulk purge deletion did not reach two concurrent
removals")
+ }
+ fsys.Release()
+
+ require.NoError(t, <-done)
+ assert.Greater(t, fsys.MaxActive(), 1)
+ assert.LessOrEqual(t, fsys.MaxActive(), maxWorkers)
+}
Review Comment:
**major** — assert.LessOrEqual(fsys.MaxActive(), maxWorkers) is vacuous -
the concurrency cap is unverified
purgeDeleteTrackingIO is constructed with targetActive: 2 (line 1838), so
the test releases the barrier as soon as two removals are concurrently active.
The worker pool never saturates, so MaxActive() never approaches the cap and
the upper-bound assertion can never fail. Bounded concurrency is the headline
safety property of this PR and nothing pins it. Fix: set targetActive to
defaultPurgeMaxConcurrency so all workers are held in flight, then assert
MaxActive() == defaultPurgeMaxConcurrency exactly.
<details><summary>Evidence</summary>
```text
Replaced `workers := min(max(maxConcurrency, 1), len(files))`
(orphan_cleanup.go:773) with `workers := len(files)` -- 64 workers against a
documented cap of 32 -- and `go test -count=50 -run
TestPurgeFilesDeletesNonBulkFilesConcurrently ./table/` => `ok ... 0.556s`
(50/50 pass). A probe using a barrier that actually holds every worker observed
the correct behaviour: 'PROBE-A bound=8 files=64 observedPeak=8', confirming
the code is right and only the assertion is toothless.
```
</details>
##########
table/orphan_cleanup.go:
##########
@@ -737,58 +760,95 @@ func deleteFilesSequential(fs iceio.IO, orphanFiles
[]string, cfg *orphanCleanup
deletedFiles = append(deletedFiles, file)
}
- return deletedFiles, result
+ return deletedFiles, errors.Join(result, cancellationErr)
}
-func deleteFilesParallel(fs iceio.IO, orphanFiles []string, cfg
*orphanCleanupConfig) ([]string, error) {
- deleteFunc := fs.Remove
- if cfg.deleteFunc != nil {
- deleteFunc = cfg.deleteFunc
+func deleteFilesParallel(
+ ctx context.Context,
+ files []string,
+ maxConcurrency int,
+ deleteFunc func(string) error,
+ wrapError func(string, error) error,
+) ([]string, error) {
+ workers := min(max(maxConcurrency, 1), len(files))
+ jobs := make(chan int)
+ deleted := make([]bool, len(files))
+ deleteErrors := make([]error, len(files))
+
+ var cancellationErr error
+ var cancellationOnce sync.Once
+ recordCancellation := func(err error) {
+ cancellationOnce.Do(func() {
+ cancellationErr = err
+ })
}
- in := make(chan string, cfg.maxConcurrency)
- out := make(chan string, cfg.maxConcurrency)
- errList := make([][]error, cfg.maxConcurrency)
-
- go func() {
- defer close(in)
- for _, file := range orphanFiles {
- in <- file
- }
- }()
-
var wg sync.WaitGroup
- wg.Add(cfg.maxConcurrency)
- for i := range cfg.maxConcurrency {
- go func(workerID int) {
+ wg.Add(workers)
+ for range workers {
+ go func() {
defer wg.Done()
- for file := range in {
- if err := deleteFunc(file); err != nil {
- errList[workerID] =
append(errList[workerID], fmt.Errorf("failed to delete orphan file %s: %w",
file, err))
- } else {
- out <- file
+ for {
+ select {
+ case <-ctx.Done():
+ recordCancellation(ctx.Err())
+
+ return
+ case index, ok := <-jobs:
+ if !ok {
+ return
+ }
+ // Both cases can be ready after
cancellation, so check the context
+ // again before starting the deletion.
+ if err := ctx.Err(); err != nil {
+ recordCancellation(err)
+
+ return
+ }
+
+ if err := deleteFunc(files[index]); err
!= nil {
+ wrappedErr :=
wrapError(files[index], err)
+ if wrappedErr == nil {
+ // Keep failed
deletions observable even if the wrapper suppresses the error.
Review Comment:
**minor** — Defensive branch for a condition no caller can produce, plus a
test that exists only to cover it
`if wrappedErr == nil { wrappedErr = err }` guards against a wrapError that
returns nil. Both production call sites (orphan_cleanup.go:731 and :1409)
unconditionally return a non-nil fmt.Errorf, so this is unreachable in
production; the only caller that can trigger it is
TestDeleteFilesParallelPreservesErrorWithNilWrapper, written for that purpose.
The prior review asked for the bench's wrapError to be fixed (it was); this
production fallback is extra. AGENTS/house style: don't add error handling for
scenarios that can't happen. Suggest dropping both the branch and the test.
##########
io/io.go:
##########
@@ -59,6 +59,7 @@ type IO interface {
// Remove removes the named file or (empty) directory.
//
// If there is an error, it will be of type *PathError.
+ // Implementations must be safe for concurrent use by multiple
goroutines.
Remove(name string) error
Review Comment:
**minor** — Retroactive concurrency contract on the public IO interface
deserves a release note
'Implementations must be safe for concurrent use by multiple goroutines.'
tightens the contract of an exported extension point. Downstream users who
registered a custom IO via io.Register and whose Remove was only ever driven
serially by PurgeFiles will now be called from up to 32 goroutines. Every
in-repo implementation already complies, so nothing breaks here, but this is a
behavioural change for third-party implementors and should be called out in the
changelog rather than only in a doc comment.
##########
table/orphan_cleanup_test.go:
##########
@@ -1616,6 +1623,256 @@ func (m mockFileInfo) ModTime() time.Time { return
m.modTime }
func (m mockFileInfo) IsDir() bool { return m.mode.IsDir() }
func (m mockFileInfo) Sys() any { return nil }
+type purgeDeleteTrackingIO struct {
+ mockListableIO
+
+ targetActive int
+ reached chan struct{}
+ release chan struct{}
+
+ mu sync.Mutex
+ active int
+ maxActive int
+ removed []string
+ reachedOnce sync.Once
+ releaseOnce sync.Once
+}
+
+func (m *purgeDeleteTrackingIO) Remove(name string) error {
+ m.mu.Lock()
+ m.active++
+ if m.active > m.maxActive {
+ m.maxActive = m.active
+ }
+ if m.targetActive > 0 && m.active >= m.targetActive {
+ m.reachedOnce.Do(func() { close(m.reached) })
+ }
+ m.mu.Unlock()
+
+ if m.targetActive > 0 {
+ <-m.release
+ }
+
+ m.mu.Lock()
+ m.active--
+ m.removed = append(m.removed, name)
+ m.mu.Unlock()
+
+ return nil
+}
+
+func (m *purgeDeleteTrackingIO) MaxActive() int {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ return m.maxActive
+}
+
+func (m *purgeDeleteTrackingIO) Release() {
+ m.releaseOnce.Do(func() { close(m.release) })
+}
+
+func TestDeleteFilesParallelCollectsPurgeErrors(t *testing.T) {
+ const (
+ firstPath = "s3://bucket/table/first.parquet"
+ missingPath = "s3://bucket/table/missing.parquet"
+ slowPath = "s3://bucket/table/slow.parquet"
+ fastPath = "s3://bucket/table/fast.parquet"
+ )
+
+ slowErr := errors.New("slow removal failed")
+ fastErr := errors.New("fast removal failed")
+ files := []string{firstPath, missingPath, slowPath, fastPath}
+ var mu sync.Mutex
+ calls := make(map[string]int)
+
+ deleted, err := deleteFilesParallel(
+ context.Background(),
+ files,
+ 4,
+ func(path string) error {
+ mu.Lock()
+ calls[path]++
+ mu.Unlock()
+
+ var err error
+ switch path {
+ case missingPath:
+ err = stdfs.ErrNotExist
+ case slowPath:
+ time.Sleep(10 * time.Millisecond)
+
+ err = slowErr
+ case fastPath:
+ err = fastErr
+ }
+ if os.IsNotExist(err) {
+ return nil
+ }
+
+ return err
+ },
+ func(path string, err error) error {
+ return fmt.Errorf("failed to remove %s: %w", path, err)
+ },
+ )
+
+ assert.Equal(t, []string{firstPath, missingPath}, deleted)
+ require.ErrorIs(t, err, slowErr)
+ require.ErrorIs(t, err, fastErr)
+ assert.NotContains(t, err.Error(), missingPath)
+ assert.Less(t, strings.Index(err.Error(), slowPath),
strings.Index(err.Error(), fastPath))
+
+ mu.Lock()
+ assert.Equal(t, map[string]int{
+ firstPath: 1,
+ missingPath: 1,
+ slowPath: 1,
+ fastPath: 1,
+ }, calls)
+ mu.Unlock()
+}
+
+func TestDeleteFilesParallelPreservesErrorWithNilWrapper(t *testing.T) {
+ deleteErr := errors.New("delete failed")
+ files := []string{
+ "s3://bucket/table/ok.parquet",
+ "s3://bucket/table/failed.parquet",
+ }
+
+ deleted, err := deleteFilesParallel(
+ context.Background(),
+ files,
+ 2,
+ func(path string) error {
+ if path == files[1] {
+ return deleteErr
+ }
+
+ return nil
+ },
+ func(string, error) error { return nil },
+ )
+
+ assert.Equal(t, []string{files[0]}, deleted)
+ require.ErrorIs(t, err, deleteErr)
+}
+
+func TestDeleteFilesParallelStopsQueuedWorkOnCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ called := false
+ _, err := deleteFilesParallel(
+ ctx,
+ []string{"s3://bucket/table/file.parquet"},
+ 4,
+ func(string) error {
+ called = true
+
+ return nil
+ },
+ func(path string, err error) error {
+ return fmt.Errorf("failed to remove %s: %w", path, err)
+ },
+ )
+
+ require.ErrorIs(t, err, context.Canceled)
+ assert.False(t, called)
+}
+
+func TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation(t
*testing.T) {
+ const (
Review Comment:
**major** — Mid-flight cancellation test still does not reach the
post-receive ctx guard
TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation calls cancel()
and only then close(release), so ctx.Done() is always ready strictly before any
worker can return to the `case index, ok := <-jobs` receive. The sender
therefore always wins the select race and no worker ever reaches the inner
guard at orphan_cleanup.go:801-806. This is the exact defect the prior review
named ('would still pass if that guard were deleted') and it is still true.
Fix: close(release) BEFORE cancel() (or release a subset first) so workers are
contending on the jobs receive at the moment of cancellation, and assert
deleteFunc is not invoked again after cancellation is observed.
<details><summary>Evidence</summary>
```text
Deleted the guard block at orphan_cleanup.go:801-806 -> `go test -count=200
-run TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation ./table/` =>
`ok ... 0.582s`; full `go test -race -count=1 ./table/` => `ok ... 20.047s`.
Instrumented guard with an atomic counter: 'guard fired total: 0' across
TestDeleteFiles*/TestPurgeFiles*/TestOrphan*. An unbarriered probe on the same
code recorded 119 firings across 300 cancelled runs, proving the guard is
reachable and load-bearing -- just never by the suite.
```
</details>
--
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]
