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

@@ -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)

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
 }

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)

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,
 	}
 }