This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 045c40537 perf(io): avoid copying MemFS contents on open (#2233)
045c40537 is described below
commit 045c4053772fab695fe73f84bb6253dd7c64e010
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 26 18:03:32 2026 +0100
perf(io): avoid copying MemFS contents on open (#2233)
* perf(io): avoid copying MemFS contents on open
Signed-off-by: Minh Vu <[email protected]>
* Merge branch 'main' into perf/memfs-zero-copy-open-20261020
Pass a context to MemFS.Open/Remove in the new tests after #2067 added ctx
to IO.Open and IO.Remove.
---------
Signed-off-by: Minh Vu <[email protected]>
Co-authored-by: Matt Topol <[email protected]>
---
io/mem.go | 8 +-
io/mem_open_bench_test.go | 54 ++++++++++++
io/mem_open_snapshot_test.go | 193 +++++++++++++++++++++++++++++++++++++++++++
3 files changed, 251 insertions(+), 4 deletions(-)
diff --git a/io/mem.go b/io/mem.go
index ecc5541d7..ef46346ba 100644
--- a/io/mem.go
+++ b/io/mem.go
@@ -58,7 +58,7 @@ func init() {
// Use the "mem://bucket/path" URI scheme to access files.
type MemFS struct {
mu sync.RWMutex
- files map[string][]byte // keyed by full URI
+ files map[string][]byte // immutable snapshots, keyed by full URI
}
// NewMemFS creates an initialized MemFS ready for use.
@@ -73,10 +73,10 @@ func (m *MemFS) Open(_ context.Context, name string) (File,
error) {
if !ok {
return nil, &fs.PathError{Op: "open", Path: name, Err:
fs.ErrNotExist}
}
- cp := make([]byte, len(data))
- copy(cp, data)
- return &memFile{data: cp, name: path.Base(name)}, nil
+ // WriteFile stores a copy that is never mutated. Replacing or removing
+ // a path does not change the bytes held by an already-open reader.
+ return &memFile{data: data, name: path.Base(name)}, nil
}
func (m *MemFS) Remove(_ context.Context, name string) error {
diff --git a/io/mem_open_bench_test.go b/io/mem_open_bench_test.go
new file mode 100644
index 000000000..a3e3db10f
--- /dev/null
+++ b/io/mem_open_bench_test.go
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package io_test
+
+import (
+ "fmt"
+ "testing"
+
+ icebergio "github.com/apache/iceberg-go/io"
+)
+
+func BenchmarkMemFSOpen(b *testing.B) {
+ for _, sizeMiB := range []int{1, 16} {
+ b.Run(fmt.Sprintf("%dMiB", sizeMiB), func(b *testing.B) {
+ memFS := icebergio.NewMemFS()
+ name := "mem://bucket/data.parquet"
+ size := sizeMiB << 20
+ if err := memFS.WriteFile(name, make([]byte, size));
err != nil {
+ b.Fatal(err)
+ }
+
+ var lastByte [1]byte
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ file, err := memFS.Open(b.Context(), name)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if _, err := file.ReadAt(lastByte[:],
int64(size-1)); err != nil {
+ b.Fatal(err)
+ }
+ if err := file.Close(); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+}
diff --git a/io/mem_open_snapshot_test.go b/io/mem_open_snapshot_test.go
new file mode 100644
index 000000000..3e011dd87
--- /dev/null
+++ b/io/mem_open_snapshot_test.go
@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package io_test
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "io/fs"
+ "sync"
+ "testing"
+
+ icebergio "github.com/apache/iceberg-go/io"
+)
+
+func TestMemFSOpenPreservesSnapshots(t *testing.T) {
+ memFS := icebergio.NewMemFS()
+ name := "mem://bucket/data.parquet"
+ input := []byte("original")
+ if err := memFS.WriteFile(name, input); err != nil {
+ t.Fatal(err)
+ }
+
+ oldFile, err := memFS.Open(t.Context(), name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer oldFile.Close()
+
+ // The caller's input must not alias a published file.
+ input[0] = 'X'
+ if err := memFS.WriteFile(name, []byte("replacement")); err != nil {
+ t.Fatal(err)
+ }
+ newFile, err := memFS.Open(t.Context(), name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer newFile.Close()
+
+ if err := memFS.Remove(t.Context(), name); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := memFS.Open(t.Context(), name); !errors.Is(err,
fs.ErrNotExist) {
+ t.Fatalf("open after remove: got %v, want fs.ErrNotExist", err)
+ }
+
+ for _, test := range []struct {
+ file icebergio.File
+ want string
+ }{
+ {oldFile, "original"},
+ {newFile, "replacement"},
+ } {
+ got, err := io.ReadAll(test.file)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(got) != test.want {
+ t.Fatalf("read after replacement/removal: got %q, want
%q", got, test.want)
+ }
+ if _, err := test.file.Seek(0, io.SeekStart); err != nil {
+ t.Fatal(err)
+ }
+ at := make([]byte, len(test.want))
+ if _, err := test.file.ReadAt(at, 0); err != nil {
+ t.Fatal(err)
+ }
+ if string(at) != test.want {
+ t.Fatalf("readat after replacement/removal: got %q,
want %q", at, test.want)
+ }
+ info, err := test.file.Stat()
+ if err != nil || info.Size() != int64(len(test.want)) {
+ t.Fatalf("stat after replacement/removal: info=%v
err=%v", info, err)
+ }
+ }
+}
+
+func TestMemFSOpenPreservesSnapshotAcrossCreate(t *testing.T) {
+ memFS := icebergio.NewMemFS()
+ name := "mem://bucket/created.parquet"
+
+ firstWriter, err := memFS.Create(name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := firstWriter.Write([]byte("first")); err != nil {
+ t.Fatal(err)
+ }
+ if err := firstWriter.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ oldFile, err := memFS.Open(t.Context(), name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer oldFile.Close()
+
+ secondWriter, err := memFS.Create(name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := secondWriter.Write([]byte("second")); err != nil {
+ t.Fatal(err)
+ }
+ if err := secondWriter.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ got, err := io.ReadAll(oldFile)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(got) != "first" {
+ t.Fatalf("open reader changed after Create replacement: got
%q", got)
+ }
+
+ current, err := memFS.Open(t.Context(), name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer current.Close()
+ got, err = io.ReadAll(current)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(got) != "second" {
+ t.Fatalf("new reader did not see Create replacement: got %q",
got)
+ }
+}
+
+func TestMemFSOpenConcurrentReplacement(t *testing.T) {
+ memFS := icebergio.NewMemFS()
+ name := "mem://bucket/data.parquet"
+ a := bytes.Repeat([]byte{'a'}, 4096)
+ b := bytes.Repeat([]byte{'b'}, 4096)
+ if err := memFS.WriteFile(name, a); err != nil {
+ t.Fatal(err)
+ }
+
+ var wg sync.WaitGroup
+ wg.Go(func() {
+ for i := range 300 {
+ value := a
+ if i%2 != 0 {
+ value = b
+ }
+ if err := memFS.WriteFile(name, value); err != nil {
+ t.Errorf("write: %v", err)
+
+ return
+ }
+ }
+ })
+
+ for range 300 {
+ file, err := memFS.Open(t.Context(), name)
+ if err != nil {
+ t.Errorf("open: %v", err)
+
+ break
+ }
+ data, readErr := io.ReadAll(file)
+ closeErr := file.Close()
+ if readErr != nil || closeErr != nil {
+ t.Errorf("read: %v, close: %v", readErr, closeErr)
+
+ break
+ }
+ if !bytes.Equal(data, a) && !bytes.Equal(data, b) {
+ t.Errorf("file changed while open")
+
+ break
+ }
+ }
+ wg.Wait()
+}
