From e610dad2a1431937c3c88fded48aeb21960bba4c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 25 Apr 2026 13:39:28 -0700 Subject: [PATCH] parquet_pushdown(M0-C2): add service skeleton and request validation Implement the gRPC service surface of the pushdown daemon: - Service struct embeds the generated UnimplementedSeaweed ParquetPushdownServer, exposes Ping (returns version + trust mode) and Pushdown (returns Unimplemented after request-shape validation). - request_validation.go enforces hard caps (max data files, columns, predicate bytes, row-id cap, vector dim, top_k) and content-type rules (Puffin DV requires blob_length + referenced_data_file; equality delete requires equality_field_ids; predicate_kind and predicate must be set together). - stats.go is a small accumulator that produces a PushdownStats message stamped with trust_mode and elapsed micros; M0 wires it into the Unimplemented status's details so the stats wire format is exercised end to end. - TrustMode constants encode the design's catalog-validated (default) and connector-trusted (dev-only) modes. Unit tests cover the validation matrix (rejects empty/oversize requests, accepts well-formed vector queries, accepts well-formed deletion-vector descriptors, rejects DVs missing blob_length / referenced_data_file). --- weed/parquet_pushdown/request_validation.go | 145 ++++++++++++++++++ .../request_validation_test.go | 140 +++++++++++++++++ weed/parquet_pushdown/service.go | 99 ++++++++++++ weed/parquet_pushdown/stats.go | 39 +++++ 4 files changed, 423 insertions(+) create mode 100644 weed/parquet_pushdown/request_validation.go create mode 100644 weed/parquet_pushdown/request_validation_test.go create mode 100644 weed/parquet_pushdown/service.go create mode 100644 weed/parquet_pushdown/stats.go diff --git a/weed/parquet_pushdown/request_validation.go b/weed/parquet_pushdown/request_validation.go new file mode 100644 index 000000000..65ebeeb36 --- /dev/null +++ b/weed/parquet_pushdown/request_validation.go @@ -0,0 +1,145 @@ +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") + } + 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 +} diff --git a/weed/parquet_pushdown/request_validation_test.go b/weed/parquet_pushdown/request_validation_test.go new file mode 100644 index 000000000..fc3e8a1b9 --- /dev/null +++ b/weed/parquet_pushdown/request_validation_test.go @@ -0,0 +1,140 @@ +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 + }}, + } + 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) + } + }) + } +} diff --git a/weed/parquet_pushdown/service.go b/weed/parquet_pushdown/service.go new file mode 100644 index 000000000..6f790217a --- /dev/null +++ b/weed/parquet_pushdown/service.go @@ -0,0 +1,99 @@ +// 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. +// +// M0 wires up the gRPC service and request validation; the actual +// pruning logic (parsed-footer cache, row-group pruning, scalar/page +// indexes, deletes, vectors) lands in M1+. +package parquet_pushdown + +import ( + "context" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + 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" +) + +// Options configures a Service. Fields the M0 skeleton does not yet +// consume (catalog client, filer client, predicate engine) will be +// added in their respective milestones. +type Options struct { + // Version is reported in PingResponse and stamped on PushdownStats. + Version string + + // TrustMode determines request-validation strictness. M0 only + // stamps the value into the response; the actual validation + // against the catalog lands in M3. + TrustMode TrustMode +} + +// Service implements parquet_pushdown_pb.SeaweedParquetPushdownServer. +// One Service instance per daemon process. +type Service struct { + pb.UnimplementedSeaweedParquetPushdownServer + + version string + trustMode TrustMode +} + +// New constructs a Service from Options. The caller is responsible +// for registering it on a gRPC server (see weed/parquet_pushdown/daemon). +func New(opts Options) *Service { + mode := opts.TrustMode + if mode == "" { + mode = TrustModeCatalogValidated + } + return &Service{ + version: opts.Version, + trustMode: mode, + } +} + +// 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. M0 returns Unimplemented after request-shape +// validation; later milestones replace the body. +// +// Unimplemented is attached with a PushdownStats detail so the wire +// format for stats is exercised end-to-end on M0. +func (s *Service) Pushdown(_ context.Context, req *pb.ParquetPushdownRequest) (*pb.ParquetPushdownResponse, error) { + stats := newStats(s.trustMode) + + if err := validateRequest(req); err != nil { + return nil, err + } + + st := status.New(codes.Unimplemented, "parquet pushdown evaluation lands in M1; M0 only validates request shape") + if withDetails, err := st.WithDetails(stats.toProto()); err == nil { + return nil, withDetails.Err() + } + return nil, st.Err() +} diff --git a/weed/parquet_pushdown/stats.go b/weed/parquet_pushdown/stats.go new file mode 100644 index 000000000..b6d8005e8 --- /dev/null +++ b/weed/parquet_pushdown/stats.go @@ -0,0 +1,39 @@ +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. M0 only records trust mode and +// elapsed server time; later milestones extend it with cache hits, +// pruning counts, and indexes-used / indexes-missing labels. +type statsRecorder struct { + start time.Time + trustMode TrustMode + indexesUsed []string + indexesMissing []string +} + +func newStats(trustMode TrustMode) *statsRecorder { + return &statsRecorder{start: time.Now(), trustMode: trustMode} +} + +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(), + IndexesUsed: r.indexesUsed, + IndexesMissing: r.indexesMissing, + } +}