Add VGI 2 function signature metadata

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(),

vgi-client/src/exchange.rs

@@ -194,7 +194,9 @@ fn parse_parent_rows(md: &vgi_rpc::wire::Metadata) -> Result<Option<Vec<i32>>> {
         )));
     }
     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(

vgi-client/src/scan.rs

@@ -87,6 +87,9 @@ pub struct BindSpec {
     pub schema_path: Option<Vec<String>>,
     /// 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<Vec<Option<String>>>,
     /// 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<I, S>(mut self, names: I) -> Self
+    where
+        I: IntoIterator<Item = Option<S>>,
+        S: Into<String>,
+    {
+        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

vgi-example-worker/src/aggregate/mod.rs

@@ -502,7 +502,9 @@ fn pct_push(state: &[u8], v: f64) -> Vec<u8> {
 }
 fn pct_vals(state: &[u8]) -> Vec<f64> {
     state
-        .chunks_exact(8)
+        .as_chunks::<8>()
+        .0
+        .iter()
         .map(|c| {
             let mut a = [0u8; 8];
             a.copy_from_slice(c);

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<ArgSpec> {
+        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<BindResponse> {
+        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<RecordBatch> {
+        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::<arrow_array::types::Int64Type>();
+        let right = right.as_primitive::<arrow_array::types::Int64Type>();
+        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
 // ---------------------------------------------------------------------------

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(),

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()

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<SchemaPath>,
+    /// 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<Vec<Option<String>>>,
     // 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<arrow_a
     let mut columns = batch.columns().to_vec();
     let mut appended = false;
 
+    if batch.column_by_name("parameter_default_values").is_none() {
+        appended = true;
+        fields.push(arrow_schema::Field::new(
+            "parameter_default_values",
+            arrow_schema::DataType::Binary,
+            true,
+        ));
+        columns.push(Arc::new(arrow_array::BinaryArray::new_null(rows)));
+    }
+
     for name in [
         "supports_splits",
         "filters_exactly_applied",
@@ -389,13 +410,17 @@ mod backfill_tests {
         assert!(col.is_null(0));
     }
 
-    /// A 1.1.0 peer already sends the column; nothing is synthesised and the
-    /// batch is handed back untouched.
+    /// A 1.1.0 peer already sends schema_path; only the newer argument-names
+    /// column is synthesized, without changing the legacy-peer classification.
     #[test]
     fn present_column_is_left_alone() {
         let (out, legacy) = backfill_bind_request(batch(true)).expect("backfill");
         assert!(!legacy);
-        assert_eq!(out.num_columns(), 2);
+        assert_eq!(out.num_columns(), 3);
+        assert!(out
+            .column_by_name("argument_names")
+            .expect("column added")
+            .is_null(0));
     }
 
     #[test]
@@ -1224,6 +1249,9 @@ pub struct FunctionInfo {
     pub function_type: DictString,
     pub arguments: Bytes,
     pub output_schema: Bytes,
+    /// 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<Bytes>,
     pub stability: Option<DictString>,
     pub null_handling: Option<DictString>,
     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<SchemaPath>,
+    /// Full logical argument order; inner `None` denotes an unnamed vararg.
+    pub argument_names: Option<Vec<Option<String>>>,
 }
 
 /// `AggregateBindResponse`.

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<Vec<Option<String>>>,
     pub input_schema: Option<SchemaRef>,
     pub settings: Settings,
     /// Statically pre-resolved secrets, delivered on `AggregateBindRequest.secrets`