diff --git a/vgi-client/src/aggregate.rs b/vgi-client/src/aggregate.rs index 6ca9cb0..751e72a 100644 --- a/vgi-client/src/aggregate.rs +++ b/vgi-client/src/aggregate.rs @@ -160,6 +160,7 @@ impl VgiClient { secrets: secrets.map(Bytes), attach_opaque_data: Some(cat.handle().clone()), schema_path: spec.schema_path.clone(), + argument_names: spec.argument_names.clone(), }; let response: AggregateBindResponse = call( self.transport_mut(), diff --git a/vgi-client/src/exchange.rs b/vgi-client/src/exchange.rs index 4a788a6..368c2ab 100644 --- a/vgi-client/src/exchange.rs +++ b/vgi-client/src/exchange.rs @@ -194,7 +194,9 @@ fn parse_parent_rows(md: &vgi_rpc::wire::Metadata) -> Result>> { ))); } Ok(Some( - raw.chunks_exact(4) + raw.as_chunks::<4>() + .0 + .iter() .map(|c| i32::from_le_bytes([c[0], c[1], c[2], c[3]])) .collect(), )) @@ -283,6 +285,7 @@ impl VgiClient { at_unit: spec.at.as_ref().map(|a| a.unit.clone()), at_value: spec.at.as_ref().map(|a| a.value.clone()), schema_path: spec.schema_path.clone(), + argument_names: spec.argument_names.clone(), }; let bind_call = envelope(request)?; let response: BindResponse = call( diff --git a/vgi-client/src/scan.rs b/vgi-client/src/scan.rs index 3f46637..724bcb1 100644 --- a/vgi-client/src/scan.rs +++ b/vgi-client/src/scan.rs @@ -87,6 +87,9 @@ pub struct BindSpec { pub schema_path: Option>, /// Call arguments. pub arguments: Arguments, + /// Full logical argument names. Inner `None` denotes an unnamed vararg; + /// outer `None` means names are unavailable. + pub argument_names: Option>>, /// Pre-serialized call arguments, used in place of [`Self::arguments`]. /// /// A catalog table's scan arguments arrive from the worker already IPC @@ -108,6 +111,7 @@ impl BindSpec { function_type: FunctionType::Table, schema_path: None, arguments: Arguments::new(), + argument_names: None, raw_arguments: None, settings: None, at: None, @@ -146,6 +150,17 @@ impl BindSpec { self.arguments = args; self } + + /// Set the names corresponding to the logical call arguments. + #[must_use] + pub fn with_argument_names(mut self, names: I) -> Self + where + I: IntoIterator>, + S: Into, + { + self.argument_names = Some(names.into_iter().map(|name| name.map(Into::into)).collect()); + self + } } /// A bound function, ready to scan. @@ -831,6 +846,7 @@ impl VgiClient { at_unit: spec.at.as_ref().map(|a| a.unit.clone()), at_value: spec.at.as_ref().map(|a| a.value.clone()), schema_path: spec.schema_path.clone(), + argument_names: spec.argument_names.clone(), }; // `init` echoes the whole bind call back, so keep the exact bytes we diff --git a/vgi-example-worker/src/aggregate/mod.rs b/vgi-example-worker/src/aggregate/mod.rs index 534db5d..b7d5548 100644 --- a/vgi-example-worker/src/aggregate/mod.rs +++ b/vgi-example-worker/src/aggregate/mod.rs @@ -502,7 +502,9 @@ fn pct_push(state: &[u8], v: f64) -> Vec { } fn pct_vals(state: &[u8]) -> Vec { state - .chunks_exact(8) + .as_chunks::<8>() + .0 + .iter() .map(|c| { let mut a = [0u8; 8]; a.copy_from_slice(c); diff --git a/vgi-example-worker/src/scalar/mod.rs b/vgi-example-worker/src/scalar/mod.rs index 8dfd7dc..1dab05a 100644 --- a/vgi-example-worker/src/scalar/mod.rs +++ b/vgi-example-worker/src/scalar/mod.rs @@ -13,8 +13,9 @@ use arrow_array::{ Array, BinaryArray, BooleanArray, Float64Array, Int64Array, RecordBatch, StringArray, StructArray, }; -use arrow_schema::DataType; +use arrow_schema::{DataType, Field, Schema}; use sha2::{Digest, Sha256}; +use std::sync::Arc; use util::*; use vgi::cache_control::CacheControl; use vgi::function::{ @@ -45,6 +46,7 @@ fn hex_of(bytes: &[u8]) -> String { pub fn register(w: &mut vgi::Worker) { w.register_scalar(DoubleFunction); w.register_scalar(AddValuesFunction); + w.register_scalar(ArgumentNamesProbeFunction); w.register_scalar(MultiplyFunction); w.register_scalar(PassthruFunction); w.register_scalar(CollatzStepsFunction); @@ -99,6 +101,73 @@ fn meta_ret(desc: &str, ret: DataType) -> FunctionMetadata { } } +/// Verifies that bind receives the complete resolved VGI 2 function signature. +pub struct ArgumentNamesProbeFunction; +impl ScalarFunction for ArgumentNamesProbeFunction { + fn name(&self) -> &str { + "argument_names_probe" + } + + fn metadata(&self) -> FunctionMetadata { + FunctionMetadata { + description: "Checks VGI 2.0 bind-time argument names".to_string(), + return_type: Some(DataType::Int64), + parameter_default_values: Some( + RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "scale", + DataType::Int64, + false, + )])), + vec![Arc::new(Int64Array::from(vec![2]))], + ) + .expect("valid argument_names_probe defaults"), + ), + ..Default::default() + } + } + + fn argument_specs(&self) -> Vec { + vec![ + ArgSpec::column("left", 0, "int64", "Left value"), + ArgSpec::column("right", 1, "int64", "Right value"), + ArgSpec::const_arg("scale", 2, "int64", "Scale factor").with_default(2), + ] + } + + fn on_bind(&self, params: &BindParams) -> Result { + let expected = Some(vec![ + Some("left".to_string()), + Some("right".to_string()), + Some("scale".to_string()), + ]); + if params.argument_names != expected { + return Err(RpcError::value_error(format!( + "argument_names_probe expected {expected:?}, got {:?}", + params.argument_names + ))); + } + Ok(BindResponse::result(DataType::Int64)) + } + + fn process(&self, params: &ProcessParams, batch: &RecordBatch) -> Result { + let left = arrow_cast::cast(batch.column(0), &DataType::Int64) + .map_err(|error| RpcError::runtime_error(error.to_string()))?; + let right = arrow_cast::cast(batch.column(1), &DataType::Int64) + .map_err(|error| RpcError::runtime_error(error.to_string()))?; + let left = left.as_primitive::(); + let right = right.as_primitive::(); + let scale = params.arguments.const_i64(2).unwrap_or(2); + let output: Int64Array = (0..batch.num_rows()) + .map(|index| { + (!left.is_null(index) && !right.is_null(index)) + .then(|| (left.value(index) + right.value(index)) * scale) + }) + .collect(); + result(params, arc(output)) + } +} + // --------------------------------------------------------------------------- // arithmetic // --------------------------------------------------------------------------- diff --git a/vgi-example-worker/src/table/splits.rs b/vgi-example-worker/src/table/splits.rs index 4008b30..9b0e131 100644 --- a/vgi-example-worker/src/table/splits.rs +++ b/vgi-example-worker/src/table/splits.rs @@ -1923,7 +1923,7 @@ impl TableFunction for SplitDynamicFilter { let rendered = params .current_pushdown_filters .as_ref() - .map(|pf| render_filters(&pf)) + .map(render_filters) .unwrap_or_else(|| "(none)".to_string()); Ok(Box::new(DynFilterProducer { schema: Self::schema(), diff --git a/vgi-protocol/src/generated/protocol_schemas.rs b/vgi-protocol/src/generated/protocol_schemas.rs index d6ec8b4..0f9f48b 100644 --- a/vgi-protocol/src/generated/protocol_schemas.rs +++ b/vgi-protocol/src/generated/protocol_schemas.rs @@ -1,7 +1,7 @@ // GENERATED by vgi.codegen.rust_schemas. DO NOT EDIT BY HAND. // // Generator: vgi-gen-rust-schemas v1 -// Content hash: 85668fb81b20 +// Content hash: af786789f87e // // To regenerate: // uv run --project ~/Development/vgi-python vgi-gen-rust-schemas \ @@ -117,6 +117,7 @@ pub fn function_info_schema() -> SchemaRef { Field::new("function_type", DataType::Dictionary(Box::new(DataType::Int16), Box::new(DataType::Utf8)), false), Field::new("arguments", DataType::Binary, false), Field::new("output_schema", DataType::Binary, false), + Field::new("parameter_default_values", DataType::Binary, true), Field::new("stability", DataType::Dictionary(Box::new(DataType::Int16), Box::new(DataType::Utf8)), true), Field::new("null_handling", DataType::Dictionary(Box::new(DataType::Int16), Box::new(DataType::Utf8)), true), Field::new("description", DataType::Utf8, false), @@ -330,6 +331,7 @@ pub fn aggregate_bind_request_schema() -> SchemaRef { Field::new("secrets", DataType::Binary, true), Field::new("attach_opaque_data", DataType::Binary, true), Field::new("schema_path", DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), true), + Field::new("argument_names", DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), true), ])) }) .clone() @@ -550,6 +552,7 @@ pub fn bind_request_schema() -> SchemaRef { Field::new("copy_from", DataType::Struct(Fields::from(vec![Field::new("format", DataType::Utf8, false), Field::new("file_path", DataType::Utf8, false), Field::new("expected_schema", DataType::Binary, false)])), true), Field::new("copy_to", DataType::Struct(Fields::from(vec![Field::new("format", DataType::Utf8, false), Field::new("file_path", DataType::Utf8, false)])), true), Field::new("schema_path", DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), true), + Field::new("argument_names", DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), true), ])) }) .clone() diff --git a/vgi-protocol/src/protocol/dtos.rs b/vgi-protocol/src/protocol/dtos.rs index 3a00741..51333e8 100644 --- a/vgi-protocol/src/protocol/dtos.rs +++ b/vgi-protocol/src/protocol/dtos.rs @@ -130,6 +130,10 @@ pub struct BindRequest { /// and a scan whose function resolved to a built-in carries none either. /// Additive nullable column; the C++ always emits it as of protocol 1.1.0. pub schema_path: Option, + /// One entry per logical call argument. Fixed arguments retain their + /// declared name, unnamed varargs are `None`, and named varargs retain the + /// caller-provided name. Outer `None` means names are unavailable. + pub argument_names: Option>>, // NOTE: the `copy_from` / `copy_to` struct columns are intentionally NOT // derived fields here. The C++ extension only appends them to the // BindRequest schema for a COPY ... FROM / COPY ... TO scan (omitting them @@ -155,7 +159,14 @@ pub struct BindRequest { pub fn backfill_bind_request( batch: arrow_array::RecordBatch, ) -> Result<(arrow_array::RecordBatch, bool)> { - ensure_schema_path(batch) + let (batch, legacy_peer) = ensure_schema_path(batch)?; + let names_type = arrow_schema::DataType::List(Arc::new(arrow_schema::Field::new( + "item", + arrow_schema::DataType::Utf8, + true, + ))); + let batch = ensure_nullable_columns(batch, &[("argument_names", names_type)])?; + Ok((batch, legacy_peer)) } /// Append a null `schema_path` column when the request batch lacks one, so a @@ -293,6 +304,16 @@ pub fn backfill_function_info(batch: arrow_array::RecordBatch) -> Result, pub stability: Option, pub null_handling: Option, pub description: String, @@ -1388,6 +1416,8 @@ pub struct AggregateBindRequest { /// RPC that re-resolves by name; `None` when the caller names no schema. /// Added in protocol 1.2.0. pub schema_path: Option, + /// Full logical argument order; inner `None` denotes an unnamed vararg. + pub argument_names: Option>>, } /// `AggregateBindResponse`. diff --git a/vgi/src/aggregate.rs b/vgi/src/aggregate.rs index fc7be3c..86f5381 100644 --- a/vgi/src/aggregate.rs +++ b/vgi/src/aggregate.rs @@ -24,6 +24,9 @@ pub const GROUP_COLUMN_NAME: &str = "__vgi_group_id"; /// Parameters for `aggregate_bind`. pub struct AggregateBindParams { pub arguments: Arguments, + /// One entry per logical call argument. Inner `None` is an unnamed vararg; + /// outer `None` means the client could not provide names. + pub argument_names: Option>>, pub input_schema: Option, pub settings: Settings, /// Statically pre-resolved secrets, delivered on `AggregateBindRequest.secrets` diff --git a/vgi/src/catalog.rs b/vgi/src/catalog.rs index 6d15617..65ccc72 100644 --- a/vgi/src/catalog.rs +++ b/vgi/src/catalog.rs @@ -447,6 +447,7 @@ pub fn default_function_info(name: &str, function_type: &str) -> FunctionInfo { function_type: enums::dict(function_type), arguments: Bytes::from(Vec::new()), output_schema: Bytes::from(Vec::new()), + parameter_default_values: None, stability: None, null_handling: None, description: String::new(), @@ -543,6 +544,47 @@ fn apply_metadata(fi: &mut FunctionInfo, meta: &FunctionMetadata) -> Result<()> Ok(()) } +fn apply_parameter_defaults( + fi: &mut FunctionInfo, + meta: &FunctionMetadata, + arguments: &Schema, +) -> Result<()> { + let Some(defaults) = &meta.parameter_default_values else { + return Ok(()); + }; + if defaults.num_rows() != 1 { + return Err(vgi_rpc::RpcError::value_error(format!( + "parameter_default_values must contain exactly one row, got {}", + defaults.num_rows() + ))); + } + let mut argument_index = 0; + for field in defaults.schema().fields() { + while argument_index < arguments.fields().len() + && arguments.field(argument_index).name() != field.name() + { + argument_index += 1; + } + if argument_index == arguments.fields().len() { + return Err(vgi_rpc::RpcError::value_error(format!( + "parameter_default_values field '{}' is not in argument signature order", + field.name() + ))); + } + if arguments.field(argument_index).data_type() != field.data_type() { + return Err(vgi_rpc::RpcError::type_error(format!( + "parameter_default_values field '{}' has type {:?}, expected {:?}", + field.name(), + field.data_type(), + arguments.field(argument_index).data_type() + ))); + } + argument_index += 1; + } + fi.parameter_default_values = Some(Bytes::from(ipc::write_batch(defaults)?)); + Ok(()) +} + /// Build the `FunctionInfo` for a scalar function. pub fn scalar_function_info(f: &dyn ScalarFunction) -> Result { let meta = f.metadata(); @@ -551,6 +593,7 @@ pub fn scalar_function_info(f: &dyn ScalarFunction) -> Result { let arg_schema = build_arg_schema(&f.argument_specs()); fi.arguments = Bytes::from(ipc::write_schema(&arg_schema)?); + apply_parameter_defaults(&mut fi, &meta, &arg_schema)?; // Scalar functions need a 1-field output schema for DuckDB. Use the fixed // return type if declared, else a `result: null` placeholder carrying the @@ -576,6 +619,7 @@ pub fn table_function_info(f: &dyn crate::table_function::TableFunction) -> Resu apply_metadata(&mut fi, &meta)?; let arg_schema = build_arg_schema(&f.argument_specs()); fi.arguments = Bytes::from(ipc::write_schema(&arg_schema)?); + apply_parameter_defaults(&mut fi, &meta, &arg_schema)?; // Output schema is resolved at bind time; advertise an empty schema. fi.output_schema = Bytes::from(ipc::write_schema(&Schema::empty())?); Ok(fi) @@ -595,6 +639,7 @@ pub fn table_in_out_function_info( fi.input_from_args = meta.input_from_args; let arg_schema = build_arg_schema(&f.argument_specs()); fi.arguments = Bytes::from(ipc::write_schema(&arg_schema)?); + apply_parameter_defaults(&mut fi, &meta, &arg_schema)?; fi.output_schema = Bytes::from(ipc::write_schema(&Schema::empty())?); Ok(fi) } @@ -609,6 +654,7 @@ pub fn buffering_function_info( fi.has_finalize = true; let arg_schema = build_arg_schema(&f.argument_specs()); fi.arguments = Bytes::from(ipc::write_schema(&arg_schema)?); + apply_parameter_defaults(&mut fi, &meta, &arg_schema)?; fi.output_schema = Bytes::from(ipc::write_schema(&Schema::empty())?); Ok(fi) } @@ -624,8 +670,10 @@ pub fn aggregate_function_info( apply_metadata(&mut fi, &meta)?; let arg_schema = build_arg_schema(&f.argument_specs()); fi.arguments = Bytes::from(ipc::write_schema(&arg_schema)?); + apply_parameter_defaults(&mut fi, &meta, &arg_schema)?; let params = crate::aggregate::AggregateBindParams { arguments: crate::arguments::Arguments::default(), + argument_names: None, input_schema: None, settings: crate::settings::Settings::default(), secrets: crate::secrets::Secrets::default(), diff --git a/vgi/src/dispatch.rs b/vgi/src/dispatch.rs index 3ecf793..4b84bfd 100644 --- a/vgi/src/dispatch.rs +++ b/vgi/src/dispatch.rs @@ -849,6 +849,7 @@ impl Dispatcher { Ok(BindParams { input_schema: opt_schema(&dto.input_schema)?, arguments: crate::arguments::Arguments::parse(&dto.arguments.0)?, + argument_names: dto.argument_names.clone(), settings: parse_settings(&dto.settings)?, secrets: parse_secrets(&dto.secrets)?, resolved_secrets_provided: dto.resolved_secrets_provided, @@ -3577,6 +3578,7 @@ impl Dispatcher { let _ = ctx; let params = AggregateBindParams { arguments: args, + argument_names: dto.argument_names, input_schema, settings: parse_settings(&dto.settings)?, // The C++ pre-resolves any advertised required secret and delivers @@ -4506,6 +4508,13 @@ fn boxed(req: &Request) -> Result { // (`row_limit`) the DuckDB extension never sends at all. let batch = if req.method == "init" { crate::protocol::dtos::backfill_init_request(batch)? + } else if req.method == "aggregate_bind" { + let names_type = arrow_schema::DataType::List(Arc::new(arrow_schema::Field::new( + "item", + arrow_schema::DataType::Utf8, + true, + ))); + crate::protocol::dtos::ensure_nullable_columns(batch, &[("argument_names", names_type)])? } else { batch }; diff --git a/vgi/src/function.rs b/vgi/src/function.rs index 8a9612f..3cc23e0 100644 --- a/vgi/src/function.rs +++ b/vgi/src/function.rs @@ -399,6 +399,9 @@ pub struct FunctionMetadata { pub tags: Vec<(String, String)>, /// Fixed scalar return type, when not computed dynamically at bind. pub return_type: Option, + /// Authoritative typed defaults. Exactly one row containing only defaulted + /// parameters in signature order; a present null is an explicit NULL. + pub parameter_default_values: Option, pub projection_pushdown: bool, pub filter_pushdown: bool, pub sampling_pushdown: bool, @@ -484,6 +487,7 @@ impl Default for FunctionMetadata { examples: Vec::new(), tags: Vec::new(), return_type: None, + parameter_default_values: None, projection_pushdown: false, filter_pushdown: false, sampling_pushdown: false, @@ -520,6 +524,9 @@ pub struct BindParams { pub input_schema: Option, /// Parsed call arguments (const values + positional types). pub arguments: crate::arguments::Arguments, + /// One entry per logical call argument. Inner `None` is an unnamed vararg; + /// outer `None` means the client could not provide names. + pub argument_names: Option>>, /// Parsed session settings. pub settings: crate::settings::Settings, /// Resolved secrets, when provided in a second-phase bind. diff --git a/vgi/src/function_shape_dispatch_tests.rs b/vgi/src/function_shape_dispatch_tests.rs index ecc7bc0..75b5ea6 100644 --- a/vgi/src/function_shape_dispatch_tests.rs +++ b/vgi/src/function_shape_dispatch_tests.rs @@ -214,6 +214,7 @@ fn init_body(function: &str, input_schema: Option<&Schema>, phase: Option<&str>) at_unit: None, at_value: None, schema_path: Some(vec![crate::catalog::MAIN_SCHEMA.to_string()]), + argument_names: Some(vec![Some("value".to_string())]), }; let bind_bytes = ipc::write_batch(&wire::to_batch(bind).unwrap()).unwrap(); let init = InitRequest { diff --git a/vgi/src/http_continuation_tests.rs b/vgi/src/http_continuation_tests.rs index b9d7c59..99e3328 100644 --- a/vgi/src/http_continuation_tests.rs +++ b/vgi/src/http_continuation_tests.rs @@ -213,6 +213,7 @@ fn init_body_with_filter( // these functions are registered without an explicit home, so they live // in the worker's own catalog under `main`. schema_path: Some(vec![crate::catalog::MAIN_SCHEMA.to_string()]), + argument_names: Some(vec![Some("count".to_string())]), }; let bind_bytes = ipc::write_batch(&wire::to_batch(bind).unwrap()).unwrap(); let init = InitRequest { @@ -629,6 +630,7 @@ fn bind_body(function: &str, path: &str, resolved_secrets_provided: bool) -> Vec // these functions are registered without an explicit home, so they live // in the worker's own catalog under `main`. schema_path: Some(vec![crate::catalog::MAIN_SCHEMA.to_string()]), + argument_names: Some(vec![Some("path".to_string())]), }; let inner = ipc::write_batch(&wire::to_batch(bind).unwrap()).unwrap(); let req_schema = Arc::new(Schema::new(vec![Field::new( @@ -770,6 +772,7 @@ fn finalize_init_body(function: &str) -> Vec { at_unit: None, at_value: None, schema_path: Some(vec![crate::catalog::MAIN_SCHEMA.to_string()]), + argument_names: Some(vec![Some("count".to_string())]), }; let bind_bytes = ipc::write_batch(&wire::to_batch(bind).unwrap()).unwrap(); let init = InitRequest {