This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new c12bf38e5 fix(io): make MemFS WalkDir emit directory entries (#1604)
c12bf38e5 is described below
commit c12bf38e523820e3f12611c8d4d89aaf4f17f23b
Author: Minh Vu <[email protected]>
AuthorDate: Fri Jul 31 16:22:18 2026 +0200
fix(io): make MemFS WalkDir emit directory entries (#1604)
## Summary
- synthesize root and intermediate directory entries in `MemFS.WalkDir`
- emit entries in lexical order
- implement `SkipDir` and `SkipAll` behavior
- keep callbacks outside the filesystem lock
## Why
`ListableIO.WalkDir` requires callbacks for every file or directory,
including the root. MemFS previously emitted only files in map iteration
order, so directory-aware behavior and `SkipDir` differed from other
filesystem implementations.
## Testing
- `go test ./io`
---
io/mem.go | 119 ++++++++++++++++++++++++++++++++++++++++++++++++---------
io/mem_test.go | 112 ++++++++++++++++++++++++++++++++++++++++++++++++-----
2 files changed, 204 insertions(+), 27 deletions(-)
diff --git a/io/mem.go b/io/mem.go
index ffb5af20b..678d3c6cd 100644
--- a/io/mem.go
+++ b/io/mem.go
@@ -23,7 +23,8 @@ import (
"io"
"io/fs"
"net/url"
- "path/filepath"
+ "path"
+ "sort"
"strings"
"sync"
"time"
@@ -75,7 +76,7 @@ func (m *MemFS) Open(name string) (File, error) {
cp := make([]byte, len(data))
copy(cp, data)
- return &memFile{data: cp, name: filepath.Base(name)}, nil
+ return &memFile{data: cp, name: path.Base(name)}, nil
}
func (m *MemFS) Remove(name string) error {
@@ -107,28 +108,103 @@ func (m *MemFS) WriteFile(name string, content []byte)
error {
func (m *MemFS) WalkDir(root string, fn fs.WalkDirFunc) error {
type walkEntry struct {
- path string
- size int64
+ path string
+ size int64
+ isDir bool
}
root = strings.TrimRight(root, "/")
m.mu.RLock()
- var entries []walkEntry
+ entriesByPath := make(map[string]walkEntry)
+ addEntry := func(entry walkEntry) {
+ current, exists := entriesByPath[entry.path]
+ if !exists || (!entry.isDir && current.isDir) {
+ entriesByPath[entry.path] = entry
+ }
+ }
for key, data := range m.files {
if !memPathInRoot(key, root) {
continue
}
- entries = append(entries, walkEntry{
- path: key,
- size: int64(len(data)),
- })
+
+ if key == root {
+ addEntry(walkEntry{path: key, size: int64(len(data))})
+
+ continue
+ }
+
+ addEntry(walkEntry{path: root, isDir: true})
+ relative := strings.TrimPrefix(key, root+"/")
+ parts := strings.Split(relative, "/")
+ for i := range parts {
+ entryPath := strings.Join(parts[:i+1], "/")
+ if root != "" {
+ entryPath = root + "/" + entryPath
+ }
+ entry := walkEntry{path: entryPath, isDir: i <
len(parts)-1}
+ if !entry.isDir {
+ entry.size = int64(len(data))
+ }
+ addEntry(entry)
+ }
}
m.mu.RUnlock()
+ if len(entriesByPath) == 0 {
+ err := fn(root, nil, &fs.PathError{Op: "lstat", Path: root,
Err: fs.ErrNotExist})
+ if err == fs.SkipDir || err == fs.SkipAll {
+ return nil
+ }
- for _, entry := range entries {
- info := &memFileInfo{name: filepath.Base(entry.path), size:
entry.size}
- if err := fn(entry.path, fs.FileInfoToDirEntry(info), nil); err
!= nil {
+ return err
+ }
+ if rootEntry, ok := entriesByPath[root]; ok && !rootEntry.isDir {
+ entriesByPath = map[string]walkEntry{root: rootEntry}
+ }
+
+ paths := make([]string, 0, len(entriesByPath))
+ for entryPath := range entriesByPath {
+ paths = append(paths, entryPath)
+ }
+ sort.Slice(paths, func(i, j int) bool {
+ if paths[i] == root {
+ return true
+ }
+ if paths[j] == root {
+ return false
+ }
+
+ left := strings.Split(strings.TrimPrefix(paths[i], root+"/"),
"/")
+ right := strings.Split(strings.TrimPrefix(paths[j], root+"/"),
"/")
+ for index := 0; index < min(len(left), len(right)); index++ {
+ if left[index] != right[index] {
+ return left[index] < right[index]
+ }
+ }
+
+ return len(left) < len(right)
+ })
+
+ var skipPrefix string
+ for _, entryPath := range paths {
+ if skipPrefix != "" && strings.HasPrefix(entryPath, skipPrefix)
{
+ continue
+ }
+
+ entry := entriesByPath[entryPath]
+ info := &memFileInfo{name: path.Base(entry.path), size:
entry.size, isDir: entry.isDir}
+ err := fn(entry.path, fs.FileInfoToDirEntry(info), nil)
+ switch err {
+ case nil:
+ case fs.SkipAll:
+ return nil
+ case fs.SkipDir:
+ if entry.isDir {
+ skipPrefix = entry.path + "/"
+ } else {
+ skipPrefix = path.Dir(entry.path) + "/"
+ }
+ default:
return err
}
}
@@ -198,15 +274,22 @@ func (f *memFile) Stat() (fs.FileInfo, error) {
}
type memFileInfo struct {
- name string
- size int64
+ name string
+ size int64
+ isDir bool
}
-func (fi *memFileInfo) Name() string { return fi.name }
-func (fi *memFileInfo) Size() int64 { return fi.size }
-func (fi *memFileInfo) Mode() fs.FileMode { return 0 }
+func (fi *memFileInfo) Name() string { return fi.name }
+func (fi *memFileInfo) Size() int64 { return fi.size }
+func (fi *memFileInfo) Mode() fs.FileMode {
+ if fi.isDir {
+ return fs.ModeDir
+ }
+
+ return 0
+}
func (fi *memFileInfo) ModTime() time.Time { return time.Time{} }
-func (fi *memFileInfo) IsDir() bool { return false }
+func (fi *memFileInfo) IsDir() bool { return fi.isDir }
func (fi *memFileInfo) Sys() any { return nil }
type memWriter struct {
diff --git a/io/mem_test.go b/io/mem_test.go
index 6c3fceb51..f396acbaa 100644
--- a/io/mem_test.go
+++ b/io/mem_test.go
@@ -186,7 +186,8 @@ func TestMemIO_WalkDir(t *testing.T) {
return nil
})
require.NoError(t, err)
- assert.ElementsMatch(t, []string{
+ assert.Equal(t, []string{
+ "mem://walkdir-bucket/a",
"mem://walkdir-bucket/a/1.txt",
"mem://walkdir-bucket/a/2.txt",
}, walked)
@@ -218,7 +219,8 @@ func TestMemIO_WalkDirDoesNotIncludeSiblingPrefixes(t
*testing.T) {
return nil
})
require.NoError(t, err)
- assert.ElementsMatch(t, []string{
+ assert.Equal(t, []string{
+ "mem://walkdir-boundary-bucket/table",
"mem://walkdir-boundary-bucket/table/data.parquet",
}, walked)
})
@@ -232,10 +234,13 @@ func TestMemIO_WalkDirCallbackCanRemoveFiles(t
*testing.T) {
done := make(chan error, 1)
go func() {
- done <- memIO.WalkDir("mem://bucket/root", func(path string, _
fs.DirEntry, err error) error {
+ done <- memIO.WalkDir("mem://bucket/root", func(path string, d
fs.DirEntry, err error) error {
if err != nil {
return err
}
+ if d.IsDir() {
+ return nil
+ }
return memIO.Remove(path)
})
@@ -248,14 +253,103 @@ func TestMemIO_WalkDirCallbackCanRemoveFiles(t
*testing.T) {
t.Fatal("WalkDir deadlocked when its callback removed a file")
}
- var remaining []string
- require.NoError(t, memIO.WalkDir("mem://bucket/root", func(path string,
_ fs.DirEntry, err error) error {
- if err != nil {
- return err
+ _, err := memIO.Open("mem://bucket/root/1.txt")
+ require.ErrorIs(t, err, fs.ErrNotExist)
+ _, err = memIO.Open("mem://bucket/root/2.txt")
+ require.ErrorIs(t, err, fs.ErrNotExist)
+}
+
+func TestMemIO_WalkDirEmitsDirectoriesLexicallyAndHonorsSkipDir(t *testing.T) {
+ memIO := icebergio.NewMemFS()
+ require.NoError(t, memIO.WriteFile("mem://bucket/root/b/file.txt",
[]byte("b")))
+ require.NoError(t,
memIO.WriteFile("mem://bucket/root/a/nested/file.txt", []byte("a")))
+
+ var walked []string
+ err := memIO.WalkDir("mem://bucket/root", func(path string, d
fs.DirEntry, err error) error {
+ require.NoError(t, err)
+ walked = append(walked, path)
+ if path == "mem://bucket/root/a" {
+ require.True(t, d.IsDir())
+
+ return fs.SkipDir
}
- remaining = append(remaining, path)
+
+ return nil
+ })
+
+ require.NoError(t, err)
+ assert.Equal(t, []string{
+ "mem://bucket/root",
+ "mem://bucket/root/a",
+ "mem://bucket/root/b",
+ "mem://bucket/root/b/file.txt",
+ }, walked)
+}
+
+func TestMemIO_WalkDirUsesPreorderTraversal(t *testing.T) {
+ memIO := icebergio.NewMemFS()
+ require.NoError(t, memIO.WriteFile("mem://bucket/root/a/child.txt",
nil))
+ require.NoError(t, memIO.WriteFile("mem://bucket/root/a.txt", nil))
+
+ var walked []string
+ require.NoError(t, memIO.WalkDir("mem://bucket/root", func(path string,
_ fs.DirEntry, err error) error {
+ require.NoError(t, err)
+ walked = append(walked, path)
+
+ return nil
+ }))
+ assert.Equal(t, []string{
+ "mem://bucket/root",
+ "mem://bucket/root/a",
+ "mem://bucket/root/a/child.txt",
+ "mem://bucket/root/a.txt",
+ }, walked)
+}
+
+func TestMemIO_WalkDirTreatsFileRootAsFile(t *testing.T) {
+ memIO := icebergio.NewMemFS()
+ require.NoError(t, memIO.WriteFile("mem://bucket/root", []byte("root")))
+ require.NoError(t, memIO.WriteFile("mem://bucket/root/child.txt", nil))
+
+ var walked []string
+ require.NoError(t, memIO.WalkDir("mem://bucket/root", func(path string,
d fs.DirEntry, err error) error {
+ require.NoError(t, err)
+ assert.False(t, d.IsDir())
+ walked = append(walked, path)
return nil
}))
- assert.Empty(t, remaining)
+ assert.Equal(t, []string{"mem://bucket/root"}, walked)
+}
+
+func TestMemIO_WalkDirReportsMissingRoot(t *testing.T) {
+ memIO := icebergio.NewMemFS()
+ var callbackErr error
+
+ err := memIO.WalkDir("mem://bucket/missing", func(_ string, d
fs.DirEntry, err error) error {
+ assert.Nil(t, d)
+ callbackErr = err
+
+ return err
+ })
+
+ require.ErrorIs(t, err, fs.ErrNotExist)
+ require.ErrorIs(t, callbackErr, fs.ErrNotExist)
+}
+
+func TestMemIO_WalkDirHonorsSkipAtRootAndSkipAll(t *testing.T) {
+ memIO := icebergio.NewMemFS()
+ require.NoError(t, memIO.WriteFile("mem://bucket/root/a.txt", nil))
+
+ for _, skipErr := range []error{fs.SkipDir, fs.SkipAll} {
+ var walked []string
+ err := memIO.WalkDir("mem://bucket/root", func(path string, _
fs.DirEntry, err error) error {
+ require.NoError(t, err)
+ walked = append(walked, path)
+
+ return skipErr
+ })
+ require.NoError(t, err)
+ assert.Equal(t, []string{"mem://bucket/root"}, walked)
+ }
}
