zeroshade commented on code in PR #1938:
URL: https://github.com/apache/iceberg-go/pull/1938#discussion_r3937634949
##########
table/arrow_scanner.go:
##########
@@ -138,6 +143,133 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO,
tasks []FileScanTask,
return deletesPerFile, nil
}
+// lazyPositionDeleteLoader indexes positional-delete metadata for a scan, but
+// waits to open each delete file until a worker reaches a task that references
+// it. A delete file can apply to more than one data file, so the cache keeps
+// the complete grouped result for the delete file rather than caching only one
+// task's positions.
+//
+// The grouped Arrow chunks are owned by the loader until release. The iterator
+// calls release after all workers have stopped, which keeps shared chunks
alive
+// while multiple tasks use them and also covers early iterator termination.
+// The loader has the lifetime of exactly one scan. Each file's first load also
+// locks in its result, including context errors, for every caller; a loader
+// must not be reused for a retry with a different context.
+type lazyPositionDeleteLoader struct {
+ fs iceio.IO
+ files map[string]*lazyPositionDeleteFile
+
+ releaseOnce sync.Once
+ released atomic.Bool
+}
+
+var errPositionDeleteLoaderReleased = errors.New("position delete loader
already released")
+
+type lazyPositionDeleteFile struct {
+ dataFile iceberg.DataFile
+
+ once sync.Once
+ deletes map[string]*arrow.Chunked
+ err error
+}
+
+func newLazyPositionDeleteLoader(fs iceio.IO, tasks []FileScanTask)
*lazyPositionDeleteLoader {
+ loader := &lazyPositionDeleteLoader{
+ fs: fs,
+ files: make(map[string]*lazyPositionDeleteFile),
+ }
+
+ for _, task := range tasks {
+ for _, deleteFile := range task.DeleteFiles {
+ if deleteFile.ContentType() !=
iceberg.EntryContentPosDeletes {
+ continue
+ }
+
+ path := deleteFile.FilePath()
+ if _, ok := loader.files[path]; !ok {
+ loader.files[path] =
&lazyPositionDeleteFile{dataFile: deleteFile}
+ }
+ }
+ }
+
+ return loader
+}
+
+func (l *lazyPositionDeleteLoader) load(ctx context.Context, task
FileScanTask) (positionDeletes, error) {
+ if l.released.Load() {
+ return nil, errPositionDeleteLoaderReleased
+ }
+
+ if len(task.DeleteFiles) == 0 {
+ return nil, nil
+ }
+
+ targetPath := task.File.FilePath()
+ deletes := make(positionDeletes, 0, len(task.DeleteFiles))
+ // Most scan tasks carry one positional delete file. Avoid allocating a
+ // deduplication map unless there can actually be duplicate entries.
+ var seen map[string]struct{}
+ if len(task.DeleteFiles) > 1 {
+ seen = make(map[string]struct{}, len(task.DeleteFiles))
+ }
+ for _, deleteFile := range task.DeleteFiles {
+ if deleteFile.ContentType() != iceberg.EntryContentPosDeletes {
+ continue
+ }
+
+ path := deleteFile.FilePath()
+ if seen != nil {
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+ }
+
+ cached, ok := l.files[path]
+ if !ok {
+ // The loader is normally built from the same task
slice supplied to
+ // this method. Keep this guard so a malformed caller
cannot panic a
+ // scan if it changes a task after loader construction.
+ continue
+ }
+
+ cached.once.Do(func() {
+ cached.deletes, cached.err = readDeletes(ctx, l.fs,
cached.dataFile)
+ if cached.err != nil {
+ // readDeletes currently returns nil on errors.
Release defensively
+ // in case a future reader returns partial
Arrow ownership.
+ releasePosDeletes(cached.deletes)
+ cached.deletes = nil
+ cached.err = fmt.Errorf("read position deletes
from %s: %w",
+ cached.dataFile.FilePath(), cached.err)
+ }
+ })
+ if cached.err != nil {
+ return nil, cached.err
+ }
+
+ if chunk := cached.deletes[targetPath]; chunk != nil {
+ deletes = append(deletes, chunk)
+ }
Review Comment:
**minor** — Lazy loader narrows positional-delete scope from global to
task-scoped, changing scan results
The eager path built perFilePosDeletes keyed by every data-file path found
*inside* each delete file (arrow_scanner.go:134-138), so a delete file
referenced by one task also applied to any other task's data file it happened
to mention. load() instead consults only the delete files listed on the task
itself and looks up cached.deletes[targetPath] (:252-254). The new semantics
match Java's task-scoped DeleteFileIndex and are, I believe, the correct ones -
but this is a result-set change in a PR whose description says only error
timing moves, and no test pins it. Planner-built tasks should not hit it;
Scan.ReadTasks accepts caller-supplied tasks and can. Suggest a regression test
asserting a delete file is applied only to tasks that reference it, plus a line
in the PR description. Flagging for your judgement rather than blocking.
--
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]
