laskoviymishka commented on code in PR #1245:
URL: https://github.com/apache/iceberg-go/pull/1245#discussion_r3455846585
##########
table/arrow_scanner.go:
##########
@@ -108,12 +108,9 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO,
tasks []FileScanTask,
if deletes == nil {
return nil
}
- select {
- case perFileChan <- deletes:
- return nil
- case <-gctx.Done():
- return gctx.Err()
- }
+ perFileChan <- deletes
Review Comment:
Dropping the select is the right simplification, but this send is now safe
only because the channel capacity equals `g`'s limit — every concurrent worker
can send without blocking, even after `gctx` is cancelled, and the for-range
below drains until close. That invariant is load-bearing now, not an
optimization, and nothing enforces it: if the buffer and `g.SetLimit` ever
drift apart this deadlocks. I'd drop a short comment right here so the next
person doesn't "fix" one without re-deriving the other.
```go
// Unconditional send is safe: the channel buffer equals g's concurrency
// limit, so every worker can send without blocking even after gctx is
// cancelled, and the for-range below drains until perFileChan is closed.
perFileChan <- deletes
```
##########
table/arrow_scanner_test.go:
##########
@@ -18,18 +18,70 @@
package table
import (
+ iofs "io/fs"
"strconv"
"strings"
+ "sync"
"testing"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+type failAfterGoodCloseFS struct {
+ *iceio.MemFS
+
+ goodPath string
+ badPath string
+
+ goodClosed chan struct{}
+ closeGoodClosed sync.Once
Review Comment:
`closeGoodClosed` parses as an adjective phrase rather than the `sync.Once`
guarding `close(goodClosed)`. I'd rename it `goodClosedOnce` to mirror the
channel name — reads as "the Once for goodClosed" at the call site.
##########
table/arrow_scanner_test.go:
##########
@@ -18,18 +18,70 @@
package table
import (
+ iofs "io/fs"
"strconv"
"strings"
+ "sync"
"testing"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+type failAfterGoodCloseFS struct {
+ *iceio.MemFS
+
+ goodPath string
+ badPath string
+
+ goodClosed chan struct{}
+ closeGoodClosed sync.Once
+}
+
+func (f *failAfterGoodCloseFS) Open(name string) (iceio.File, error) {
+ if name == f.goodPath {
+ file, err := f.MemFS.Open(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return &closeSignalFile{
+ File: file,
+ onClose: func() {
+ f.closeGoodClosed.Do(func() {
close(f.goodClosed) })
+ },
+ }, nil
+ }
+
+ if name == f.badPath {
+ <-f.goodClosed
Review Comment:
This blocking receive is the whole point of the helper and it's fine as-is,
but it has no escape hatch: if the good file's `Close` ever stops firing (a
panic before the defer, or a future refactor that drops it) the test hangs with
no diagnostic. A `select` with a `case <-time.After(...)` → `t.Fatal` would
turn that into an explicit failure instead of a silent timeout. Minor — only if
you feel like hardening the helper.
##########
table/arrow_scanner_test.go:
##########
@@ -142,6 +223,41 @@ func chunkedPosDelete(t *testing.T, mem memory.Allocator,
positions []int64) *ar
return arrow.NewChunked(arrow.PrimitiveTypes.Int64, []arrow.Array{arr})
}
+func TestReadAllDeleteFilesReturnsPartialDeletesOnError(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ ctx := compute.WithAllocator(t.Context(), mem)
+ goodDeletePath := "mem://bucket/deletes/good.parquet"
+ badDeletePath := "mem://bucket/deletes/missing.parquet"
+ dataPath := "mem://bucket/data/data.parquet"
+
+ memFS := iceio.NewMemFS()
+ writePosDeleteParquetToMemFS(t, memFS, goodDeletePath, `[
+ {"file_path": "`+dataPath+`", "pos": 1},
+ {"file_path": "`+dataPath+`", "pos": 3}
+ ]`)
+ testFS := &failAfterGoodCloseFS{
+ MemFS: memFS,
+ goodPath: goodDeletePath,
+ badPath: badDeletePath,
+ goodClosed: make(chan struct{}),
+ }
+ tasks := []FileScanTask{{
+ DeleteFiles: []iceberg.DataFile{
+ newPosDeleteFile(t, goodDeletePath, 2, 128),
+ newPosDeleteFile(t, badDeletePath, 1, 128),
+ },
+ }}
+
+ deletesPerFile, err := readAllDeleteFiles(ctx, testFS, tasks, 2)
Review Comment:
With concurrency=2 both files run in the same wave, so the good worker
finishes and sends before the bad one fails — this lands in the
success-then-return-partial path but never the cancel-while-between-send window
the description calls out, which means it'd pass against the old code too and
doesn't isolate that race. The `return deletesPerFile, err` half is what this
actually pins down (without it `deletesPerFile` comes back nil and
`require.NotNil` fails), so the leak side is covered — I'd just add a line
noting concurrency=2 is enough to observe the partial result and you're
deliberately not asserting the cancellation race.
While we're here, `require.NotEmpty` passes on any non-empty chunk — the
good file has 2 rows (pos 1, 3), so `require.Len(t, deletesPerFile[dataPath],
1)` (and ideally checking the two positions) would catch a regression that
accumulates only a partial chunk.
--
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]
