From e1a09416fcd863eef2410791ac676076ba939e0d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 25 Apr 2026 15:18:16 -0700 Subject: [PATCH] parquet_pushdown(M1-C4): file/range-level pushdown for predicate-less requests Wire the parser, cache, and Loader through the Service: - Options gains Loader and Cache (with CacheSize fallback). Defaults to LocalLoader and a footer-cache of defaultFooterCacheSize entries. - Pushdown handles the predicate-less, no-vector path by parsing each DataFile's footer (cached by Identity = (Path, SizeBytes, RecordCount, ETag)) and emitting one FileRange per (data file, row group, projected column) triple. Predicate / vector requests still return Unimplemented with stats. - statsRecorder gains per-request footer cache hit/miss counters that populate PushdownStats.footer_cache_hits / footer_cache_misses. - Column projection matches by ColumnRef.Path. Field-id-only refs return InvalidArgument until M3 (catalog-validated mode) plumbs the Iceberg schema through; the path-hint fallback is enough for the M1 round-trip test. Daemon supplies LocalLoader for now; the filer-backed Loader lands when M2+ needs it. Update the M0 Unimplemented test to send a predicate so it exercises the still-Unimplemented branch (file/range-level pushdown is now implemented for the no-predicate case). --- weed/parquet_pushdown/daemon/server.go | 3 + weed/parquet_pushdown/service.go | 190 ++++++++++++++++++--- weed/parquet_pushdown/service_grpc_test.go | 9 +- weed/parquet_pushdown/stats.go | 28 +-- 4 files changed, 198 insertions(+), 32 deletions(-) diff --git a/weed/parquet_pushdown/daemon/server.go b/weed/parquet_pushdown/daemon/server.go index 4b22d8095..22338c4cc 100644 --- a/weed/parquet_pushdown/daemon/server.go +++ b/weed/parquet_pushdown/daemon/server.go @@ -43,6 +43,9 @@ func Run(cfg Config) error { svc := parquet_pushdown.New(parquet_pushdown.Options{ Version: cfg.Version, TrustMode: cfg.TrustMode, + // M1 uses LocalLoader; the filer-backed Loader lands when the + // catalog-validated path needs to fetch real data files (M2+). + Loader: parquet_pushdown.LocalLoader{}, }) grpcL, localL, err := util.NewIpAndLocalListeners(cfg.IP, cfg.Port, 0) diff --git a/weed/parquet_pushdown/service.go b/weed/parquet_pushdown/service.go index 6f790217a..4703913a2 100644 --- a/weed/parquet_pushdown/service.go +++ b/weed/parquet_pushdown/service.go @@ -3,17 +3,20 @@ // PARQUET_PUSHDOWN_DEV_PLAN.md at the repo root for the surrounding // design and the milestone plan. // -// M0 wires up the gRPC service and request validation; the actual -// pruning logic (parsed-footer cache, row-group pruning, scalar/page -// indexes, deletes, vectors) lands in M1+. +// M1 wires up the parsed-footer cache and file/range-level pushdown: +// for predicate-less, no-vector requests the service returns the +// column-chunk byte ranges of the projected columns. Predicate / +// vector evaluation lands in later milestones. package parquet_pushdown import ( "context" + "fmt" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "github.com/seaweedfs/seaweedfs/weed/parquet_pushdown/footer" pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb" ) @@ -34,17 +37,35 @@ const ( TrustModeConnectorTrusted TrustMode = "connector-trusted" ) -// Options configures a Service. Fields the M0 skeleton does not yet -// consume (catalog client, filer client, predicate engine) will be -// added in their respective milestones. +// defaultFooterCacheSize is the parsed-footer LRU's default capacity +// when Options.Cache is unset. Footer entries are small (a few KiB +// typically; the heaviest field is the per-row-group ColumnChunks +// slice that scales with column count and row group count). 4096 +// covers a few thousand active Parquet files comfortably; can be +// overridden via Options.CacheSize. +const defaultFooterCacheSize = 4096 + +// Options configures a Service. Loader and Cache are pluggable so +// integration tests can wire deterministic local files; the daemon +// supplies a filer-backed Loader and a sized Cache. type Options struct { // Version is reported in PingResponse and stamped on PushdownStats. Version string - // TrustMode determines request-validation strictness. M0 only - // stamps the value into the response; the actual validation - // against the catalog lands in M3. + // TrustMode determines request-validation strictness. The actual + // validation against the catalog lands in M3. TrustMode TrustMode + + // Loader resolves DataFile paths to FileHandles. Defaults to + // LocalLoader{} (filesystem-backed). + Loader Loader + + // Cache is the parsed-footer LRU. When nil, a cache of CacheSize + // (or defaultFooterCacheSize) entries is created. + Cache *footer.Cache + + // CacheSize is the entry cap when Cache is nil. + CacheSize int } // Service implements parquet_pushdown_pb.SeaweedParquetPushdownServer. @@ -54,18 +75,43 @@ type Service struct { version string trustMode TrustMode + loader Loader + cache *footer.Cache } -// New constructs a Service from Options. The caller is responsible -// for registering it on a gRPC server (see weed/parquet_pushdown/daemon). +// New constructs a Service from Options, applying defaults. Returns +// an error if the cache cannot be constructed (e.g. nonsense size). func New(opts Options) *Service { mode := opts.TrustMode if mode == "" { mode = TrustModeCatalogValidated } + loader := opts.Loader + if loader == nil { + loader = LocalLoader{} + } + cache := opts.Cache + if cache == nil { + size := opts.CacheSize + if size <= 0 { + size = defaultFooterCacheSize + } + c, err := footer.NewCache(size) + if err != nil { + // Size is validated above to be > 0; this is a + // programmer-error path the cache implementation should + // never hit. Promoting to panic keeps New non-error and + // callers oblivious to a configuration knob they did not + // touch. + panic(fmt.Errorf("parquet_pushdown: build footer cache: %w", err)) + } + cache = c + } return &Service{ version: opts.Version, trustMode: mode, + loader: loader, + cache: cache, } } @@ -79,21 +125,125 @@ func (s *Service) Ping(_ context.Context, _ *pb.PingRequest) (*pb.PingResponse, } // Pushdown evaluates a request and returns surviving file/row-group/ -// page byte ranges. M0 returns Unimplemented after request-shape -// validation; later milestones replace the body. +// page byte ranges. // -// Unimplemented is attached with a PushdownStats detail so the wire -// format for stats is exercised end-to-end on M0. -func (s *Service) Pushdown(_ context.Context, req *pb.ParquetPushdownRequest) (*pb.ParquetPushdownResponse, error) { +// M1 handles only the predicate-less, no-vector path: parse each +// data file's footer (cached by Identity) and return one FileRange +// per (data file, row group, projected column) triple covering that +// column chunk's byte range. Predicate / vector clauses fall through +// to Unimplemented until later milestones. +func (s *Service) Pushdown(ctx context.Context, req *pb.ParquetPushdownRequest) (*pb.ParquetPushdownResponse, error) { stats := newStats(s.trustMode) if err := validateRequest(req); err != nil { return nil, err } - st := status.New(codes.Unimplemented, "parquet pushdown evaluation lands in M1; M0 only validates request shape") - if withDetails, err := st.WithDetails(stats.toProto()); err == nil { - return nil, withDetails.Err() + if len(req.Predicate) > 0 || req.VectorQuery != nil { + st := status.New(codes.Unimplemented, "predicate / vector evaluation lands in a later milestone; M1 only handles file/range-level pushdown") + if d, err := st.WithDetails(stats.toProto()); err == nil { + return nil, d.Err() + } + return nil, st.Err() } - return nil, st.Err() + + resp := &pb.ParquetPushdownResponse{} + for _, df := range req.DataFiles { + pf, err := s.loadFooter(ctx, df, stats) + if err != nil { + return nil, status.Errorf(codes.Internal, "load footer for %q: %v", df.Path, err) + } + ranges, err := projectColumnRanges(df.Path, pf, req.Columns) + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "project columns for %q: %v", df.Path, err) + } + resp.FileRanges = append(resp.FileRanges, ranges...) + } + resp.Stats = stats.toProto() + return resp, nil +} + +func (s *Service) loadFooter(ctx context.Context, df *pb.DataFileDescriptor, stats *statsRecorder) (*footer.ParsedFooter, error) { + id := footer.Identity{ + Path: df.Path, + SizeBytes: df.SizeBytes, + RecordCount: df.RecordCount, + ETag: df.Etag, + } + if pf, ok := s.cache.Get(id); ok { + stats.recordFooterCacheHit() + return pf, nil + } + stats.recordFooterCacheMiss() + + h, err := s.loader.Open(ctx, df.Path) + if err != nil { + return nil, fmt.Errorf("open: %w", err) + } + defer h.Close() + + pf, err := footer.ParseFromReader(h, h.Size()) + if err != nil { + return nil, fmt.Errorf("parse: %w", err) + } + s.cache.Add(id, pf) + return pf, nil +} + +// projectColumnRanges expands the requested column projection into +// per-row-group byte ranges over the data file. An empty Columns +// list means "every column"; when columns are listed they are +// matched by ColumnRef.Path against the file's leaf-column paths. +// +// Field-id-only projection (Path == "") needs the catalog schema to +// resolve to a leaf path and is therefore unsupported until M3 +// (catalog-validated mode) plumbs the schema through. +func projectColumnRanges(filePath string, pf *footer.ParsedFooter, cols []*pb.ColumnRef) ([]*pb.FileRange, error) { + pathIdx := make(map[string]int, len(pf.ColumnPaths)) + for i, p := range pf.ColumnPaths { + pathIdx[p] = i + } + + indices, err := resolveColumnIndices(pathIdx, pf.NumCols, cols) + if err != nil { + return nil, err + } + + out := make([]*pb.FileRange, 0, len(pf.RowGroups)*len(indices)) + for _, rg := range pf.RowGroups { + for _, idx := range indices { + cc := rg.ColumnChunks[idx] + out = append(out, &pb.FileRange{ + File: filePath, + Offset: cc.Offset, + Length: cc.Length, + }) + } + } + return out, nil +} + +func resolveColumnIndices(pathIdx map[string]int, numCols int, cols []*pb.ColumnRef) ([]int, error) { + if len(cols) == 0 { + out := make([]int, numCols) + for i := range out { + out[i] = i + } + return out, nil + } + out := make([]int, 0, len(cols)) + for _, c := range cols { + if c == nil { + return nil, fmt.Errorf("nil column ref") + } + if c.Path == "" { + return nil, fmt.Errorf("column ref field_id=%d has no path hint; field-id-only resolution requires catalog schema (M3)", c.FieldId) + } + idx, ok := pathIdx[c.Path] + if !ok { + return nil, fmt.Errorf("column %q not in file", c.Path) + } + out = append(out, idx) + } + return out, nil } diff --git a/weed/parquet_pushdown/service_grpc_test.go b/weed/parquet_pushdown/service_grpc_test.go index 467c9c8bb..1a85dbc25 100644 --- a/weed/parquet_pushdown/service_grpc_test.go +++ b/weed/parquet_pushdown/service_grpc_test.go @@ -97,6 +97,10 @@ func TestPing_DefaultsTrustModeToCatalogValidated(t *testing.T) { } } +// As of M1 the Unimplemented path is reserved for predicate / +// vector requests (file/range-level pushdown is implemented). Use a +// predicate-bearing request to exercise the Unimplemented + stats +// detail wire format. func TestPushdown_ReturnsUnimplementedWithStats(t *testing.T) { client, cleanup := startTestServer(t, Options{ Version: "test", @@ -113,7 +117,10 @@ func TestPushdown_ReturnsUnimplementedWithStats(t *testing.T) { DataFiles: []*pb.DataFileDescriptor{ {Path: "s3://b/p.parquet", SizeBytes: 1, RecordCount: 1}, }, - Columns: []*pb.ColumnRef{{FieldId: 1}}, + Columns: []*pb.ColumnRef{{FieldId: 1, Path: "id"}}, + PredicateKind: pb.PredicateKind_PREDICATE_KIND_SUBSTRAIT, + Predicate: []byte("bound"), + SchemaId: 1, } _, err := client.Pushdown(ctx, req) diff --git a/weed/parquet_pushdown/stats.go b/weed/parquet_pushdown/stats.go index b6d8005e8..27471a5e1 100644 --- a/weed/parquet_pushdown/stats.go +++ b/weed/parquet_pushdown/stats.go @@ -7,20 +7,24 @@ import ( ) // statsRecorder accumulates per-request stats and converts to the -// wire form when the handler returns. M0 only records trust mode and -// elapsed server time; later milestones extend it with cache hits, -// pruning counts, and indexes-used / indexes-missing labels. +// wire form when the handler returns. Counts here are scoped to a +// single Pushdown call, not process-wide. type statsRecorder struct { - start time.Time - trustMode TrustMode - indexesUsed []string - indexesMissing []string + start time.Time + trustMode TrustMode + footerCacheHits int64 + footerCacheMisses int64 + indexesUsed []string + indexesMissing []string } func newStats(trustMode TrustMode) *statsRecorder { return &statsRecorder{start: time.Now(), trustMode: trustMode} } +func (r *statsRecorder) recordFooterCacheHit() { r.footerCacheHits++ } +func (r *statsRecorder) recordFooterCacheMiss() { r.footerCacheMisses++ } + func (r *statsRecorder) markIndexUsed(kind string) { r.indexesUsed = append(r.indexesUsed, kind) } @@ -31,9 +35,11 @@ func (r *statsRecorder) markIndexMissing(kind string) { func (r *statsRecorder) toProto() *pb.PushdownStats { return &pb.PushdownStats{ - TrustMode: string(r.trustMode), - ServerTimeMicros: time.Since(r.start).Microseconds(), - IndexesUsed: r.indexesUsed, - IndexesMissing: r.indexesMissing, + TrustMode: string(r.trustMode), + ServerTimeMicros: time.Since(r.start).Microseconds(), + FooterCacheHits: r.footerCacheHits, + FooterCacheMisses: r.footerCacheMisses, + IndexesUsed: r.indexesUsed, + IndexesMissing: r.indexesMissing, } }