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).
2026-09-01 13:47:51 +00:00 · 2026-04-25 13:39:28 -07:00
parent 2e423c0f31
commit e610dad2a1
4 changed files with 423 additions and 0 deletions
@@ -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
+}
@@ -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)
+			}
+		})
+	}
+}
@@ -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()
+}
@@ -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,
+	}
+}