docs(parquet-design): rewrite plan code-layout and milestones

Realign with the new catalog-server + volume-server architecture.

Code layout:

- Catalog-side code lives at weed/s3api/iceberg/pushdown/ (the
  existing catalog-server tree). Houses the service entry, request
  validation, footer cache, predicate decoders, planner, registry,
  and Iceberg delete metadata tracking.
- Volume-side code (lands in Phase 3) lives at weed/storage/pushdown/
  with a volume_grpc_pushdown.go entry point, plus index/* readers
  and exec/* (predicate eval, vector compute, bitmap merge).
- Proto stays shared at weed/pb/parquet_pushdown.proto.
- Note: existing M0/M1 code at weed/parquet_pushdown/ moves in
  M2-rebase, the first task of the revised milestones.

Milestones:

- "Already shipped" section records M0/M1 status.
- M2-rebase: relocate package, drop standalone-daemon command and
  daemon/ subpackage, register service on s3api gRPC.
- Phase 1 (catalog candidates): M3 row-group pruning, M4 catalog-
  validated trust mode.
- Phase 2 (catalog index registry): M5 schema/CRUD, M6 Iceberg
  delete metadata, M7 background reconciler.
- Phase 3 (volume execution): M8 bloom, M9 bitmap+btree, M10 page,
  M11 delete bitmaps, M12 vector IVF, M13 hybrid scalar+vector.
- Phase 4 deferred to measurements.
- Connector integrations run in parallel with Phases 1-3.

PARQUET_PUSHDOWN_DEV_PLAN.md

@@ -0,0 +1,237 @@
+# Parquet Pushdown Dev Plan
+
+Companion to [PARQUET_PUSHDOWN_DESIGN.md](./PARQUET_PUSHDOWN_DESIGN.md). The design defines *what* the system does; this doc defines *how it lands in this codebase, in what order, with what milestones*.
+
+## Status
+
+Open questions resolved (see [Decisions Recorded](#decisions-recorded)). M0 is unblocked.
+
+## Goals for the implementation
+
+1. Ship the design's Phase 1 in production-quality form: a pushdown service that, given an Iceberg snapshot resolution, returns column-chunk byte ranges and prunes row groups using parsed-footer statistics.
+2. Establish the package layout, wire protocol, and trust-mode plumbing so later phases (scalar indexes, page-level pruning, deletes, vector search) attach without reshaping the surface.
+3. Avoid breaking any existing read path — pushdown is additive; standard S3 reads keep working.
+
+## What already exists in the codebase
+
+| Component | Path | Reuse |
+|---|---|---|
+| Iceberg-Go client and table/manifest types | `weed/s3api/iceberg/`, `weed/plugin/worker/iceberg/` | Catalog reads, manifest parsing, planning |
+| Parquet reader (`parquet-go/parquet-go v0.28.0`) | `weed/mq/logstore/`, `weed/query/engine/parquet_scanner.go` | Footer parse, ColumnIndex/OffsetIndex |
+| SQL/value type conversion | `weed/query/engine/parquet_scanner.go`, `weed/query/sqltypes/` | Predicate constant typing |
+| Iceberg compaction worker (uses iceberg-go for planning) | `weed/plugin/worker/iceberg/planning_index.go` | Reference for snapshot → file-list resolution |
+| Filer gRPC + protobuf scaffolding | `weed/pb/filer_pb/`, `weed/server/filer_grpc_server.go` | Wire-protocol pattern, auth |
+| S3 IAM / bucket ACL | `weed/s3api/`, `weed/iam/` | Trust enforcement on data files |
+| Substrait Go SDK (already a transitive dep) | `github.com/substrait-io/substrait-go/v7` | Predicate decode in v2 |
+
+The dependencies needed for v1 (`parquet-go`, `iceberg-go`, `RoaringBitmap/roaring`, `substrait-go/v7`) are already in `go.mod` (some as indirect; will become direct).
+
+## Architectural decisions for v1
+
+These are commitments unless an open question below changes them.
+
+- **Residency: catalog server (planning) + volume servers (heavy execution).** Pushdown is *not* a separate daemon. Planning (snapshot resolution, file-level pruning, side-index registry, footer cache, request validation) runs inside the existing catalog-server role — today colocated with the S3 gateway's Iceberg REST handlers in `weed/s3api/iceberg/`. Heavy execution (local index lookup, scalar predicate evaluation against indexes, vector distance + local top-K) runs on the volume server next to the data. The S3 gateway remains the compatibility path for unmodified readers. Rationale: the catalog already knows tables, snapshots, and table-to-files mapping, which is exactly what the planner needs — adding a separate daemon would duplicate that state. Heavy work goes to the volume servers so a single hot table cannot bottleneck cluster-wide planning. A standalone query-worker tier remains an option (design's Phase 4) only if measurements demand it.
+- **Wire protocol: gRPC.** New service definition `parquet_pushdown_pb.ParquetPushdown` in `weed/pb/parquet_pushdown_pb/`. REST/HTTP shim only if a connector requires it. JSON variant deferred.
+- **Trust mode for v1: catalog-validated, with connector-trusted as a developer-only flag.** The default deployment validates every request's `DataFiles` and `Deletes` against the Iceberg catalog at the requested snapshot before serving (full rules in [Trust Model](./PARQUET_PUSHDOWN_DESIGN.md#trust-model-and-catalog-validation)). Connector-trusted is supported but only enabled by an explicit flag for local development and benchmarks; it is not the default and is rejected in production builds. Manifest-signed is not on the roadmap.
+- **Predicate engine for v1: built-in subset evaluator.** The wire format accepts Substrait `ExtendedExpression`, but v1 implements only: comparisons (`=, !=, <, <=, >, >=`), `IN`, `BETWEEN`, `IS NULL`, and boolean `AND/OR/NOT` over them. Anything outside this subset returns `unsupported predicate`, and the connector falls back to a standard scan. Full Substrait evaluation lands in Phase 2.
+- **Side-index physical location:** co-located with the data in a hidden `.index/` directory next to each Parquet file or its parent folder, governed by the same filer ACLs as the data. Path layout matches the design's `<parquet_parent>/.index/<file_name>/<identity>/...` (file-scoped) and `<parquet_parent>/.index/<identity>/...` (folder-scoped) shapes — see [Physical Layout](./PARQUET_PUSHDOWN_DESIGN.md#physical-layout). The hidden-prefix name is configurable for deployments whose readers do not respect dot-prefix conventions.
+- **Parquet library: `parquet-go/parquet-go`** (already in use elsewhere in the repo). Apache `arrow-go/v18/parquet` is *not* introduced.
+- **Bitmap library: `RoaringBitmap/roaring`.** Used for position-delete bitmaps and DV decoding.
+
+## Code layout
+
+Code lives in two trees that match the architecture's two tiers. The proto stays shared.
+
+**Catalog-server side** — registered on the gRPC server hosted by the catalog (today colocated with the S3 gateway's Iceberg REST handlers in `weed/s3api/iceberg/`):
+
+```text
+weed/s3api/iceberg/pushdown/
+  service.go                  # SeaweedParquetPushdown service entry
+  request_validation.go       # request-shape limits, trust-mode enforcement
+  catalog/                    # Phase 1: catalog-validated mode
+    validator.go              # cross-checks request DataFiles against manifest
+  footer/
+    cache.go                  # parsed-footer LRU, keyed by file identity
+    parser.go                 # footer + ColumnIndex/OffsetIndex extraction
+  predicate/
+    substrait_decode.go       # decode Substrait ExtendedExpression
+    iceberg_decode.go         # decode Iceberg Expression JSON
+    eval_subset.go            # v1 built-in evaluator (comparisons + boolean)
+    rowgroup_prune.go         # zone-map pruning for v1
+  registry/                   # Phase 2: side-index registry
+    schema.go                 # registry rows + filer-backed CRUD
+    reconciler.go             # background freshness checker
+  delete/                     # Phase 2: Iceberg delete metadata
+    position.go               # tracks v2 position-delete files + DVs
+    equality.go               # tracks v2 equality-delete files
+  loader.go                   # FileHandle + Loader (filer-backed for production)
+  stats.go                    # PushdownStats accumulation
+```
+
+**Volume-server side** (lands in Phase 3) — registered on the volume server's gRPC surface so index lookup, predicate evaluation, and vector compute run colocated with data:
+
+```text
+weed/server/volume_grpc_pushdown.go   # entry point on the volume server
+weed/storage/pushdown/
+  index/
+    bloom/                    # Phase 3
+    bitmap/                   # Phase 3
+    btree/                    # Phase 3
+    page/                     # Phase 3
+    vector/ivf/               # Phase 3
+  exec/
+    predicate.go              # evaluate v1 predicate subset against indexes
+    vector.go                 # local distance + top-K
+    bitmap_merge.go           # delete-bitmap union / subtraction
+```
+
+**Shared proto** stays as today:
+
+```text
+weed/pb/parquet_pushdown.proto      # request/response messages
+weed/pb/parquet_pushdown_pb/        # generated
+```
+
+Tests live alongside the code. Cross-tier integration tests live in `test/parquet_pushdown/`.
+
+> **Existing code at `weed/parquet_pushdown/`** (M0 + M1) was scaffolded under the prior standalone-daemon decision. A code-move commit relocates the package to `weed/s3api/iceberg/pushdown/`, drops the `weed pushdown` subcommand and the standalone-daemon bootstrap, and rewires the service into the existing s3api gRPC server. That move is sequenced as the first task of M2 work in the revised milestones below.
+
+## Milestones
+
+Milestones are organized by the design's four-phase rollout: catalog candidates, catalog index registry, volume execution, optional workers. M0 and M1 already shipped under the previous standalone-daemon decision; **M2-rebase** is the first task of the revised plan and unblocks everything that follows.
+
+### Already shipped under the prior architecture
+
+Two milestones live at `weed/parquet_pushdown/` and need the rebase below before further work:
+
+- **M0 (shipped):** proto + service skeleton + bufconn integration test.
+- **M1 (shipped):** parsed-footer cache + file/range-level pushdown for predicate-less requests.
+
+### M2-rebase — Move pushdown into the catalog server (1 PR)
+
+- Move `weed/parquet_pushdown/` to `weed/s3api/iceberg/pushdown/`. Update import paths.
+- Drop `weed pushdown` and `weed pushdown.ping`; remove `weed/parquet_pushdown/daemon/`. The standalone-daemon path is retired.
+- Register `SeaweedParquetPushdown` on the s3api gRPC server alongside the existing Iceberg REST handlers, gated by a `-pushdown.enabled` flag (default off until Phase 1 ships).
+- Update the existing M0/M1 tests to exercise the service via the s3api gRPC registration rather than a bufconn-only listener.
+- Acceptance: existing M0/M1 tests pass against the relocated package; `weed s3 -pushdown.enabled` exposes the service.
+
+### Phase 1: Catalog returns candidate files / row groups / ranges
+
+#### M3 — Row-group zone-map pruning (1 PR)
+
+- `predicate/eval_subset.go`: evaluator for the v1 predicate subset.
+- `predicate/rowgroup_prune.go`: read row-group statistics from the parsed footer, drop row groups where the predicate cannot be true, return per-data-file `RowGroupRef` list.
+- `predicate/substrait_decode.go`: decode just enough Substrait to feed the subset evaluator; everything else returns `unsupported`.
+- Acceptance:
+  - Predicate `WHERE timestamp BETWEEN t1 AND t2` against a 5M-row partitioned Iceberg table prunes to the expected row groups.
+  - `unsupported predicate` returns a status the connector can recognize and fall back from.
+
+#### M4 — Catalog-validated trust mode (1 PR; required for Phase 1)
+
+- `catalog/validator.go`: read the Iceberg snapshot via the existing internal Iceberg package (decision 6 below), verify `(Path, SizeBytes, RecordCount, DataSequenceNumber, PartitionSpecId, PartitionValues)` for each `DataFile` and that the `Deletes` list matches the manifest's attached delete files.
+- Configurable per-endpoint: `pushdown.trust=catalog-validated|connector-trusted` (default `catalog-validated`; `connector-trusted` requires an explicit dev-mode flag).
+- Cache validated planning result keyed by `(table, SnapshotId)` to amortize manifest reads across requests in the same snapshot.
+- Acceptance: a tampered request (omitted delete file, wrong sequence number, file not in manifest, wrong partition values) is rejected with `permission_denied`. Cache hit/miss visible in `PushdownStats`.
+
+**End of design Phase 1.** Phase 1 does not ship without M2-rebase + M3 + M4.
+
+### Phase 2: Catalog tracks side-index registry
+
+#### M5 — Index registry schema and CRUD (1 PR)
+
+- `registry/schema.go`: registry rows of shape `(table, file_identity, kind, field_id) -> (.index path, build identity, build_time, healthy)`. Stored in the filer under a system path, separate from the table's bucket prefix.
+- CRUD APIs (gRPC + internal Go) for register, lookup, mark-stale, delete.
+- Planner consults the registry to know which indexes exist; missing entries fall through gracefully and are reported in stats.
+- Acceptance: register, lookup, list, and delete round-trips pass; planner stats show `indexes_used` / `indexes_missing` correctly populated.
+
+#### M6 — Iceberg delete metadata in the registry (1 PR)
+
+- Register position-delete files, equality-delete files, and v3 deletion vectors as registry entries attached to the data file they target.
+- Planner resolves applicable deletes per the design's applicability rules (sequence + partition + file scope, with per-row `file_path` filter for multi-target position-delete files).
+- Acceptance: planner returns the same data-file → delete-files mapping as a brute-force Iceberg manifest scan.
+
+#### M7 — Background reconciler (1 PR)
+
+- Worker that iterates the registry, verifies each `.index/` payload exists at the recorded path with the recorded identity, marks stale rows, and rebuilds when possible.
+- Compaction / snapshot expiration eviction: registry rows whose `file_identity` is no longer referenced by any active snapshot are reclaimable.
+- Acceptance: tampering with an index payload causes the next request to report it missing; reconciler restores the registry on the next pass.
+
+### Phase 3: Volume servers execute local index/vector pushdown
+
+#### M8 — Volume-server bloom-filter index (1–2 PRs)
+
+- Build path: a worker reads Parquet column chunks and writes SplitBlock Bloom Filters to `.index/<file_name>/<identity>/bloom.fid_<id>` next to the data needles. Registry entry written on completion.
+- Read path: volume server gains a `PushdownExec` gRPC service that, given a (path, identity, kind, field_id, predicate), returns a per-row-group bitmap of surviving rows. Catalog plans for predicates touching bloom-able columns include the volume-server endpoint.
+- Acceptance: equality predicate `WHERE user_id = 'x'` against a 10M-row column with no matches reads zero column-chunk bytes after planning.
+
+#### M9 — Volume-server bitmap and B-tree (1 PR each)
+
+- `index/bitmap/` for low/medium-cardinality columns; integrates with `RoaringBitmap`.
+- `index/btree/` for sortable columns; range and point-lookup predicates.
+- Acceptance: ≥10× row-group reduction vs M3's footer-only pruning on the standard test workload.
+
+#### M10 — Volume-server page-level pruning (1 PR)
+
+- `index/page/`: wraps Parquet's `ColumnIndex` + `OffsetIndex` per [Page-Level Index](./PARQUET_PUSHDOWN_DESIGN.md#3-page-level-index); row-range translation across columns; dictionary-page byte-range inclusion.
+- Volume server returns `PageRef` entries with byte ranges for each projected column.
+- Acceptance: predicate selectivity ≥0.99 returns ≤1 page per surviving row group across all projected columns.
+
+#### M11 — Volume-server delete bitmap evaluation (1 PR)
+
+- v2 position-delete bitmap merge per the design's filter-by-`file_path` rule; cached with the registry's identity key.
+- v3 deletion-vector (Puffin `deletion-vector-v1`) reader + cache.
+- v2 equality-delete per-pair bitmap evaluation, accelerated by the volume server's scalar indexes.
+- Acceptance: a reference table with both delete forms returns identical row sets to a Spark scan of the same snapshot.
+
+#### M12 — Volume-server vector index (1 PR)
+
+- Partitioned IVF index per [Filtered-ANN strategy](./PARQUET_PUSHDOWN_DESIGN.md#filtered-ann-strategy); per-partition top-K runs entirely on the volume server. Catalog merges per-partition top-Ks across volume servers.
+- Acceptance: recall ≥0.95 vs brute force at top-20, 10× end-to-end speedup on the standard test set.
+
+#### M13 — Hybrid scalar+vector pushdown (1 PR)
+
+- Planner selects partitioned-IVF, post-filter+overscan, or filtered-HNSW per the design's strategy section based on selectivity estimate.
+- Acceptance: hybrid query that combines `tenant_id = 'x' AND timestamp >= t` with `ORDER BY embedding <-> q LIMIT 20` returns recall-correct results in <2× the cost of the lighter clause alone.
+
+**End of design Phase 3.**
+
+### Phase 4: Optional standalone query workers
+
+Built only if profiling shows planning, hot-path coordination, or cross-volume merges become a bottleneck. Out of scope until measurements demand it.
+
+### Connector integrations (parallel with Phases 1–3)
+
+These run in parallel with Phases 1–3 once the catalog API stabilizes (post-M4). Each consumer reads byte ranges directly via S3 (compatibility path) or dispatches to volume-server endpoints (hot path) per request.
+
+- DuckDB extension as PoC; consumes the gRPC API directly.
+- Trino connector pushdown.
+- Spark DataSource / Iceberg integration.
+
+## Test strategy
+
+- **Unit tests:** every package, with parquet-go in-memory fixtures.
+- **Integration tests:** `test/parquet_pushdown/` — spins up filer + a small Iceberg catalog (reuse `weed/s3api/iceberg/` infrastructure), generates Parquet data with `parquet-go` writers, compares pushdown results to a brute-force scan.
+- **Correctness oracle:** every milestone validates that pushdown output is a *strict subset* of a brute-force Iceberg+Parquet scan with the same predicate, and that the projected-column byte ranges decode to the same rows. The oracle is the safety net that lets us refactor index internals freely.
+- **Performance suite:** runs in CI weekly, not on every PR. Tracks footer-cache hit rate, row-group reduction, page reduction, end-to-end query latency vs a baseline standard scan.
+
+## Decisions Recorded
+
+These were the open questions before M0. All have been resolved; recording for traceability and to anchor future amendments.
+
+1. **Residency:** catalog server (planning, registry, footer cache) + volume servers (local index lookup, predicate eval, vector compute). Standalone daemons are *not* introduced; query workers as an extra tier remain a future option contingent on measurements. (See [Architectural decisions](#architectural-decisions-for-v1).)
+2. **Service surface:** new `weed/pb/parquet_pushdown_pb/` package; not extending `filer_pb`.
+3. **Trust mode:** catalog-validated is the default and required for Phase 1 shipping; connector-trusted is a dev-only flag rejected in production builds.
+4. **Side-index path:** co-located in a hidden `.index/` directory next to the Parquet data, under the same parent folder. File-scoped: `<parquet_parent>/.index/<file_name>/<identity>/...`. Folder-scoped: `<parquet_parent>/.index/<identity>/...`. Hidden-prefix name configurable for non-dot-aware readers.
+5. **Predicate subset for v1:** `=`, `!=`, `<`, `<=`, `>`, `>=`, `IN`, `BETWEEN`, `IS NULL`, boolean `AND/OR/NOT`. `LIKE` deferred to the inverted-index milestone. Wire format is Substrait `ExtendedExpression` from M0.
+6. **Iceberg snapshot integration:** route through the existing internal Iceberg packages (`weed/s3api/iceberg/`, `weed/plugin/worker/iceberg/`); do not add a parallel `iceberg-go` integration in `weed/parquet_pushdown/`.
+7. **Auth:** reuse the filer's existing auth. The request principal must have read access to every `DataFile` listed; the daemon enforces this by re-using the filer's authorization on every blob fetch.
+8. **Test corpus:** generated from scratch in `test/parquet_pushdown/fixtures/`, parameterized over (with-position-deletes, with-equality-deletes, with-DVs, partitioned/unpartitioned).
+9. **Phase 1 done bar:** passing CI tests + an I/O-reduction benchmark on a reference query. Connector PoC slips to M12.
+10. **Performance regression budget for v1:** ≤10% added latency vs a direct ranged GET on a 10 MiB single-file scan in the worst case (predicate prunes nothing). The threshold is exposed as a configuration knob (`pushdown.max_overhead_pct`, default `10`) so deployments can tighten or loosen it without code changes.
+
+## Risk register
+
+- **Iceberg-Go maturity for v3 deletion vectors / Puffin.** If `iceberg-go v0.5.0` does not expose Puffin DV blobs, M8 needs a vendored Puffin parser. Spike during M0 to verify.
+- **parquet-go ColumnIndex/OffsetIndex completeness.** M6 depends on full ColumnIndex/OffsetIndex support. If the library returns nil for some files, fall back to row-group-only pruning per the design.
+- **Catalog read latency.** M3's per-request manifest read must be cacheable per snapshot, otherwise the catalog-validated mode adds >100ms per request. Cache eviction strategy is part of M3's spec.
+- **Trust boundary leakage.** A pushdown response must not return data references that a direct read would deny. The auth check (open question 7) is load-bearing; integration tests in M0 must include negative cases.

weed/command/command.go

@@ -37,6 +37,8 @@ var Commands = []*Command{
 	cmdMqAgent,
 	cmdMqBroker,
 	cmdMqKafkaGateway,
+	cmdPushdown,
+	cmdPushdownPing,
 	cmdS3,
 	cmdScaffold,
 	cmdServer,

weed/command/pushdown.go

@@ -0,0 +1,71 @@
+package command
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/parquet_pushdown"
+	"github.com/seaweedfs/seaweedfs/weed/parquet_pushdown/daemon"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/version"
+)
+
+var (
+	pushdownOptions PushdownOptions
+)
+
+type PushdownOptions struct {
+	filer               *string
+	ip                  *string
+	port                *int
+	trustMode           *string
+	connectorTrustedAck *bool
+}
+
+func init() {
+	cmdPushdown.Run = runPushdown // break init cycle
+	pushdownOptions.filer = cmdPushdown.Flag.String("filer", "localhost:8888", "comma-separated filer server addresses for high availability")
+	pushdownOptions.ip = cmdPushdown.Flag.String("ip", util.DetectedHostAddress(), "pushdown daemon gRPC listen ip address")
+	pushdownOptions.port = cmdPushdown.Flag.Int("port", 18888, "pushdown daemon gRPC listen port")
+	pushdownOptions.trustMode = cmdPushdown.Flag.String("trust", string(parquet_pushdown.TrustModeCatalogValidated), "trust mode: catalog-validated (default) or connector-trusted (dev only)")
+	pushdownOptions.connectorTrustedAck = cmdPushdown.Flag.Bool("dev.connector_trusted_ack", false, "explicitly acknowledge running with -trust=connector-trusted; required for that mode")
+}
+
+var cmdPushdown = &Command{
+	UsageLine: "pushdown [-port=18888] [-filer=<ip:port>[,<ip:port>]...] [-trust=catalog-validated]",
+	Short:     "<WIP> start the SeaweedFS Parquet pushdown daemon (M0 skeleton)",
+	Long: `start the SeaweedFS Parquet pushdown daemon.
+
+	The pushdown daemon is a standalone gRPC service that accelerates Parquet/
+	Iceberg scans by pruning files, row groups, and pages before the engine
+	reads data. See PARQUET_PUSHDOWN_DESIGN.md and PARQUET_PUSHDOWN_DEV_PLAN.md
+	at the repo root for the full design and milestone plan.
+
+	M0 ships request validation and the gRPC surface only; pruning logic
+	lands in M1+.
+
+	Trust modes:
+	  catalog-validated (default)
+	      Server validates each request's DataFiles and Deletes against the
+	      Iceberg catalog at the requested snapshot before serving.
+	  connector-trusted
+	      Skip catalog validation. Developer-only; production builds must
+	      not enable. Requires the explicit -dev.connector_trusted_ack flag.
+
+	Multiple filer addresses can be specified for high availability,
+	separated by commas.`,
+}
+
+func runPushdown(cmd *Command, args []string) bool {
+	cfg := daemon.Config{
+		IP:                  *pushdownOptions.ip,
+		Port:                *pushdownOptions.port,
+		FilerAddresses:      pb.ServerAddresses(*pushdownOptions.filer).ToAddresses(),
+		TrustMode:           parquet_pushdown.TrustMode(*pushdownOptions.trustMode),
+		ConnectorTrustedAck: *pushdownOptions.connectorTrustedAck,
+		Version:             version.Version(),
+	}
+	if err := daemon.Run(cfg); err != nil {
+		glog.Fatalf("pushdown daemon: %v", err)
+	}
+	return true
+}

weed/command/pushdown_ping.go

@@ -0,0 +1,71 @@
+package command
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+var (
+	pushdownPingOptions PushdownPingOptions
+)
+
+type PushdownPingOptions struct {
+	target  *string
+	timeout *time.Duration
+	useTLS  *bool
+}
+
+func init() {
+	cmdPushdownPing.Run = runPushdownPing
+	pushdownPingOptions.target = cmdPushdownPing.Flag.String("server", "localhost:18888", "pushdown daemon gRPC address")
+	pushdownPingOptions.timeout = cmdPushdownPing.Flag.Duration("timeout", 5*time.Second, "RPC deadline")
+	pushdownPingOptions.useTLS = cmdPushdownPing.Flag.Bool("tls", false, "dial with TLS using grpc.client config")
+}
+
+var cmdPushdownPing = &Command{
+	UsageLine: "pushdown.ping [-server=<ip:port>]",
+	Short:     "<WIP> ping a running pushdown daemon (M0 smoke test)",
+	Long: `ping a running pushdown daemon.
+
+	Calls the SeaweedParquetPushdown.Ping RPC and prints the daemon's
+	reported version and trust mode. Used as a deployment smoke test
+	and by the M0 integration test.`,
+}
+
+func runPushdownPing(cmd *Command, args []string) bool {
+	util.LoadSecurityConfiguration()
+
+	dialOpt := grpc.WithTransportCredentials(insecure.NewCredentials())
+	if *pushdownPingOptions.useTLS {
+		dialOpt = security.LoadClientTLS(util.GetViper(), "grpc.client")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), *pushdownPingOptions.timeout)
+	defer cancel()
+
+	conn, err := grpc.NewClient(*pushdownPingOptions.target, dialOpt)
+	if err != nil {
+		glog.Errorf("dial %s: %v", *pushdownPingOptions.target, err)
+		return false
+	}
+	defer conn.Close()
+
+	resp, err := pb.NewSeaweedParquetPushdownClient(conn).Ping(ctx, &pb.PingRequest{})
+	if err != nil {
+		glog.Errorf("ping %s: %v", *pushdownPingOptions.target, err)
+		return false
+	}
+
+	fmt.Fprintf(os.Stdout, "ok: server=%s version=%q trust=%q\n", *pushdownPingOptions.target, resp.Version, resp.TrustMode)
+	return true
+}

weed/parquet_pushdown/daemon/config.go

@@ -0,0 +1,66 @@
+// Package daemon hosts the standalone `weed pushdown` process.
+// It owns gRPC server bootstrap, the filer client used by later
+// milestones to read Parquet data and side-index blobs, and the
+// process-level config plumbing.
+package daemon
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/parquet_pushdown"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// Config controls how the daemon listens, talks to the filer, and
+// validates incoming requests. Caller (the `weed pushdown` command)
+// builds this from CLI flags.
+type Config struct {
+	// IP is the bind address for the gRPC listener. Empty defaults to
+	// the auto-detected host address.
+	IP string
+
+	// Port is the gRPC listener port.
+	Port int
+
+	// FilerAddresses is the comma-resolved list of filer endpoints
+	// the daemon talks to for data reads and (in M3) catalog access.
+	// Required: at least one entry.
+	FilerAddresses []pb.ServerAddress
+
+	// TrustMode is the request-validation strictness. Default is
+	// "catalog-validated"; "connector-trusted" is a developer-only
+	// mode that the daemon refuses to accept in production builds
+	// (see ConnectorTrustedAck).
+	TrustMode parquet_pushdown.TrustMode
+
+	// ConnectorTrustedAck is the explicit acknowledgement required to
+	// run with TrustMode == TrustModeConnectorTrusted. The flag exists
+	// to make the dev-only mode awkward to enable by accident.
+	ConnectorTrustedAck bool
+
+	// Version is reported in PingResponse / PushdownStats. Caller
+	// usually fills this with weed/util/version.Version().
+	Version string
+}
+
+// Validate checks the config before the daemon starts.
+func (c *Config) Validate() error {
+	if c.Port <= 0 || c.Port > 65535 {
+		return fmt.Errorf("port %d is out of range", c.Port)
+	}
+	if len(c.FilerAddresses) == 0 {
+		return errors.New("at least one filer address is required")
+	}
+	switch c.TrustMode {
+	case "", parquet_pushdown.TrustModeCatalogValidated:
+		c.TrustMode = parquet_pushdown.TrustModeCatalogValidated
+	case parquet_pushdown.TrustModeConnectorTrusted:
+		if !c.ConnectorTrustedAck {
+			return errors.New("connector-trusted is a developer-only trust mode; pass -dev.connector_trusted_ack to enable")
+		}
+	default:
+		return fmt.Errorf("unknown trust mode %q", c.TrustMode)
+	}
+	return nil
+}

weed/parquet_pushdown/daemon/config_test.go

@@ -0,0 +1,76 @@
+package daemon
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/parquet_pushdown"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+func baseConfig() Config {
+	return Config{
+		IP:             "127.0.0.1",
+		Port:           18888,
+		FilerAddresses: []pb.ServerAddress{"localhost:8888"},
+		Version:        "test",
+	}
+}
+
+func TestConfig_ValidateOK(t *testing.T) {
+	c := baseConfig()
+	if err := c.Validate(); err != nil {
+		t.Fatalf("expected ok, got %v", err)
+	}
+	if c.TrustMode != parquet_pushdown.TrustModeCatalogValidated {
+		t.Errorf("default trust mode = %q, want %q", c.TrustMode, parquet_pushdown.TrustModeCatalogValidated)
+	}
+}
+
+func TestConfig_RejectsBadPort(t *testing.T) {
+	c := baseConfig()
+	c.Port = 0
+	if err := c.Validate(); err == nil {
+		t.Fatal("expected error for port=0")
+	}
+	c.Port = 70000
+	if err := c.Validate(); err == nil {
+		t.Fatal("expected error for port>65535")
+	}
+}
+
+func TestConfig_RequiresFiler(t *testing.T) {
+	c := baseConfig()
+	c.FilerAddresses = nil
+	if err := c.Validate(); err == nil {
+		t.Fatal("expected error when filer addresses empty")
+	}
+}
+
+// Connector-trusted is dev-only and must require an explicit
+// acknowledgement flag. Operators must not be able to flip it on by
+// passing -trust=connector-trusted alone.
+func TestConfig_ConnectorTrustedRequiresAck(t *testing.T) {
+	c := baseConfig()
+	c.TrustMode = parquet_pushdown.TrustModeConnectorTrusted
+	err := c.Validate()
+	if err == nil {
+		t.Fatal("expected error when connector-trusted lacks ack")
+	}
+	if !strings.Contains(err.Error(), "connector_trusted_ack") {
+		t.Errorf("error %q should mention the ack flag", err.Error())
+	}
+
+	c.ConnectorTrustedAck = true
+	if err := c.Validate(); err != nil {
+		t.Fatalf("connector-trusted with ack should pass, got %v", err)
+	}
+}
+
+func TestConfig_RejectsUnknownTrustMode(t *testing.T) {
+	c := baseConfig()
+	c.TrustMode = parquet_pushdown.TrustMode("nonsense")
+	if err := c.Validate(); err == nil {
+		t.Fatal("expected error for unknown trust mode")
+	}
+}

weed/parquet_pushdown/daemon/filer_client.go

@@ -0,0 +1,48 @@
+package daemon
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// filerClient is a thin handle around the daemon's filer endpoints.
+// M0 only uses it to confirm filer reachability at startup; later
+// milestones add read paths for Parquet data and side-index blobs.
+type filerClient struct {
+	addresses      []pb.ServerAddress
+	grpcDialOption grpc.DialOption
+}
+
+func newFilerClient(addresses []pb.ServerAddress, grpcDialOption grpc.DialOption) *filerClient {
+	return &filerClient{addresses: addresses, grpcDialOption: grpcDialOption}
+}
+
+// waitUntilReachable blocks until the daemon can complete a
+// GetFilerConfiguration RPC against any of its configured filer
+// addresses, or the context is cancelled. Mirrors the iam command's
+// startup-time filer probe.
+func (c *filerClient) waitUntilReachable(ctx context.Context) error {
+	for {
+		err := pb.WithOneOfGrpcFilerClients(false, c.addresses, c.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+			_, getErr := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{})
+			return getErr
+		})
+		if err == nil {
+			glog.V(0).Infof("pushdown daemon connected to filers %v", c.addresses)
+			return nil
+		}
+		glog.V(0).Infof("pushdown daemon waiting for filers %v: %v", c.addresses, err)
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("filer reachability wait cancelled: %w", ctx.Err())
+		case <-time.After(time.Second):
+		}
+	}
+}

weed/parquet_pushdown/daemon/server.go

@@ -0,0 +1,80 @@
+package daemon
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/parquet_pushdown"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	pbpushdown "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/grace"
+)
+
+// Run starts the pushdown daemon. It blocks until the gRPC server
+// returns (graceful shutdown on SIGTERM, or error from Serve). Always
+// validates the config first.
+func Run(cfg Config) error {
+	if err := cfg.Validate(); err != nil {
+		return fmt.Errorf("pushdown config: %w", err)
+	}
+
+	util.LoadSecurityConfiguration()
+	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+
+	// Probe the filer until reachable so we fail fast on bad config
+	// and emit a clear "waiting for filers" log otherwise. Bound the
+	// probe so the daemon doesn't sit forever if the operator hands
+	// in an unreachable filer.
+	probeCtx, probeCancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer probeCancel()
+	fc := newFilerClient(cfg.FilerAddresses, grpcDialOption)
+	if err := fc.waitUntilReachable(probeCtx); err != nil {
+		return fmt.Errorf("filer probe failed: %w", err)
+	}
+
+	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)
+	if err != nil {
+		return fmt.Errorf("listen on grpc port %d: %w", cfg.Port, err)
+	}
+
+	grpcS := pb.NewGrpcServer()
+	pbpushdown.RegisterSeaweedParquetPushdownServer(grpcS, svc)
+	reflection.Register(grpcS)
+
+	grace.OnInterrupt(grpcS.GracefulStop)
+
+	if localL != nil {
+		localGrpcS := pb.NewGrpcServer()
+		pbpushdown.RegisterSeaweedParquetPushdownServer(localGrpcS, svc)
+		reflection.Register(localGrpcS)
+		grace.OnInterrupt(localGrpcS.GracefulStop)
+		go serve(localGrpcS, localL, "pushdown localhost")
+	}
+
+	glog.V(0).Infof("pushdown daemon listening on %s:%d (trust=%s)", cfg.IP, cfg.Port, cfg.TrustMode)
+	return serve(grpcS, grpcL, "pushdown")
+}
+
+func serve(s *grpc.Server, l net.Listener, label string) error {
+	if err := s.Serve(l); err != nil && err != grpc.ErrServerStopped {
+		glog.Errorf("%s server serve error: %v", label, err)
+		return err
+	}
+	return nil
+}

weed/parquet_pushdown/footer/cache.go

@@ -0,0 +1,99 @@
+package footer
+
+import (
+	"errors"
+	"fmt"
+	"sync/atomic"
+
+	lru "github.com/hashicorp/golang-lru"
+)
+
+// Identity uniquely names a Parquet file's content per the design's
+// Index Consistency rules. The cache key is the same identity used
+// to invalidate side indexes when a file's content changes.
+//
+// Iceberg manifest fields are preferred when available (Path,
+// SizeBytes, RecordCount + ETag) because they pin the file by both
+// its bytes and its declared row count — defending against the
+// pathological case where a write produces a file with the same
+// length but different content. ETag fills in for raw-S3 access
+// without an Iceberg manifest. Mtime is intentionally not part of
+// identity (see PARQUET_PUSHDOWN_DESIGN.md "Index Consistency").
+type Identity struct {
+	Path        string
+	SizeBytes   int64
+	RecordCount int64 // 0 when unknown (non-Iceberg-managed files)
+	ETag        string
+}
+
+// Cache is a concurrent LRU of parsed footers keyed by Identity.
+// Capacity is in entries, not bytes; Phase 1 footers are small
+// enough that entry-count budgeting is easier to reason about.
+type Cache struct {
+	c      *lru.Cache
+	hits   atomic.Int64
+	misses atomic.Int64
+}
+
+// NewCache returns a Cache holding up to size entries. Returns an
+// error when size is non-positive (a zero-size LRU is too easy to
+// misconfigure to silently disable caching).
+func NewCache(size int) (*Cache, error) {
+	if size <= 0 {
+		return nil, fmt.Errorf("cache size must be positive, got %d", size)
+	}
+	c, err := lru.New(size)
+	if err != nil {
+		return nil, fmt.Errorf("new lru: %w", err)
+	}
+	return &Cache{c: c}, nil
+}
+
+// Get returns the cached ParsedFooter for id, or nil + false on miss.
+func (c *Cache) Get(id Identity) (*ParsedFooter, bool) {
+	if v, ok := c.c.Get(id); ok {
+		c.hits.Add(1)
+		return v.(*ParsedFooter), true
+	}
+	c.misses.Add(1)
+	return nil, false
+}
+
+// Add stores pf under id. Existing entries are replaced.
+func (c *Cache) Add(id Identity, pf *ParsedFooter) {
+	c.c.Add(id, pf)
+}
+
+// Stats returns cumulative hit and miss counters.
+func (c *Cache) Stats() (hits, misses int64) {
+	return c.hits.Load(), c.misses.Load()
+}
+
+// Len returns the current entry count.
+func (c *Cache) Len() int {
+	return c.c.Len()
+}
+
+// GetOrLoad atomically returns a cached entry or invokes load to
+// produce one and inserts it. Two concurrent loads for the same id
+// may both call load — the cache is not a singleflight; that
+// optimization is deferred until profiling shows it matters. The
+// returned pointer is the one stored in the cache so callers must
+// treat it as read-only.
+func (c *Cache) GetOrLoad(id Identity, load func() (*ParsedFooter, error)) (*ParsedFooter, error) {
+	if pf, ok := c.Get(id); ok {
+		return pf, nil
+	}
+	if load == nil {
+		return nil, errors.New("nil load func")
+	}
+	pf, err := load()
+	if err != nil {
+		return nil, err
+	}
+	if pf == nil {
+		return nil, errors.New("load returned nil ParsedFooter")
+	}
+	c.Add(id, pf)
+	return pf, nil
+}

weed/parquet_pushdown/footer/cache_test.go

@@ -0,0 +1,142 @@
+package footer
+
+import (
+	"errors"
+	"sync"
+	"testing"
+)
+
+func TestCache_NewSizeRejectsZero(t *testing.T) {
+	if _, err := NewCache(0); err == nil {
+		t.Error("size=0 should be rejected")
+	}
+	if _, err := NewCache(-1); err == nil {
+		t.Error("negative size should be rejected")
+	}
+}
+
+func TestCache_HitsAndMissesTracked(t *testing.T) {
+	c, err := NewCache(8)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	id := Identity{Path: "a", SizeBytes: 1}
+	pf := &ParsedFooter{NumRows: 10}
+
+	if _, ok := c.Get(id); ok {
+		t.Fatal("first Get should miss")
+	}
+	c.Add(id, pf)
+	if got, ok := c.Get(id); !ok || got != pf {
+		t.Fatalf("expected cached pointer, got=%v ok=%v", got, ok)
+	}
+	hits, misses := c.Stats()
+	if hits != 1 || misses != 1 {
+		t.Errorf("hits=%d misses=%d, want 1,1", hits, misses)
+	}
+}
+
+// Two distinct identities must not alias. Same path, different size
+// is still a different file from the cache's POV.
+func TestCache_IdentityIsCompositeKey(t *testing.T) {
+	c, err := NewCache(8)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	a := Identity{Path: "x", SizeBytes: 100, RecordCount: 1}
+	b := Identity{Path: "x", SizeBytes: 200, RecordCount: 1}
+	c.Add(a, &ParsedFooter{NumRows: 1})
+	c.Add(b, &ParsedFooter{NumRows: 2})
+
+	gotA, _ := c.Get(a)
+	gotB, _ := c.Get(b)
+	if gotA == gotB {
+		t.Errorf("entries with different sizes must not alias")
+	}
+	if gotA.NumRows != 1 || gotB.NumRows != 2 {
+		t.Errorf("entries swapped: gotA=%d gotB=%d", gotA.NumRows, gotB.NumRows)
+	}
+}
+
+// Eviction is the LRU's job; the test just verifies the cache obeys
+// its size cap and doesn't grow without bound.
+func TestCache_EvictsAboveCap(t *testing.T) {
+	c, err := NewCache(2)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	c.Add(Identity{Path: "a", SizeBytes: 1}, &ParsedFooter{NumRows: 1})
+	c.Add(Identity{Path: "b", SizeBytes: 1}, &ParsedFooter{NumRows: 2})
+	c.Add(Identity{Path: "c", SizeBytes: 1}, &ParsedFooter{NumRows: 3})
+
+	if got := c.Len(); got != 2 {
+		t.Fatalf("Len = %d, want 2 after eviction", got)
+	}
+}
+
+func TestCache_GetOrLoadCachesAndDedupes(t *testing.T) {
+	c, err := NewCache(8)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	id := Identity{Path: "a", SizeBytes: 1}
+
+	calls := 0
+	load := func() (*ParsedFooter, error) {
+		calls++
+		return &ParsedFooter{NumRows: 42}, nil
+	}
+
+	first, err := c.GetOrLoad(id, load)
+	if err != nil {
+		t.Fatalf("first GetOrLoad: %v", err)
+	}
+	second, err := c.GetOrLoad(id, load)
+	if err != nil {
+		t.Fatalf("second GetOrLoad: %v", err)
+	}
+	if first != second {
+		t.Errorf("second GetOrLoad returned different pointer")
+	}
+	if calls != 1 {
+		t.Errorf("load called %d times, want 1", calls)
+	}
+}
+
+func TestCache_GetOrLoadPropagatesError(t *testing.T) {
+	c, err := NewCache(8)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	wantErr := errors.New("io boom")
+	_, err = c.GetOrLoad(Identity{Path: "a"}, func() (*ParsedFooter, error) { return nil, wantErr })
+	if !errors.Is(err, wantErr) {
+		t.Errorf("err = %v, want %v", err, wantErr)
+	}
+	if c.Len() != 0 {
+		t.Errorf("failed load should not insert; Len=%d", c.Len())
+	}
+}
+
+// Parallel reads on the same key should all observe a hit after the
+// first writer wins, and the cache must not race-detect.
+func TestCache_ConcurrentSafe(t *testing.T) {
+	c, err := NewCache(64)
+	if err != nil {
+		t.Fatalf("new: %v", err)
+	}
+	id := Identity{Path: "a", SizeBytes: 1}
+	c.Add(id, &ParsedFooter{NumRows: 1})
+
+	var wg sync.WaitGroup
+	for i := 0; i < 32; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for k := 0; k < 256; k++ {
+				_, _ = c.Get(id)
+			}
+		}()
+	}
+	wg.Wait()
+}

weed/parquet_pushdown/footer/parser.go

@@ -0,0 +1,139 @@
+// Package footer holds the parsed-footer cache and the parser that
+// fills it. Phase 1 of the dev plan: parsed-footer cache + file/range-
+// level pushdown. Page-level (ColumnIndex/OffsetIndex) consumption
+// lands in M6; for M1 we capture the column-chunk byte ranges that
+// file/range-level pushdown needs.
+package footer
+
+import (
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// ParsedFooter is the daemon's compact, cache-friendly view of a
+// Parquet file's footer. It carries only the fields M1 actually
+// consumes; later milestones extend it (ColumnIndex / OffsetIndex,
+// statistics, bloom-filter offsets).
+type ParsedFooter struct {
+	NumRows   int64
+	NumCols   int
+	NumGroups int
+
+	// ColumnPaths is the dotted path-in-schema for each leaf column,
+	// in the canonical order parquet-go assigns to the file.
+	ColumnPaths []string
+
+	// RowGroups[i] holds the byte ranges and row count for row group i.
+	RowGroups []RowGroupRanges
+}
+
+// RowGroupRanges enumerates the column chunks of one row group, in
+// the canonical column order.
+type RowGroupRanges struct {
+	NumRows      int64
+	ColumnChunks []ColumnChunkRange
+}
+
+// ColumnChunkRange is the byte range to read for one column chunk in
+// one row group, including the dictionary page region when present.
+//
+// Iceberg-style page-level pruning narrows reads further; M1 returns
+// the full column-chunk byte range. Including the dictionary-page
+// prefix when DictionaryPageOffset > 0 follows the rule from the
+// design's Page-Level Index section: the dictionary page is not in
+// OffsetIndex but must be fetched whenever any data page is read.
+type ColumnChunkRange struct {
+	Path          []string
+	Offset        int64
+	Length        int64
+	NumValues     int64
+	TotalCompSize int64
+}
+
+// ParseFromReader reads only the footer of a Parquet file and
+// returns the projection used by file/range-level pushdown.
+//
+// SkipPageIndex is set: M1 does not need ColumnIndex/OffsetIndex,
+// and skipping them avoids two extra reads per parse for files that
+// have page indexes written. M6 will turn this back on for the page-
+// level pruning milestone.
+func ParseFromReader(r io.ReaderAt, size int64) (*ParsedFooter, error) {
+	if r == nil {
+		return nil, errors.New("nil reader")
+	}
+	if size <= 0 {
+		return nil, fmt.Errorf("invalid size %d", size)
+	}
+
+	pf, err := parquet.OpenFile(r, size, parquet.SkipPageIndex(true), parquet.SkipBloomFilters(true))
+	if err != nil {
+		return nil, fmt.Errorf("open parquet footer: %w", err)
+	}
+
+	md := pf.Metadata()
+	out := &ParsedFooter{
+		NumRows:   md.NumRows,
+		NumGroups: len(md.RowGroups),
+		RowGroups: make([]RowGroupRanges, len(md.RowGroups)),
+	}
+
+	// All row groups share the same leaf-column layout; sample row
+	// group 0 to get column paths and column count.
+	if len(md.RowGroups) > 0 {
+		out.NumCols = len(md.RowGroups[0].Columns)
+		out.ColumnPaths = make([]string, out.NumCols)
+		for j, c := range md.RowGroups[0].Columns {
+			out.ColumnPaths[j] = joinPath(c.MetaData.PathInSchema)
+		}
+	}
+
+	for i, rg := range md.RowGroups {
+		out.RowGroups[i].NumRows = rg.NumRows
+		out.RowGroups[i].ColumnChunks = make([]ColumnChunkRange, len(rg.Columns))
+		for j, c := range rg.Columns {
+			start, length := columnChunkBytes(&c)
+			out.RowGroups[i].ColumnChunks[j] = ColumnChunkRange{
+				Path:          c.MetaData.PathInSchema,
+				Offset:        start,
+				Length:        length,
+				NumValues:     c.MetaData.NumValues,
+				TotalCompSize: c.MetaData.TotalCompressedSize,
+			}
+		}
+	}
+	return out, nil
+}
+
+// columnChunkBytes returns the (offset, length) of the contiguous
+// byte region that a reader needs to fetch to consume this column
+// chunk: the dictionary page (when present) plus all data pages.
+//
+// parquet-go exposes DictionaryPageOffset as the start of the
+// dictionary page when one exists, otherwise zero, and DataPageOffset
+// as the start of the first data page. The end of the chunk is
+// `start + TotalCompressedSize`.
+func columnChunkBytes(c *format.ColumnChunk) (int64, int64) {
+	start := c.MetaData.DataPageOffset
+	if dict := c.MetaData.DictionaryPageOffset; dict > 0 && dict < start {
+		start = dict
+	}
+	return start, c.MetaData.TotalCompressedSize
+}
+
+func joinPath(parts []string) string {
+	switch len(parts) {
+	case 0:
+		return ""
+	case 1:
+		return parts[0]
+	}
+	out := parts[0]
+	for _, p := range parts[1:] {
+		out += "." + p
+	}
+	return out
+}

weed/parquet_pushdown/footer/parser_test.go

@@ -0,0 +1,129 @@
+package footer
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/parquet-go/parquet-go"
+)
+
+type sampleRow struct {
+	ID    int64
+	Name  string
+	Value float64
+}
+
+// writeFixture builds a small Parquet buffer with two row groups so
+// the parser exercises both row-group iteration and per-column-chunk
+// byte ranges. RowGroupTargetSize is set tight enough that the writer
+// flushes mid-input.
+func writeFixture(t *testing.T, rows []sampleRow) []byte {
+	t.Helper()
+
+	schema := parquet.SchemaOf(sampleRow{})
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[sampleRow](&buf, schema)
+	if _, err := w.Write(rows); err != nil {
+		t.Fatalf("write rows: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+	return buf.Bytes()
+}
+
+func TestParseFromReader_ColumnChunksAndCounts(t *testing.T) {
+	rows := make([]sampleRow, 1000)
+	for i := range rows {
+		rows[i] = sampleRow{ID: int64(i), Name: "row", Value: float64(i)}
+	}
+	data := writeFixture(t, rows)
+
+	pf, err := ParseFromReader(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if pf.NumRows != int64(len(rows)) {
+		t.Errorf("NumRows = %d, want %d", pf.NumRows, len(rows))
+	}
+	if pf.NumCols != 3 {
+		t.Errorf("NumCols = %d, want 3", pf.NumCols)
+	}
+	wantCols := []string{"ID", "Name", "Value"}
+	if len(pf.ColumnPaths) != len(wantCols) {
+		t.Fatalf("ColumnPaths len %d, want %d", len(pf.ColumnPaths), len(wantCols))
+	}
+	for i, want := range wantCols {
+		if pf.ColumnPaths[i] != want {
+			t.Errorf("ColumnPaths[%d] = %q, want %q", i, pf.ColumnPaths[i], want)
+		}
+	}
+	if pf.NumGroups < 1 {
+		t.Fatalf("NumGroups = %d, want >= 1", pf.NumGroups)
+	}
+
+	// Every column chunk has a non-zero offset (data starts after the
+	// 4-byte magic header) and a positive length, and the byte ranges
+	// must not overlap within a single row group when sorted.
+	for gi, rg := range pf.RowGroups {
+		if rg.NumRows == 0 {
+			t.Errorf("row group %d has zero rows", gi)
+		}
+		if len(rg.ColumnChunks) != pf.NumCols {
+			t.Errorf("row group %d ColumnChunks len = %d, want %d", gi, len(rg.ColumnChunks), pf.NumCols)
+		}
+		for ci, cc := range rg.ColumnChunks {
+			if cc.Offset <= 0 {
+				t.Errorf("row group %d col %d offset = %d, want > 0", gi, ci, cc.Offset)
+			}
+			if cc.Length <= 0 {
+				t.Errorf("row group %d col %d length = %d, want > 0", gi, ci, cc.Length)
+			}
+			if cc.Offset+cc.Length > int64(len(data)) {
+				t.Errorf("row group %d col %d overruns file: end %d > size %d", gi, ci, cc.Offset+cc.Length, len(data))
+			}
+		}
+	}
+}
+
+func TestParseFromReader_RejectsInvalid(t *testing.T) {
+	if _, err := ParseFromReader(nil, 100); err == nil {
+		t.Error("nil reader should fail")
+	}
+	if _, err := ParseFromReader(bytes.NewReader([]byte("PAR1")), 0); err == nil {
+		t.Error("zero size should fail")
+	}
+	if _, err := ParseFromReader(bytes.NewReader([]byte("not parquet at all")), 18); err == nil {
+		t.Error("non-parquet bytes should fail")
+	}
+}
+
+// columnChunkBytes prefers the dictionary page offset when present
+// because the dictionary page must be fetched alongside the data
+// pages of the same chunk.
+func TestColumnChunkBytes_DictionaryOffsetPreferred(t *testing.T) {
+	// Synthesize a fixture where dictionary encoding is likely:
+	// repeating low-cardinality strings nudge parquet-go to write a
+	// dictionary page. Then check that at least one column chunk
+	// reports a start offset before its data-page offset.
+	rows := make([]sampleRow, 2000)
+	for i := range rows {
+		rows[i] = sampleRow{ID: int64(i % 5), Name: "tenant-" + string(rune('a'+(i%5))), Value: float64(i % 5)}
+	}
+	data := writeFixture(t, rows)
+	pf, err := ParseFromReader(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+
+	// Sanity: at least one chunk's range should be plausible (length
+	// is positive and offset doesn't exceed the file). Beyond that we
+	// can't depend on a specific encoding choice from parquet-go.
+	for _, rg := range pf.RowGroups {
+		for _, cc := range rg.ColumnChunks {
+			if cc.Offset+cc.Length > int64(len(data)) {
+				t.Fatalf("chunk overruns file")
+			}
+		}
+	}
+}

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 }

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")
+	}
+}

weed/parquet_pushdown/m1_round_trip_test.go

@@ -0,0 +1,260 @@
+package parquet_pushdown
+
+// m1_round_trip_test.go is the M1 round-trip integration test from
+// PARQUET_PUSHDOWN_DEV_PLAN.md:
+//
+//   "Round-trip integration test against a 3-file Iceberg table built
+//   in test/parquet_pushdown/fixtures/.
+//
+//   Footer cache hit-rate test (second call avoids re-parse; verified
+//   via PushdownStats.FooterCacheHits)."
+//
+// It generates three Parquet files in a temp dir using parquet-go,
+// boots the SeaweedParquetPushdown service on a bufconn listener
+// with LocalLoader pointing at the temp dir, and submits real RPCs.
+// It does NOT require a filer; the filer-backed loader is M2+.
+
+import (
+	"bytes"
+	"context"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/parquet-go/parquet-go"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+type fixtureRow struct {
+	ID    int64
+	Name  string
+	Value float64
+}
+
+func writeParquetFile(t *testing.T, path string, rows []fixtureRow) {
+	t.Helper()
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[fixtureRow](&buf, parquet.SchemaOf(fixtureRow{}))
+	if _, err := w.Write(rows); err != nil {
+		t.Fatalf("write rows: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+	if err := os.WriteFile(path, buf.Bytes(), 0o600); err != nil {
+		t.Fatalf("write file: %v", err)
+	}
+}
+
+func makeFixtureTable(t *testing.T) (dir string, files []string, sizes []int64) {
+	t.Helper()
+	dir = t.TempDir()
+	for i, count := range []int{500, 750, 1000} {
+		name := filepath.Join(dir, "part-0000"+string(rune('1'+i))+".parquet")
+		rows := make([]fixtureRow, count)
+		for k := range rows {
+			rows[k] = fixtureRow{ID: int64(i*10000 + k), Name: "r", Value: float64(k)}
+		}
+		writeParquetFile(t, name, rows)
+		st, err := os.Stat(name)
+		if err != nil {
+			t.Fatalf("stat: %v", err)
+		}
+		files = append(files, name)
+		sizes = append(sizes, st.Size())
+	}
+	return dir, files, sizes
+}
+
+func startServiceForFixtures(t *testing.T) (pb.SeaweedParquetPushdownClient, func()) {
+	t.Helper()
+	lis := bufconn.Listen(1 << 20)
+	srv := grpc.NewServer()
+	svc := New(Options{
+		Version:   "m1-test",
+		TrustMode: TrustModeCatalogValidated,
+		Loader:    LocalLoader{},
+		CacheSize: 32,
+	})
+	pb.RegisterSeaweedParquetPushdownServer(srv, svc)
+	go func() { _ = srv.Serve(lis) }()
+
+	dialer := func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }
+	conn, err := grpc.NewClient(
+		"passthrough:///bufnet",
+		grpc.WithContextDialer(dialer),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		srv.Stop()
+		t.Fatalf("dial: %v", err)
+	}
+	return pb.NewSeaweedParquetPushdownClient(conn), func() {
+		_ = conn.Close()
+		srv.Stop()
+	}
+}
+
+func descriptorsFor(t *testing.T, files []string, sizes []int64) []*pb.DataFileDescriptor {
+	t.Helper()
+	out := make([]*pb.DataFileDescriptor, len(files))
+	for i, p := range files {
+		out[i] = &pb.DataFileDescriptor{Path: p, SizeBytes: sizes[i], RecordCount: 1}
+	}
+	return out
+}
+
+// Round-trip: three files, project two columns, no predicate.
+// Expect 3 files × N row groups × 2 columns of FileRanges, each with
+// a positive offset / length within the underlying file.
+func TestPushdown_M1_RoundTrip(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "ID"}, {Path: "Value"}},
+	})
+	if err != nil {
+		t.Fatalf("pushdown: %v", err)
+	}
+	if len(resp.FileRanges) == 0 {
+		t.Fatal("expected file ranges, got none")
+	}
+
+	// Per file, every range must reference that file and fit inside it.
+	byFile := map[string]int64{}
+	for i, p := range files {
+		byFile[p] = sizes[i]
+	}
+	for i, r := range resp.FileRanges {
+		size, ok := byFile[r.File]
+		if !ok {
+			t.Fatalf("range %d references unknown file %q", i, r.File)
+		}
+		if r.Offset <= 0 || r.Length <= 0 {
+			t.Errorf("range %d has invalid bounds (offset=%d length=%d)", i, r.Offset, r.Length)
+		}
+		if r.Offset+r.Length > size {
+			t.Errorf("range %d overruns file %q (offset=%d length=%d size=%d)", i, r.File, r.Offset, r.Length, size)
+		}
+	}
+
+	// First call: every footer is a miss; no hits.
+	if got := resp.Stats.FooterCacheHits; got != 0 {
+		t.Errorf("first call FooterCacheHits = %d, want 0", got)
+	}
+	if got := resp.Stats.FooterCacheMisses; got != int64(len(files)) {
+		t.Errorf("first call FooterCacheMisses = %d, want %d", got, len(files))
+	}
+}
+
+// Footer cache hit-rate: a second identical request must hit the
+// cache for every file (zero misses).
+func TestPushdown_M1_FooterCacheHits(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	req := &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "ID"}},
+	}
+
+	if _, err := client.Pushdown(ctx, req); err != nil {
+		t.Fatalf("warm-up: %v", err)
+	}
+
+	resp, err := client.Pushdown(ctx, req)
+	if err != nil {
+		t.Fatalf("second call: %v", err)
+	}
+	if got := resp.Stats.FooterCacheHits; got != int64(len(files)) {
+		t.Errorf("FooterCacheHits = %d, want %d", got, len(files))
+	}
+	if got := resp.Stats.FooterCacheMisses; got != 0 {
+		t.Errorf("FooterCacheMisses = %d, want 0", got)
+	}
+}
+
+// Empty Columns list returns ranges for every column in every row group.
+func TestPushdown_M1_NoColumnProjection(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+	})
+	if err != nil {
+		t.Fatalf("pushdown: %v", err)
+	}
+	// Every fixture has 3 columns; the count must be a multiple of 3.
+	if len(resp.FileRanges)%3 != 0 {
+		t.Errorf("FileRanges count %d not a multiple of column count 3", len(resp.FileRanges))
+	}
+}
+
+// An unknown column path is rejected with InvalidArgument before any
+// range is emitted.
+func TestPushdown_M1_UnknownColumn(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{Path: "definitely_not_a_real_column"}},
+	})
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+}
+
+// A field-id-only ColumnRef without a Path hint is unsupported in M1
+// because schema-based resolution lands in M3.
+func TestPushdown_M1_FieldIdOnlyRejected(t *testing.T) {
+	_, files, sizes := makeFixtureTable(t)
+	client, cleanup := startServiceForFixtures(t)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles:  descriptorsFor(t, files, sizes),
+		Columns:    []*pb.ColumnRef{{FieldId: 1}}, // no Path
+	})
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+}

weed/parquet_pushdown/request_validation.go

@@ -0,0 +1,152 @@
+package parquet_pushdown
+
+import (
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+// Hard caps applied to every incoming request before any work is done.
+// These guard against accidental or hostile blow-up; they sit well
+// above any plausible legitimate request and are not user-tunable.
+const (
+	maxDataFiles      = 100_000
+	maxColumns        = 10_000
+	maxPredicateBytes = 16 * 1024 * 1024 // matches gRPC default max message size
+	maxRowIdsCap      = 10_000_000
+	maxVectorDim      = 4096
+	maxTopK           = 10_000
+	maxDeletesPerFile = 10_000
+)
+
+// validateRequest enforces request-shape limits before the server
+// touches any data. Returns a gRPC status error with InvalidArgument
+// when something is wrong, nil otherwise.
+//
+// This is intentionally schema-only validation — it does not consult
+// the Iceberg catalog (that is M3 / catalog-validated mode) or read
+// any files. Schema-only checks run in every trust mode.
+func validateRequest(req *pb.ParquetPushdownRequest) error {
+	if req == nil {
+		return status.Error(codes.InvalidArgument, "request is nil")
+	}
+	if req.Table == "" {
+		return status.Error(codes.InvalidArgument, "table is required")
+	}
+	if len(req.DataFiles) == 0 {
+		return status.Error(codes.InvalidArgument, "data_files must not be empty")
+	}
+	if len(req.DataFiles) > maxDataFiles {
+		return status.Errorf(codes.InvalidArgument, "data_files exceeds cap %d", maxDataFiles)
+	}
+	if len(req.Columns) > maxColumns {
+		return status.Errorf(codes.InvalidArgument, "columns exceeds cap %d", maxColumns)
+	}
+	if len(req.Predicate) > maxPredicateBytes {
+		return status.Errorf(codes.InvalidArgument, "predicate exceeds cap %d bytes", maxPredicateBytes)
+	}
+	if (req.PredicateKind != pb.PredicateKind_PREDICATE_KIND_UNSPECIFIED) != (len(req.Predicate) > 0) {
+		return status.Error(codes.InvalidArgument, "predicate_kind and predicate must be set together")
+	}
+	// A bound predicate is meaningless without naming the schema its
+	// field references resolve under, so require schema_id whenever a
+	// predicate is present. The deeper check (every field id is in
+	// the snapshot's schema) needs catalog access and runs in M3.
+	if len(req.Predicate) > 0 && req.SchemaId == 0 {
+		return status.Error(codes.InvalidArgument, "predicate requires schema_id so field references can be bound")
+	}
+	if req.MaxRowIds < 0 || req.MaxRowIds > maxRowIdsCap {
+		return status.Errorf(codes.InvalidArgument, "max_row_ids must be in [0, %d]", maxRowIdsCap)
+	}
+	if req.RequestRowIds && req.MaxRowIds == 0 {
+		return status.Error(codes.InvalidArgument, "request_row_ids requires a positive max_row_ids")
+	}
+	if req.Limit < 0 {
+		return status.Error(codes.InvalidArgument, "limit must be non-negative")
+	}
+	if err := validateVectorQuery(req.VectorQuery); err != nil {
+		return err
+	}
+	for i, df := range req.DataFiles {
+		if err := validateDataFile(df); err != nil {
+			return status.Errorf(codes.InvalidArgument, "data_files[%d]: %s", i, err.Error())
+		}
+	}
+	return nil
+}
+
+func validateVectorQuery(q *pb.VectorQuery) error {
+	if q == nil {
+		return nil
+	}
+	if q.Column == nil || (q.Column.FieldId == 0 && q.Column.Path == "") {
+		return status.Error(codes.InvalidArgument, "vector_query.column must identify a column")
+	}
+	if len(q.Vector) == 0 {
+		return status.Error(codes.InvalidArgument, "vector_query.vector must not be empty")
+	}
+	if len(q.Vector) > maxVectorDim {
+		return status.Errorf(codes.InvalidArgument, "vector_query.vector dim exceeds cap %d", maxVectorDim)
+	}
+	if q.TopK <= 0 || q.TopK > maxTopK {
+		return status.Errorf(codes.InvalidArgument, "vector_query.top_k must be in [1, %d]", maxTopK)
+	}
+	if q.Metric == pb.VectorMetric_VECTOR_METRIC_UNSPECIFIED {
+		return status.Error(codes.InvalidArgument, "vector_query.metric must be set")
+	}
+	return nil
+}
+
+func validateDataFile(df *pb.DataFileDescriptor) error {
+	if df == nil {
+		return status.Error(codes.InvalidArgument, "descriptor is nil")
+	}
+	if df.Path == "" {
+		return status.Error(codes.InvalidArgument, "path is required")
+	}
+	if df.SizeBytes < 0 {
+		return status.Error(codes.InvalidArgument, "size_bytes must be non-negative")
+	}
+	if df.RecordCount < 0 {
+		return status.Error(codes.InvalidArgument, "record_count must be non-negative")
+	}
+	if len(df.Deletes) > maxDeletesPerFile {
+		return status.Errorf(codes.InvalidArgument, "deletes exceeds cap %d", maxDeletesPerFile)
+	}
+	for i, del := range df.Deletes {
+		if err := validateDeleteFile(del); err != nil {
+			return status.Errorf(codes.InvalidArgument, "deletes[%d]: %s", i, err.Error())
+		}
+	}
+	return nil
+}
+
+func validateDeleteFile(del *pb.DeleteFileRef) error {
+	if del == nil {
+		return status.Error(codes.InvalidArgument, "delete is nil")
+	}
+	if del.Path == "" {
+		return status.Error(codes.InvalidArgument, "path is required")
+	}
+	switch del.Content {
+	case pb.DeleteContent_POSITION_DELETES:
+		if del.FileFormat == pb.FileFormat_FILE_FORMAT_PUFFIN {
+			// Deletion vector: blob bounds are mandatory and the DV
+			// must reference a single data file.
+			if del.BlobLength <= 0 {
+				return status.Error(codes.InvalidArgument, "deletion vector requires positive blob_length")
+			}
+			if del.ReferencedDataFile == "" {
+				return status.Error(codes.InvalidArgument, "deletion vector requires referenced_data_file")
+			}
+		}
+	case pb.DeleteContent_EQUALITY_DELETES:
+		if len(del.EqualityFieldIds) == 0 {
+			return status.Error(codes.InvalidArgument, "equality delete requires equality_field_ids")
+		}
+	default:
+		return status.Error(codes.InvalidArgument, "unsupported delete content type")
+	}
+	return nil
+}

weed/parquet_pushdown/request_validation_predicate_test.go

@@ -0,0 +1,33 @@
+package parquet_pushdown
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+// A predicate paired with a SchemaId is the well-formed shape; a
+// well-formed predicate without SchemaId must be rejected (deeper
+// catalog binding lands in M3).
+func TestValidateRequest_PredicateRequiresSchemaId(t *testing.T) {
+	req := validRequest()
+	req.PredicateKind = pb.PredicateKind_PREDICATE_KIND_SUBSTRAIT
+	req.Predicate = []byte("bound expression bytes")
+	req.SchemaId = 7
+
+	if err := validateRequest(req); err != nil {
+		t.Fatalf("predicate + schema_id should pass, got %v", err)
+	}
+
+	req.SchemaId = 0
+	err := validateRequest(req)
+	if err == nil {
+		t.Fatal("predicate without schema_id should be rejected")
+	}
+	if status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("got %v, want InvalidArgument (err=%v)", status.Code(err), err)
+	}
+}

weed/parquet_pushdown/request_validation_test.go

@@ -0,0 +1,145 @@
+package parquet_pushdown
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+func validRequest() *pb.ParquetPushdownRequest {
+	return &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 42,
+		DataFiles: []*pb.DataFileDescriptor{
+			{Path: "s3://bkt/db/t/data/part-00001.parquet", SizeBytes: 1024, RecordCount: 100, DataSequenceNumber: 1},
+		},
+		Columns: []*pb.ColumnRef{{FieldId: 1}},
+	}
+}
+
+func TestValidateRequest_OK(t *testing.T) {
+	if err := validateRequest(validRequest()); err != nil {
+		t.Fatalf("expected ok, got %v", err)
+	}
+}
+
+func TestValidateRequest_RejectsEmpty(t *testing.T) {
+	cases := []struct {
+		name string
+		mut  func(*pb.ParquetPushdownRequest)
+	}{
+		{"nil request", func(r *pb.ParquetPushdownRequest) { *r = pb.ParquetPushdownRequest{} }},
+		{"missing table", func(r *pb.ParquetPushdownRequest) { r.Table = "" }},
+		{"empty data files", func(r *pb.ParquetPushdownRequest) { r.DataFiles = nil }},
+		{"data file missing path", func(r *pb.ParquetPushdownRequest) { r.DataFiles[0].Path = "" }},
+		{"negative size", func(r *pb.ParquetPushdownRequest) { r.DataFiles[0].SizeBytes = -1 }},
+		{"negative record count", func(r *pb.ParquetPushdownRequest) { r.DataFiles[0].RecordCount = -1 }},
+		{"negative limit", func(r *pb.ParquetPushdownRequest) { r.Limit = -1 }},
+		{"row ids without cap", func(r *pb.ParquetPushdownRequest) { r.RequestRowIds = true; r.MaxRowIds = 0 }},
+		{"max row ids over cap", func(r *pb.ParquetPushdownRequest) { r.MaxRowIds = maxRowIdsCap + 1 }},
+		{"predicate without kind", func(r *pb.ParquetPushdownRequest) { r.Predicate = []byte("x") }},
+		{"predicate kind without bytes", func(r *pb.ParquetPushdownRequest) {
+			r.PredicateKind = pb.PredicateKind_PREDICATE_KIND_SUBSTRAIT
+		}},
+		{"predicate without schema_id", func(r *pb.ParquetPushdownRequest) {
+			r.PredicateKind = pb.PredicateKind_PREDICATE_KIND_SUBSTRAIT
+			r.Predicate = []byte("bound expression")
+			r.SchemaId = 0
+		}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			req := validRequest()
+			c.mut(req)
+			err := validateRequest(req)
+			if err == nil {
+				t.Fatal("expected error")
+			}
+			if status.Code(err) != codes.InvalidArgument {
+				t.Fatalf("expected InvalidArgument, got %v", status.Code(err))
+			}
+		})
+	}
+}
+
+func TestValidateRequest_VectorQuery(t *testing.T) {
+	cases := []struct {
+		name string
+		q    *pb.VectorQuery
+		want codes.Code
+	}{
+		{"nil ok", nil, codes.OK},
+		{"missing column", &pb.VectorQuery{Vector: []float32{1}, TopK: 1, Metric: pb.VectorMetric_VECTOR_METRIC_L2}, codes.InvalidArgument},
+		{"empty vector", &pb.VectorQuery{Column: &pb.ColumnRef{FieldId: 1}, TopK: 1, Metric: pb.VectorMetric_VECTOR_METRIC_L2}, codes.InvalidArgument},
+		{"top_k zero", &pb.VectorQuery{Column: &pb.ColumnRef{FieldId: 1}, Vector: []float32{1}, Metric: pb.VectorMetric_VECTOR_METRIC_L2}, codes.InvalidArgument},
+		{"unspecified metric", &pb.VectorQuery{Column: &pb.ColumnRef{FieldId: 1}, Vector: []float32{1}, TopK: 1}, codes.InvalidArgument},
+		{"ok", &pb.VectorQuery{Column: &pb.ColumnRef{FieldId: 1}, Vector: []float32{1}, TopK: 1, Metric: pb.VectorMetric_VECTOR_METRIC_L2}, codes.OK},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			req := validRequest()
+			req.VectorQuery = c.q
+			err := validateRequest(req)
+			if status.Code(err) != c.want {
+				t.Fatalf("got %v, want %v (err=%v)", status.Code(err), c.want, err)
+			}
+		})
+	}
+}
+
+func TestValidateRequest_DeleteFiles(t *testing.T) {
+	cases := []struct {
+		name string
+		del  *pb.DeleteFileRef
+		want codes.Code
+	}{
+		{
+			"position-delete file ok",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_POSITION_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PARQUET},
+			codes.OK,
+		},
+		{
+			"deletion vector ok",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_POSITION_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PUFFIN, BlobLength: 16, ReferencedDataFile: "data.parquet"},
+			codes.OK,
+		},
+		{
+			"deletion vector missing blob length",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_POSITION_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PUFFIN, ReferencedDataFile: "data.parquet"},
+			codes.InvalidArgument,
+		},
+		{
+			"deletion vector missing referenced data file",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_POSITION_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PUFFIN, BlobLength: 16},
+			codes.InvalidArgument,
+		},
+		{
+			"equality delete missing field ids",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_EQUALITY_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PARQUET},
+			codes.InvalidArgument,
+		},
+		{
+			"equality delete ok",
+			&pb.DeleteFileRef{Path: "p", Content: pb.DeleteContent_EQUALITY_DELETES, FileFormat: pb.FileFormat_FILE_FORMAT_PARQUET, EqualityFieldIds: []int32{1}},
+			codes.OK,
+		},
+		{
+			"unspecified content rejected",
+			&pb.DeleteFileRef{Path: "p"},
+			codes.InvalidArgument,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			req := validRequest()
+			req.DataFiles[0].Deletes = []*pb.DeleteFileRef{c.del}
+			err := validateRequest(req)
+			if status.Code(err) != c.want {
+				t.Fatalf("got %v, want %v (err=%v)", status.Code(err), c.want, err)
+			}
+		})
+	}
+}

weed/parquet_pushdown/service.go

@@ -0,0 +1,249 @@
+// Package parquet_pushdown implements the SeaweedFS-aware Parquet
+// pushdown service. See PARQUET_PUSHDOWN_DESIGN.md and
+// PARQUET_PUSHDOWN_DEV_PLAN.md at the repo root for the surrounding
+// design and the milestone plan.
+//
+// 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"
+)
+
+// TrustMode controls how the server validates the request's
+// DataFiles and Deletes against the Iceberg catalog. See the design's
+// "Trust Model and Catalog Validation" section.
+type TrustMode string
+
+const (
+	// TrustModeCatalogValidated is the default. The server reads the
+	// Iceberg snapshot and verifies every DataFileDescriptor and
+	// DeleteFileRef against the manifest before serving.
+	TrustModeCatalogValidated TrustMode = "catalog-validated"
+
+	// TrustModeConnectorTrusted is a developer-only mode that skips
+	// catalog validation. Production builds must reject this mode at
+	// configuration time.
+	TrustModeConnectorTrusted TrustMode = "connector-trusted"
+)
+
+// 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. 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.
+// One Service instance per daemon process.
+type Service struct {
+	pb.UnimplementedSeaweedParquetPushdownServer
+
+	version   string
+	trustMode TrustMode
+	loader    Loader
+	cache     *footer.Cache
+}
+
+// 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,
+	}
+}
+
+// Ping returns daemon liveness information. Used by smoke tests and
+// connector health checks. Cheap; never reads files.
+func (s *Service) Ping(_ context.Context, _ *pb.PingRequest) (*pb.PingResponse, error) {
+	return &pb.PingResponse{
+		Version:   s.version,
+		TrustMode: string(s.trustMode),
+	}, nil
+}
+
+// Pushdown evaluates a request and returns surviving file/row-group/
+// page byte ranges.
+//
+// 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
+	}
+
+	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()
+	}
+
+	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

@@ -0,0 +1,175 @@
+package parquet_pushdown
+
+// service_grpc_test.go is the M0 integration test. It boots the
+// Service on an in-process gRPC bufconn listener (no filer, no
+// network) and exercises Ping and Pushdown end-to-end over real gRPC.
+//
+// The acceptance criterion from PARQUET_PUSHDOWN_DEV_PLAN.md is
+// "calls Pushdown over gRPC, gets the expected Unimplemented error
+// path with stats populated"; this file proves both, plus that the
+// Pushdown Unimplemented status carries a PushdownStats detail with
+// trust_mode and a non-zero server_time_micros.
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+const bufSize = 1 << 20
+
+func startTestServer(t *testing.T, opts Options) (pb.SeaweedParquetPushdownClient, func()) {
+	t.Helper()
+
+	lis := bufconn.Listen(bufSize)
+	srv := grpc.NewServer()
+	pb.RegisterSeaweedParquetPushdownServer(srv, New(opts))
+
+	go func() {
+		if err := srv.Serve(lis); err != nil {
+			t.Logf("test server stopped: %v", err)
+		}
+	}()
+
+	dialer := func(ctx context.Context, _ string) (net.Conn, error) {
+		return lis.DialContext(ctx)
+	}
+	conn, err := grpc.NewClient(
+		"passthrough:///bufnet",
+		grpc.WithContextDialer(dialer),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		srv.Stop()
+		t.Fatalf("dial bufconn: %v", err)
+	}
+	cleanup := func() {
+		_ = conn.Close()
+		srv.Stop()
+	}
+	return pb.NewSeaweedParquetPushdownClient(conn), cleanup
+}
+
+func TestPing_ReportsConfiguredVersionAndTrustMode(t *testing.T) {
+	client, cleanup := startTestServer(t, Options{
+		Version:   "test-1.2.3",
+		TrustMode: TrustModeCatalogValidated,
+	})
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	resp, err := client.Ping(ctx, &pb.PingRequest{})
+	if err != nil {
+		t.Fatalf("ping failed: %v", err)
+	}
+	if resp.Version != "test-1.2.3" {
+		t.Errorf("version = %q, want %q", resp.Version, "test-1.2.3")
+	}
+	if resp.TrustMode != string(TrustModeCatalogValidated) {
+		t.Errorf("trust_mode = %q, want %q", resp.TrustMode, TrustModeCatalogValidated)
+	}
+}
+
+func TestPing_DefaultsTrustModeToCatalogValidated(t *testing.T) {
+	client, cleanup := startTestServer(t, Options{Version: "test"})
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	resp, err := client.Ping(ctx, &pb.PingRequest{})
+	if err != nil {
+		t.Fatalf("ping failed: %v", err)
+	}
+	if resp.TrustMode != string(TrustModeCatalogValidated) {
+		t.Errorf("trust_mode default = %q, want %q", resp.TrustMode, TrustModeCatalogValidated)
+	}
+}
+
+// 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",
+		TrustMode: TrustModeCatalogValidated,
+	})
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	req := &pb.ParquetPushdownRequest{
+		Table:      "db.t",
+		SnapshotId: 1,
+		DataFiles: []*pb.DataFileDescriptor{
+			{Path: "s3://b/p.parquet", SizeBytes: 1, RecordCount: 1},
+		},
+		Columns:       []*pb.ColumnRef{{FieldId: 1, Path: "id"}},
+		PredicateKind: pb.PredicateKind_PREDICATE_KIND_SUBSTRAIT,
+		Predicate:     []byte("bound"),
+		SchemaId:      1,
+	}
+
+	_, err := client.Pushdown(ctx, req)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		t.Fatalf("error is not a gRPC status: %v", err)
+	}
+	if st.Code() != codes.Unimplemented {
+		t.Fatalf("got code %v, want Unimplemented", st.Code())
+	}
+
+	// Acceptance: stats are attached as a detail on the Unimplemented
+	// status, with the daemon's trust mode and a non-zero elapsed.
+	var stats *pb.PushdownStats
+	for _, d := range st.Details() {
+		if s, ok := d.(*pb.PushdownStats); ok {
+			stats = s
+			break
+		}
+	}
+	if stats == nil {
+		t.Fatal("expected PushdownStats detail on Unimplemented status")
+	}
+	if stats.TrustMode != string(TrustModeCatalogValidated) {
+		t.Errorf("stats.trust_mode = %q, want %q", stats.TrustMode, TrustModeCatalogValidated)
+	}
+	if stats.ServerTimeMicros < 0 {
+		t.Errorf("stats.server_time_micros = %d, want >= 0", stats.ServerTimeMicros)
+	}
+}
+
+func TestPushdown_RejectsInvalidRequestBeforeUnimplemented(t *testing.T) {
+	client, cleanup := startTestServer(t, Options{Version: "test"})
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	// Empty data_files: validation should fire and produce
+	// InvalidArgument, not Unimplemented.
+	_, err := client.Pushdown(ctx, &pb.ParquetPushdownRequest{Table: "db.t"})
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("got %v, want InvalidArgument (err=%v)", status.Code(err), err)
+	}
+}

weed/parquet_pushdown/stats.go

@@ -0,0 +1,45 @@
+package parquet_pushdown
+
+import (
+	"time"
+
+	pb "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb"
+)
+
+// statsRecorder accumulates per-request stats and converts to the
+// 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
+	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)
+}
+
+func (r *statsRecorder) markIndexMissing(kind string) {
+	r.indexesMissing = append(r.indexesMissing, kind)
+}
+
+func (r *statsRecorder) toProto() *pb.PushdownStats {
+	return &pb.PushdownStats{
+		TrustMode:         string(r.trustMode),
+		ServerTimeMicros:  time.Since(r.start).Microseconds(),
+		FooterCacheHits:   r.footerCacheHits,
+		FooterCacheMisses: r.footerCacheMisses,
+		IndexesUsed:       r.indexesUsed,
+		IndexesMissing:    r.indexesMissing,
+	}
+}

weed/pb/Makefile

@@ -18,6 +18,8 @@ gen:
 	protoc worker.proto --go_out=./worker_pb --go-grpc_out=./worker_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
 	mkdir -p ./plugin_pb
 	protoc plugin.proto --go_out=./plugin_pb --go-grpc_out=./plugin_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
+	mkdir -p ./parquet_pushdown_pb
+	protoc parquet_pushdown.proto --go_out=./parquet_pushdown_pb --go-grpc_out=./parquet_pushdown_pb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative
 	# protoc filer.proto --java_out=../../other/java/client/src/main/java
 	cp filer.proto ../../other/java/client/src/main/proto
 	cp volume_server.proto master.proto remote.proto ../../seaweed-volume/proto/

weed/pb/parquet_pushdown.proto

@@ -0,0 +1,251 @@
+syntax = "proto3";
+
+package parquet_pushdown_pb;
+
+option go_package = "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb";
+
+// SeaweedParquetPushdown is the gRPC surface of the standalone
+// `weed pushdown` daemon. See PARQUET_PUSHDOWN_DESIGN.md and
+// PARQUET_PUSHDOWN_DEV_PLAN.md for the surrounding design.
+service SeaweedParquetPushdown {
+    // Ping returns daemon liveness information. Cheap; intended for
+    // smoke tests and connector health checks.
+    rpc Ping (PingRequest) returns (PingResponse);
+
+    // Pushdown takes a planner-resolved set of Iceberg data files and
+    // returns the byte ranges, row groups, pages, and (optionally) row
+    // refs that satisfy the request's predicate and vector clauses.
+    // M0 returns Unimplemented; later milestones fill in pruning logic.
+    rpc Pushdown (ParquetPushdownRequest) returns (ParquetPushdownResponse);
+}
+
+// -- Ping --------------------------------------------------------------------
+
+message PingRequest {}
+
+message PingResponse {
+    string version = 1;
+    // TrustMode the daemon is configured for: "catalog-validated" or
+    // "connector-trusted". Connector-trusted is dev-only.
+    string trust_mode = 2;
+}
+
+// -- Request -----------------------------------------------------------------
+
+message ParquetPushdownRequest {
+    string table = 1;
+    int64 snapshot_id = 2;
+
+    // The authoritative list of files to scan, already resolved by the
+    // client's Iceberg planner. The server validates these against the
+    // catalog when running in catalog-validated trust mode.
+    repeated DataFileDescriptor data_files = 3;
+
+    // Columns to project. Identified by Iceberg field id (preferred)
+    // or path hint (fallback for non-Iceberg-managed Parquet).
+    repeated ColumnRef columns = 4;
+
+    PredicateKind predicate_kind = 5;
+
+    // Predicate is a *bound* expression: every column reference must
+    // resolve to an Iceberg field id (Substrait field-reference IDs
+    // matching schema_id, or Iceberg Expression JSON with id-based
+    // references). Name-only references are rejected.
+    bytes predicate = 6;
+
+    VectorQuery vector_query = 7;
+
+    int32 limit = 8;
+
+    // If true, the response may include per-row refs (RowRef list).
+    // Bounded by max_row_ids.
+    bool request_row_ids = 9;
+    int32 max_row_ids = 10;
+
+    // schema_id is the Iceberg schema id the predicate (and any
+    // ColumnRef.field_id values) are bound to. The server confirms it
+    // matches the snapshot's current_schema_id (or a known historical
+    // schema) before evaluating. Field IDs not present in this schema
+    // cause the request to be rejected.
+    int32 schema_id = 11;
+}
+
+message DataFileDescriptor {
+    string path = 1;
+    int64 size_bytes = 2;       // Iceberg manifest file_size_in_bytes
+    int64 record_count = 3;     // Iceberg manifest record_count
+    string etag = 4;            // optional, when no Iceberg manifest
+
+    // Iceberg manifest entry's data_sequence_number; drives
+    // delete-file applicability (NOT file_sequence_number).
+    int64 data_sequence_number = 5;
+
+    // Partition spec id and serialized partition values; required to
+    // match delete-file applicability for partitioned tables.
+    int32 partition_spec_id = 6;
+    bytes partition_values = 7;
+
+    // Position-delete files, equality-delete files, and deletion
+    // vectors that apply to this data file. Discriminated by
+    // DeleteFileRef.content + DeleteFileRef.file_format.
+    repeated DeleteFileRef deletes = 8;
+}
+
+message DeleteFileRef {
+    string path = 1;
+    int64 size_bytes = 2;
+
+    // Iceberg manifest entry's data_sequence_number for this delete
+    // file. Applicability rule:
+    //   data_file.seq <= delete_file.seq  for position deletes / DVs
+    //   data_file.seq <  delete_file.seq  for equality deletes
+    int64 data_sequence_number = 3;
+
+    // Partition spec id and serialized partition values; must match
+    // the data file's partition for the delete to apply.
+    int32 partition_spec_id = 4;
+    bytes partition_values = 5;
+
+    DeleteContent content = 6;     // POSITION_DELETES / EQUALITY_DELETES
+    FileFormat file_format = 7;    // PARQUET / AVRO / ORC / PUFFIN
+
+    // Required for content == EQUALITY_DELETES, empty otherwise.
+    repeated int32 equality_field_ids = 8;
+
+    // Puffin-only: when content == POSITION_DELETES and file_format
+    // == PUFFIN (deletion vectors), the DV blob lives at
+    // (path, blob_offset, blob_length). blob_crc32 is the Puffin
+    // footer's per-blob CRC, used for tamper detection (not as
+    // identity for cache lookup).
+    int64 blob_offset = 9;
+    int64 blob_length = 10;
+    uint32 blob_crc32 = 11;
+
+    // When this delete file targets one specific data file (mandatory
+    // for v3 deletion vectors). Empty when the delete file may target
+    // many data files.
+    string referenced_data_file = 12;
+}
+
+enum DeleteContent {
+    DELETE_CONTENT_UNSPECIFIED = 0;
+    POSITION_DELETES = 1; // Iceberg manifest content=1 (file or DV)
+    EQUALITY_DELETES = 2; // Iceberg manifest content=2
+    // Iceberg v3 deletion vectors are POSITION_DELETES with
+    // file_format == PUFFIN; there is no separate enum value.
+}
+
+enum FileFormat {
+    FILE_FORMAT_UNSPECIFIED = 0;
+    FILE_FORMAT_PARQUET = 1;
+    FILE_FORMAT_AVRO = 2;
+    FILE_FORMAT_ORC = 3;
+    FILE_FORMAT_PUFFIN = 4;
+}
+
+enum PredicateKind {
+    PREDICATE_KIND_UNSPECIFIED = 0;
+    PREDICATE_KIND_SUBSTRAIT = 1; // Substrait ExtendedExpression protobuf
+    PREDICATE_KIND_ICEBERG = 2;   // Iceberg Expression JSON
+}
+
+enum VectorMetric {
+    VECTOR_METRIC_UNSPECIFIED = 0;
+    VECTOR_METRIC_L2 = 1;
+    VECTOR_METRIC_COSINE = 2;
+    VECTOR_METRIC_DOT = 3;
+}
+
+message VectorQuery {
+    ColumnRef column = 1;
+    repeated float vector = 2;
+    VectorMetric metric = 3;
+    int32 top_k = 4;
+    int32 nprobe = 5;
+}
+
+// ColumnRef identifies a column by Iceberg field id (stable across
+// rename and reordering). Path is an optional hint used only when
+// field_id == 0 (e.g. non-Iceberg Parquet that has no field ids).
+message ColumnRef {
+    int32 field_id = 1;
+    string path = 2;
+}
+
+// -- Response ----------------------------------------------------------------
+
+message ParquetPushdownResponse {
+    repeated FileRange file_ranges = 1;
+    repeated RowGroupRef row_groups = 2;
+    repeated PageRef pages = 3;
+
+    // Optional, emitted only when request.request_row_ids is true and
+    // the result fits within request.max_row_ids.
+    repeated ScoredRowRef row_refs = 4;
+
+    // True if a row-ref list was omitted or truncated due to size cap.
+    bool truncated = 5;
+
+    PushdownStats stats = 6;
+}
+
+message FileRange {
+    string file = 1;
+    int64 offset = 2;
+    int64 length = 3;
+}
+
+message RowGroupRef {
+    string file = 1;
+    int32 row_group = 2;
+}
+
+message PageRef {
+    string file = 1;
+    int32 row_group = 2;
+    ColumnRef column = 3;
+    int32 page = 4;
+    int64 offset = 5;
+    int64 length = 6;
+}
+
+// RowRef identifies a row by file-absolute position (matching Iceberg
+// position-delete semantics). row_group is a locality hint, not
+// authoritative.
+message RowRef {
+    string file = 1;
+    int32 row_group = 2;
+    int64 file_position = 3;
+}
+
+message ScoredRowRef {
+    RowRef ref = 1;
+    // has_score distinguishes "no score available" from "score is
+    // 0.0", which is a legal vector-similarity value (zero L2
+    // distance, zero dot product). Non-vector queries leave
+    // has_score false; vector queries always set it true.
+    bool has_score = 2;
+    float score = 3;
+}
+
+message PushdownStats {
+    // The trust mode that serviced this request.
+    string trust_mode = 1;
+
+    // Server-side wall time spent on this request, in microseconds.
+    int64 server_time_micros = 2;
+
+    // Filled in by later milestones as the corresponding subsystems
+    // come online. Names use the design doc's per-feature labels.
+    int64 footer_cache_hits = 3;
+    int64 footer_cache_misses = 4;
+    int64 row_groups_pruned = 5;
+    int64 pages_pruned = 6;
+    int64 bytes_planned_scan = 7;
+
+    // Side indexes consulted vs missing, by index kind ("bloom",
+    // "bitmap", "btree", "page", "vector", ...). Useful both for
+    // observability and for the connector's cost-model feedback loop.
+    repeated string indexes_used = 8;
+    repeated string indexes_missing = 9;
+}

weed/pb/parquet_pushdown_pb/parquet_pushdown_grpc.pb.go

@@ -0,0 +1,180 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.5.1
+// - protoc             v6.33.4
+// source: parquet_pushdown.proto
+
+package parquet_pushdown_pb
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	SeaweedParquetPushdown_Ping_FullMethodName     = "/parquet_pushdown_pb.SeaweedParquetPushdown/Ping"
+	SeaweedParquetPushdown_Pushdown_FullMethodName = "/parquet_pushdown_pb.SeaweedParquetPushdown/Pushdown"
+)
+
+// SeaweedParquetPushdownClient is the client API for SeaweedParquetPushdown service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// SeaweedParquetPushdown is the gRPC surface of the standalone
+// `weed pushdown` daemon. See PARQUET_PUSHDOWN_DESIGN.md and
+// PARQUET_PUSHDOWN_DEV_PLAN.md for the surrounding design.
+type SeaweedParquetPushdownClient interface {
+	// Ping returns daemon liveness information. Cheap; intended for
+	// smoke tests and connector health checks.
+	Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error)
+	// Pushdown takes a planner-resolved set of Iceberg data files and
+	// returns the byte ranges, row groups, pages, and (optionally) row
+	// refs that satisfy the request's predicate and vector clauses.
+	// M0 returns Unimplemented; later milestones fill in pruning logic.
+	Pushdown(ctx context.Context, in *ParquetPushdownRequest, opts ...grpc.CallOption) (*ParquetPushdownResponse, error)
+}
+
+type seaweedParquetPushdownClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewSeaweedParquetPushdownClient(cc grpc.ClientConnInterface) SeaweedParquetPushdownClient {
+	return &seaweedParquetPushdownClient{cc}
+}
+
+func (c *seaweedParquetPushdownClient) Ping(ctx context.Context, in *PingRequest, opts ...grpc.CallOption) (*PingResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(PingResponse)
+	err := c.cc.Invoke(ctx, SeaweedParquetPushdown_Ping_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *seaweedParquetPushdownClient) Pushdown(ctx context.Context, in *ParquetPushdownRequest, opts ...grpc.CallOption) (*ParquetPushdownResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ParquetPushdownResponse)
+	err := c.cc.Invoke(ctx, SeaweedParquetPushdown_Pushdown_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// SeaweedParquetPushdownServer is the server API for SeaweedParquetPushdown service.
+// All implementations must embed UnimplementedSeaweedParquetPushdownServer
+// for forward compatibility.
+//
+// SeaweedParquetPushdown is the gRPC surface of the standalone
+// `weed pushdown` daemon. See PARQUET_PUSHDOWN_DESIGN.md and
+// PARQUET_PUSHDOWN_DEV_PLAN.md for the surrounding design.
+type SeaweedParquetPushdownServer interface {
+	// Ping returns daemon liveness information. Cheap; intended for
+	// smoke tests and connector health checks.
+	Ping(context.Context, *PingRequest) (*PingResponse, error)
+	// Pushdown takes a planner-resolved set of Iceberg data files and
+	// returns the byte ranges, row groups, pages, and (optionally) row
+	// refs that satisfy the request's predicate and vector clauses.
+	// M0 returns Unimplemented; later milestones fill in pruning logic.
+	Pushdown(context.Context, *ParquetPushdownRequest) (*ParquetPushdownResponse, error)
+	mustEmbedUnimplementedSeaweedParquetPushdownServer()
+}
+
+// UnimplementedSeaweedParquetPushdownServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedSeaweedParquetPushdownServer struct{}
+
+func (UnimplementedSeaweedParquetPushdownServer) Ping(context.Context, *PingRequest) (*PingResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+}
+func (UnimplementedSeaweedParquetPushdownServer) Pushdown(context.Context, *ParquetPushdownRequest) (*ParquetPushdownResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Pushdown not implemented")
+}
+func (UnimplementedSeaweedParquetPushdownServer) mustEmbedUnimplementedSeaweedParquetPushdownServer() {
+}
+func (UnimplementedSeaweedParquetPushdownServer) testEmbeddedByValue() {}
+
+// UnsafeSeaweedParquetPushdownServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to SeaweedParquetPushdownServer will
+// result in compilation errors.
+type UnsafeSeaweedParquetPushdownServer interface {
+	mustEmbedUnimplementedSeaweedParquetPushdownServer()
+}
+
+func RegisterSeaweedParquetPushdownServer(s grpc.ServiceRegistrar, srv SeaweedParquetPushdownServer) {
+	// If the following call pancis, it indicates UnimplementedSeaweedParquetPushdownServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&SeaweedParquetPushdown_ServiceDesc, srv)
+}
+
+func _SeaweedParquetPushdown_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PingRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SeaweedParquetPushdownServer).Ping(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: SeaweedParquetPushdown_Ping_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SeaweedParquetPushdownServer).Ping(ctx, req.(*PingRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _SeaweedParquetPushdown_Pushdown_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ParquetPushdownRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SeaweedParquetPushdownServer).Pushdown(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: SeaweedParquetPushdown_Pushdown_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SeaweedParquetPushdownServer).Pushdown(ctx, req.(*ParquetPushdownRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// SeaweedParquetPushdown_ServiceDesc is the grpc.ServiceDesc for SeaweedParquetPushdown service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var SeaweedParquetPushdown_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "parquet_pushdown_pb.SeaweedParquetPushdown",
+	HandlerType: (*SeaweedParquetPushdownServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Ping",
+			Handler:    _SeaweedParquetPushdown_Ping_Handler,
+		},
+		{
+			MethodName: "Pushdown",
+			Handler:    _SeaweedParquetPushdown_Pushdown_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "parquet_pushdown.proto",
+}