laskoviymishka commented on code in PR #1973:
URL: https://github.com/apache/iceberg-go/pull/1973#discussion_r3903436370
##########
table/orphan_cleanup_test.go:
##########
@@ -1616,6 +1622,194 @@ 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 TestDeleteFilesParallelStopsQueuedWorkOnCancellation(t *testing.T) {
Review Comment:
This only exercises the pre-cancelled case: ctx is already cancelled before
the call, and with a single file `workers` is `min(max(4,1), 1) == 1`, so the
sender's `case <-ctx.Done()` fires immediately and nothing ever lands on
`jobs`. The post-receive guard (`if err := ctx.Err(); err != nil` inside the
worker) never runs, so this would still pass if that guard were deleted.
I'd add a second case that actually exercises mid-dispatch: say 50 files
with 4 workers, cancel from inside `deleteFunc` after a handful of calls
complete, then assert the deleted count is well short of 50 and `err` carries
`context.Canceled`. That's the path that proves queued work actually stops.
wdyt?
##########
table/orphan_cleanup_bench_test.go:
##########
@@ -124,3 +124,45 @@ func (fs *benchmarkDelayIO) Open(name string) (iceio.File,
error) {
return f, nil
}
+
+// BenchmarkPurgeFilesNonBulkDeletion measures the bounded fallback used when
+// the filesystem does not implement BulkRemovableIO. The delay models the
+// round trip to a remote object store; the zero-delay cases show the worker
+// pool overhead on local-style deletes.
+func BenchmarkPurgeFilesNonBulkDeletion(b *testing.B) {
+ for _, fileCount := range []int{100, 1_000, 10_000} {
+ files := make([]string, fileCount)
+ for i := range files {
+ files[i] =
fmt.Sprintf("s3://bucket/table/data/file-%d.parquet", i)
+ }
+
+ for _, delay := range []time.Duration{0, 100 *
time.Microsecond, time.Millisecond} {
+ for _, concurrency := range []int{1, 4, 16} {
+
b.Run(fmt.Sprintf("files=%d/delay=%s/concurrency=%d", fileCount, delay,
concurrency), func(b *testing.B) {
+ deleteFunc := func(string) error {
+ time.Sleep(delay)
+
+ return nil
+ }
+
+ b.ReportAllocs()
+ b.ReportMetric(float64(fileCount),
"files/op")
+ b.ResetTimer()
+ for b.Loop() {
+ deleted, err :=
deleteFilesParallel(
+ context.Background(),
+ files,
+ concurrency,
+ deleteFunc,
+ func(string, error)
error { return nil },
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ orphanCleanupBenchmarkSink =
deleted[len(deleted)-1]
Review Comment:
`deleted[len(deleted)-1]` is safe only because this bench's `deleteFunc`
always succeeds, so every index ends up in `deleted`. Worth noting the
underlying shape though: if a delete fails and `wrapError` returns nil (as it
does here), that index lands in neither `deleted` nor `deleteErrors` and
silently drops from both return slices. Extend this bench to model partial
failures and `deleted` could be empty, so this indexes out of bounds.
Simplest guard is to make `wrapError` actually wrap (`func(_ string, err
error) error { return err }`); that keeps the sink safe and surfaces any
failure through `b.Fatal`. wdyt?
##########
table/orphan_cleanup.go:
##########
@@ -1334,15 +1382,23 @@ func (t Table) PurgeFiles(ctx context.Context) error {
errs = append(errs, fmt.Errorf("bulk deletion
failed: %w", bulkErr))
}
} else {
- for _, file := range files {
- if err := ctx.Err(); err != nil {
- errs = append(errs, err)
+ _, removeErr := deleteFilesParallel(
+ ctx,
+ files,
+ runtime.GOMAXPROCS(0),
Review Comment:
`runtime.GOMAXPROCS(0)` is a CPU-count heuristic, but what we're
parallelizing here is object-store round trips, not CPU work. On a typical
cloud box that caps us at 2-8 concurrent deletes, and since throughput is
roughly concurrency over latency, 4 workers at ~100ms/op is ~40 files/sec.
That's a real win over sequential, but object-store clients usually run 32-64
in flight, so we're leaving close to an order of magnitude on the table.
The orphan-cleanup path already exposes `WithCleanupMaxConcurrency`;
`PurgeFiles` has no knob at all. I'd either bump the default to something
I/O-appropriate (32 is defensible) or add a `WithPurgeMaxConcurrency` option
mirroring the cleanup one, and note in the doc that GOMAXPROCS is a CPU default
that wants raising for remote stores. wdyt?
##########
table/orphan_cleanup.go:
##########
@@ -740,55 +758,85 @@ func deleteFilesSequential(fs iceio.IO, orphanFiles
[]string, cfg *orphanCleanup
return deletedFiles, result
}
-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
+ }
+ if err := ctx.Err(); err != nil {
Review Comment:
The outer `case <-ctx.Done()` and this inner `ctx.Err()` check look
redundant but both are load-bearing: when a job and `ctx.Done()` are both
ready, `select` picks at random, so this inner check is what stops a worker
from doing real I/O after cancellation when the random pick happened to land on
`jobs`. Correct and known pattern, just subtle. A one-line comment here would
save the next maintainer from re-deriving it.
##########
table/orphan_cleanup.go:
##########
@@ -740,55 +758,85 @@ func deleteFilesSequential(fs iceio.IO, orphanFiles
[]string, cfg *orphanCleanup
return deletedFiles, result
}
-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
+ }
+ if err := ctx.Err(); err != nil {
+ recordCancellation(err)
+
+ return
+ }
+
+ if err := deleteFunc(files[index]); err
!= nil {
+ deleteErrors[index] =
wrapError(files[index], err)
+ } else {
+ deleted[index] = true
+ }
}
}
- }(i)
+ }()
}
- go func() {
- wg.Wait()
- close(out)
- }()
+send:
+ for index := range files {
+ select {
+ case jobs <- index:
+ case <-ctx.Done():
+ recordCancellation(ctx.Err())
- deletedFiles := make([]string, 0, len(orphanFiles))
- for file := range out {
- deletedFiles = append(deletedFiles, file)
+ break send
+ }
}
+ close(jobs)
+ wg.Wait()
- var allErrors []error
- for _, workerErrors := range errList {
- allErrors = append(allErrors, workerErrors...)
+ deletedFiles := make([]string, 0, len(files))
+ allErrors := make([]error, 0)
Review Comment:
Small one: `var allErrors []error` reads cleaner here. `errors.Join` treats
nil and empty-non-nil the same, so the explicit `make([]error, 0)` signals an
intentional non-nil-empty that doesn't actually matter, and it trips the
perfsprint-style lints.
##########
table/orphan_cleanup.go:
##########
@@ -729,6 +742,11 @@ func deleteFilesSequential(fs iceio.IO, orphanFiles
[]string, cfg *orphanCleanup
var result error
for _, file := range orphanFiles {
+ if err := ctx.Err(); err != nil {
+ result = errors.Join(result, err)
Review Comment:
Minor asymmetry with the parallel path: here the context error gets
`errors.Join`'d into `result` inline, mid-chain with any delete failures,
whereas `deleteFilesParallel` appends `cancellationErr` last. Both share the
same caller in `deleteFiles`, so depending only on how `maxConcurrency` slices
the same input, callers get structurally different error nesting.
`errors.Is(context.Canceled)` works either way, but string and unwrap
inspection differ.
I'd accumulate the context error separately here and join it last too, so
the two paths produce the same shape. While we're at it, a line in the
`PurgeFiles` doc noting non-bulk errors now come back as a single joined value
would help, since the old path returned per-file errors directly.
--
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]
