mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-31 21:31:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

parquet_pushdown: predicate must be a bound expression with schema id

Browse Source 
Without naming a schema, a serialized predicate's column references
are ambiguous: the same field name can refer to different field IDs
across schema-evolution events, and Substrait field-reference IDs
mean nothing without saying which schema they index into.

Make the request shape unambiguous:

- Add ParquetPushdownRequest.SchemaId (Iceberg schema id) to the
  proto and to the design's API sketch.
- Document the binding contract in the design: the predicate must
  carry only field-id references; the connector binds before
  sending, the server checks ids against the schema.
- Validation now rejects (predicate set, schema_id == 0) with
  InvalidArgument before any work is done. Deeper "every field id
  exists in the snapshot's schema" check needs catalog access and
  lands in M3.
- Update / add tests covering the new gate.
This commit is contained in:
Chris Lu
2026-04-25 15:07:12 -07:00
parent 5fec310f73
commit 9561d95507
6 changed files with 101 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PARQUET_PUSHDOWN_DESIGN.md
+20 -1
View File
@@ -445,6 +445,15 @@ type ParquetPushdownRequest struct {
Table string
SnapshotId int64
// SchemaId is the Iceberg schema id the predicate is bound to —
// the schema the connector used when resolving column-name
// references to field IDs. The server confirms it matches the
// snapshot's current_schema_id (or a known historical schema)
// before evaluating the predicate. If the predicate refers to
// field IDs that are not present in this schema, the request is
// rejected.
SchemaId int32
// DataFiles is the authoritative list of files to scan. Each entry
// carries enough identity for the server to validate that its cached
// side indexes still apply, and enough delete-file context that the
@@ -454,7 +463,17 @@ type ParquetPushdownRequest struct {
Columns []ColumnRef
PredicateKind PredicateKind // SUBSTRAIT or ICEBERG_EXPRESSION
Predicate []byte // serialized per PredicateKind
// Predicate is a *bound* expression: every column reference must
// resolve to an Iceberg field id (Substrait field-reference IDs
// matching this request's SchemaId, or Iceberg Expression JSON
// with id-based references). Name-only references — the kind a
// SQL parser emits — are rejected; the connector is responsible
// for binding before sending. This avoids name-resolution
// ambiguity under schema evolution and removes a server-side
// dependency on the catalog's symbol table.
Predicate []byte
VectorQuery *VectorQuery
Limit int
RequestRowIds bool // include per-row refs in response (default false)
weed/parquet_pushdown/request_validation.go
+7
View File
@@ -49,6 +49,13 @@ func validateRequest(req *pb.ParquetPushdownRequest) error {
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)
}
weed/parquet_pushdown/request_validation_predicate_test.go
+33
View File
@@ -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
+5
View File
@@ -44,6 +44,11 @@ func TestValidateRequest_RejectsEmpty(t *testing.T) {
{"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) {
weed/pb/parquet_pushdown.proto
+13 -1
View File
@@ -46,7 +46,12 @@ message ParquetPushdownRequest {
repeated ColumnRef columns = 4;
PredicateKind predicate_kind = 5;
bytes predicate = 6; // serialized per predicate_kind
// 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;
@@ -56,6 +61,13 @@ message ParquetPushdownRequest {
// 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 {
weed/pb/parquet_pushdown_pb/parquet_pushdown.pb.go
+23 -5
View File
@@ -328,13 +328,23 @@ type ParquetPushdownRequest struct {
// or path hint (fallback for non-Iceberg-managed Parquet).
Columns []*ColumnRef `protobuf:"bytes,4,rep,name=columns,proto3" json:"columns,omitempty"`
PredicateKind PredicateKind `protobuf:"varint,5,opt,name=predicate_kind,json=predicateKind,proto3,enum=parquet_pushdown_pb.PredicateKind" json:"predicate_kind,omitempty"`
Predicate []byte `protobuf:"bytes,6,opt,name=predicate,proto3" json:"predicate,omitempty"` // serialized per predicate_kind
VectorQuery *VectorQuery `protobuf:"bytes,7,opt,name=vector_query,json=vectorQuery,proto3" json:"vector_query,omitempty"`
Limit int32 `protobuf:"varint,8,opt,name=limit,proto3" json:"limit,omitempty"`
// 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.
Predicate []byte `protobuf:"bytes,6,opt,name=predicate,proto3" json:"predicate,omitempty"`
VectorQuery *VectorQuery `protobuf:"bytes,7,opt,name=vector_query,json=vectorQuery,proto3" json:"vector_query,omitempty"`
Limit int32 `protobuf:"varint,8,opt,name=limit,proto3" json:"limit,omitempty"`
// If true, the response may include per-row refs (RowRef list).
// Bounded by max_row_ids.
RequestRowIds bool `protobuf:"varint,9,opt,name=request_row_ids,json=requestRowIds,proto3" json:"request_row_ids,omitempty"`
MaxRowIds int32 `protobuf:"varint,10,opt,name=max_row_ids,json=maxRowIds,proto3" json:"max_row_ids,omitempty"`
// 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.
SchemaId int32 `protobuf:"varint,11,opt,name=schema_id,json=schemaId,proto3" json:"schema_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -439,6 +449,13 @@ func (x *ParquetPushdownRequest) GetMaxRowIds() int32 {
return 0
}
func (x *ParquetPushdownRequest) GetSchemaId() int32 {
if x != nil {
return x.SchemaId
}
return 0
}
type DataFileDescriptor struct {
state protoimpl.MessageState `protogen:"open.v1"`
Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"`
@@ -1347,7 +1364,7 @@ const file_parquet_pushdown_proto_rawDesc = "" +
"\fPingResponse\x12\x18\n" +
"\aversion\x18\x01 \x01(\tR\aversion\x12\x1d\n" +
"\n" +
"trust_mode\x18\x02 \x01(\tR\ttrustMode\"\xdd\x03\n" +
"trust_mode\x18\x02 \x01(\tR\ttrustMode\"\xfa\x03\n" +
"\x16ParquetPushdownRequest\x12\x14\n" +
"\x05table\x18\x01 \x01(\tR\x05table\x12\x1f\n" +
"\vsnapshot_id\x18\x02 \x01(\x03R\n" +
@@ -1361,7 +1378,8 @@ const file_parquet_pushdown_proto_rawDesc = "" +
"\x05limit\x18\b \x01(\x05R\x05limit\x12&\n" +
"\x0frequest_row_ids\x18\t \x01(\bR\rrequestRowIds\x12\x1e\n" +
"\vmax_row_ids\x18\n" +
" \x01(\x05R\tmaxRowIds\"\xc5\x02\n" +
" \x01(\x05R\tmaxRowIds\x12\x1b\n" +
"\tschema_id\x18\v \x01(\x05R\bschemaId\"\xc5\x02\n" +
"\x12DataFileDescriptor\x12\x12\n" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x1d\n" +
"\n" +