From 737cc0f33011f43d0ce4a41391164d38ad7b1edb Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 25 Apr 2026 15:15:46 -0700 Subject: [PATCH] parquet_pushdown(M1-C3): add Loader interface and LocalLoader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a small Loader interface so the service can stay oblivious to the storage backend behind a Parquet path: tests inject a LocalLoader backed by os.File, and the daemon will plug in a filer-backed Loader in a later milestone. FileHandle is the parser-friendly view: io.ReaderAt + io.Closer + Size(). LocalLoader rejects missing files and directories up front so service-level error paths see a clear "open failed" rather than a later ReadAt EOF. The Service does not yet consume Loader — that wiring lands in M1-C4 along with the predicate-less Pushdown handler that actually issues loads. --- weed/parquet_pushdown/loader.go | 57 ++++++++++++++++++++++++++++ weed/parquet_pushdown/loader_test.go | 48 +++++++++++++++++++++++ 2 files changed, 105 insertions(+) create mode 100644 weed/parquet_pushdown/loader.go create mode 100644 weed/parquet_pushdown/loader_test.go diff --git a/weed/parquet_pushdown/loader.go b/weed/parquet_pushdown/loader.go new file mode 100644 index 0000000000..a4e96e880a --- /dev/null +++ b/weed/parquet_pushdown/loader.go @@ -0,0 +1,57 @@ +package parquet_pushdown + +import ( + "context" + "fmt" + "io" + "os" +) + +// FileHandle is what the service hands to the parser: a positional +// reader bound to a known size, with a Close so the caller can +// release any underlying resources. +type FileHandle interface { + io.ReaderAt + io.Closer + Size() int64 +} + +// Loader resolves a Parquet file's path to a FileHandle. The split +// between "local file" and "filer-backed" implementations lives here +// so the service stays oblivious to the storage backend, and so +// integration tests can plug in deterministic local files instead of +// spinning up a filer. +type Loader interface { + Open(ctx context.Context, path string) (FileHandle, error) +} + +// LocalLoader treats `path` as a filesystem path. Used by integration +// tests and for local-development smoke runs against a Parquet file +// that is not under filer management. +type LocalLoader struct{} + +func (LocalLoader) Open(_ context.Context, path string) (FileHandle, error) { + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("open %q: %w", path, err) + } + st, err := f.Stat() + if err != nil { + _ = f.Close() + return nil, fmt.Errorf("stat %q: %w", path, err) + } + if st.IsDir() { + _ = f.Close() + return nil, fmt.Errorf("%q is a directory", path) + } + return &localFile{f: f, size: st.Size()}, nil +} + +type localFile struct { + f *os.File + size int64 +} + +func (l *localFile) ReadAt(p []byte, off int64) (int, error) { return l.f.ReadAt(p, off) } +func (l *localFile) Close() error { return l.f.Close() } +func (l *localFile) Size() int64 { return l.size } diff --git a/weed/parquet_pushdown/loader_test.go b/weed/parquet_pushdown/loader_test.go new file mode 100644 index 0000000000..b549017c14 --- /dev/null +++ b/weed/parquet_pushdown/loader_test.go @@ -0,0 +1,48 @@ +package parquet_pushdown + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +func TestLocalLoader_Open(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "a.bin") + want := []byte("hello pushdown") + if err := os.WriteFile(path, want, 0o600); err != nil { + t.Fatalf("write: %v", err) + } + + h, err := LocalLoader{}.Open(context.Background(), path) + if err != nil { + t.Fatalf("open: %v", err) + } + defer h.Close() + + if h.Size() != int64(len(want)) { + t.Errorf("Size = %d, want %d", h.Size(), len(want)) + } + got := make([]byte, len(want)) + n, err := h.ReadAt(got, 0) + if err != nil { + t.Fatalf("ReadAt: %v", err) + } + if n != len(want) || string(got) != string(want) { + t.Errorf("got %q (n=%d), want %q", got[:n], n, want) + } +} + +func TestLocalLoader_RejectsMissing(t *testing.T) { + if _, err := (LocalLoader{}).Open(context.Background(), filepath.Join(t.TempDir(), "missing.bin")); err == nil { + t.Error("expected error opening missing file") + } +} + +func TestLocalLoader_RejectsDirectory(t *testing.T) { + dir := t.TempDir() + if _, err := (LocalLoader{}).Open(context.Background(), dir); err == nil { + t.Error("expected error opening a directory") + } +}