vgi 0.19.0

// Copyright 2025, 2026 Query Farm LLC - https://query.farm

//! VGI wire DTOs. Field order and types follow the canonical wire schema
//! exactly (the C++ extension reads several result schemas positionally, so
//! order matters).
//!
//! Serialize/deserialize the flat ones with [`crate::wire`]; the catalog
//! item structs (`SchemaInfo`, `FunctionInfo`, …) are IPC-serialized into
//! the `items: list<binary>` result via [`crate::ipc`].

use std::sync::Arc;

use vgi_rpc::{Bytes, DictString, LargeBytes, Result, RpcError, VgiArrow};

/// `map<utf8, utf8>` payload (Python-canonical `keys`/`values` child names).
pub type StrMap = Vec<(String, String)>;
/// `map<utf8, int64>` payload.
pub type IntMap = Vec<(String, i64)>;

/// An optionally-inlined `int64` whose wire field is declared **non-nullable**
/// (the C++ extension's result-schema check requires `int64 not null`) yet may
/// carry a NULL value. The extension reads it via `row[...].as<int64_t>()`,
/// which yields `nullopt` for NULL — its signal for "not inlined, fire the RPC".
/// Matches the canonical convention (null in a non-nullable column).
#[derive(Debug, Clone, Copy, Default)]
pub struct InlineI64(pub Option<i64>);

impl From<Option<i64>> for InlineI64 {
    fn from(v: Option<i64>) -> Self {
        InlineI64(v)
    }
}

impl VgiArrow for InlineI64 {
    fn arrow_data_type() -> arrow_schema::DataType {
        arrow_schema::DataType::Int64
    }
    // Built nullable (so `arrow`'s StructArray accepts the NULL child), then
    // the serialized item's schema is tightened back to non-nullable by
    // `serialize_items` to satisfy the C++ schema check — matching Go/Python,
    // which emit NULL in a column the schema declares non-nullable.
    fn nullable() -> bool {
        true
    }
    fn describe_name() -> String {
        "int".into()
    }
    fn read(arr: &dyn arrow_array::Array, idx: usize) -> Result<Self> {
        if arr.is_null(idx) {
            return Ok(InlineI64(None));
        }
        if let Some(a) = arr.as_any().downcast_ref::<arrow_array::Int64Array>() {
            return Ok(InlineI64(Some(a.value(idx))));
        }
        if let Some(a) = arr.as_any().downcast_ref::<arrow_array::Int32Array>() {
            return Ok(InlineI64(Some(a.value(idx) as i64)));
        }
        Err(RpcError::type_error("expected Int64/Int32 array"))
    }
    fn build_singleton(value: Self) -> Result<arrow_array::ArrayRef> {
        Ok(Arc::new(arrow_array::Int64Array::from(vec![value.0])))
    }
}

// ---------------------------------------------------------------------------
// bind
// ---------------------------------------------------------------------------

/// `CopyFromContext` — the `COPY ... FROM` context threaded onto a
/// [`BindRequest`] when a bind/init opens a custom COPY-FROM scan. Present only
/// for COPY-FROM (the `copy_from` field is `None` otherwise). Encoded on the
/// wire as a nested **struct** column (not IPC binary), matching the Python
/// `vgi.protocol.CopyFromContext`. The handler's options arrive through the
/// normal [`BindRequest::arguments`], so they are not duplicated here.
#[derive(Debug, Clone, VgiArrow)]
pub struct CopyFromContext {
    /// The `FORMAT` name resolved at COPY bind time.
    pub format: String,
    /// The source path from the `COPY ... FROM 'path'` statement.
    pub file_path: String,
    /// IPC-serialized Arrow schema of the COPY target's columns (name + type,
    /// in target order). The worker must bind its output to, and emit columns
    /// whose types match, this schema exactly — DuckDB inserts no cast between
    /// the scan and the INSERT.
    pub expected_schema: Bytes,
}

/// `CopyToContext` — the `COPY ... TO` context threaded onto a [`BindRequest`]
/// when a bind/init opens a custom COPY-TO sink. Present only for COPY-TO (the
/// `copy_to` field is `None` otherwise). Encoded on the wire as a nested
/// **struct** column (not IPC binary), matching the Python
/// `vgi.protocol.CopyToContext`. The handler's options arrive through the normal
/// [`BindRequest::arguments`]; the **source** columns ride the existing
/// [`BindRequest::input_schema`], so neither is duplicated here.
#[derive(Debug, Clone, VgiArrow)]
pub struct CopyToContext {
    /// The `FORMAT` name resolved at COPY bind time.
    pub format: String,
    /// The destination path from the `COPY ... TO 'path'` statement.
    pub file_path: String,
}

/// `BindRequest` — carried IPC-serialized inside the `request` binary column
/// of `bind`, and as the nested `bind_call` struct of `init` / cardinality.
#[derive(Debug, Clone, VgiArrow)]
pub struct BindRequest {
    pub function_name: String,
    pub arguments: Bytes,
    pub function_type: DictString,
    pub input_schema: Option<Bytes>,
    pub settings: Option<Bytes>,
    pub secrets: Option<Bytes>,
    pub attach_opaque_data: Option<Bytes>,
    pub transaction_opaque_data: Option<Bytes>,
    pub resolved_secrets_provided: bool,
    /// Time travel: the `AT (TIMESTAMP|VERSION ...)` clause for this scan,
    /// threaded through from DuckDB's per-reference bind (both `None` when there
    /// is no AT clause). For inline-bound (function-backed) tables the actual
    /// `on_bind` RPC runs once at attach with no AT, so the per-scan AT is
    /// carried on the bind request embedded in each `InitRequest` and read at
    /// init. Additive nullable columns — the C++ always emits them.
    pub at_unit: Option<String>,
    pub at_value: 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
    // entirely for ordinary binds — matching Python's match-by-name /
    // default-None semantics). The `VgiArrow` derive would reject the missing
    // column, so they are read out-of-band via [`read_copy_from`] /
    // [`read_copy_to`] from the request batch. See `dispatch.rs`.
}

/// Read the optional `copy_from` nested-struct column from a (flat) BindRequest
/// batch. Returns `None` when the column is absent (ordinary bind) or null.
pub fn read_copy_from(batch: &arrow_array::RecordBatch) -> Result<Option<CopyFromContext>> {
    use arrow_array::Array;
    match batch.column_by_name("copy_from") {
        Some(col) if !col.is_empty() && !col.is_null(0) => {
            Ok(Some(<CopyFromContext as VgiArrow>::read(col.as_ref(), 0)?))
        }
        _ => Ok(None),
    }
}

/// Read the optional `copy_to` nested-struct column from a (flat) BindRequest
/// batch. Returns `None` when the column is absent (ordinary bind) or null.
pub fn read_copy_to(batch: &arrow_array::RecordBatch) -> Result<Option<CopyToContext>> {
    use arrow_array::Array;
    match batch.column_by_name("copy_to") {
        Some(col) if !col.is_empty() && !col.is_null(0) => {
            Ok(Some(<CopyToContext as VgiArrow>::read(col.as_ref(), 0)?))
        }
        _ => Ok(None),
    }
}

/// `BindResponse` — flat result of `bind`.
#[derive(Debug, Clone, VgiArrow)]
pub struct BindResponse {
    pub output_schema: Bytes,
    pub opaque_data: Bytes,
    pub lookup_secret_types: Vec<String>,
    pub lookup_scopes: Vec<String>,
    pub lookup_names: Vec<String>,
}

// ---------------------------------------------------------------------------
// init
// ---------------------------------------------------------------------------

/// `InitRequest` — carried IPC-serialized inside the `request` binary column
/// of `init`. `bind_call` is itself an IPC-serialized [`BindRequest`] (Python
/// serializes nested dataclasses as binary), decoded separately.
#[derive(Debug, Clone, VgiArrow)]
pub struct InitRequest {
    pub bind_call: Bytes,
    pub output_schema: Bytes,
    pub bind_opaque_data: Option<Bytes>,
    pub projection_ids: Option<Vec<i64>>,
    pub pushdown_filters: Option<LargeBytes>,
    pub join_keys: Option<Vec<LargeBytes>>,
    pub phase: Option<DictString>,
    pub execution_id: Option<Bytes>,
    pub init_opaque_data: Option<Bytes>,
    pub order_by_column_name: Option<String>,
    pub order_by_direction: Option<DictString>,
    pub order_by_null_order: Option<DictString>,
    pub order_by_limit: Option<i64>,
    pub tablesample_percentage: Option<f64>,
    pub tablesample_seed: Option<i64>,
    pub finalize_state_id: Option<Bytes>,
}

/// `GlobalInitResponse` — the streaming header for `init`.
#[derive(Debug, Clone, VgiArrow)]
pub struct GlobalInitResponse {
    pub execution_id: Bytes,
    pub max_workers: i64,
    pub opaque_data: Option<Bytes>,
}

// ---------------------------------------------------------------------------
// catalog_attach
// ---------------------------------------------------------------------------

/// `CatalogAttachRequest` — IPC-serialized inside `request` of `catalog_attach`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogAttachRequest {
    pub name: String,
    pub options: Option<Bytes>,
    pub data_version_spec: Option<String>,
    pub implementation_version: Option<String>,
}

/// Request for `table_function_cardinality` / `table_function_statistics`
/// (boxes an IPC-serialized `BindRequest`).
#[derive(Debug, Clone, VgiArrow)]
pub struct CardinalityRequest {
    pub bind_call: Bytes,
    pub bind_opaque_data: Option<Bytes>,
}

/// Response for `table_function_cardinality`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CardinalityResponse {
    pub estimate: Option<i64>,
    pub max: Option<i64>,
}

/// `aggregate_window_init` — caches a partition for windowed evaluation.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateWindowInitRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub partition_id: i64,
    pub row_count: i64,
    pub partition_batch: Bytes,
    pub output_schema: Bytes,
    pub filter_mask: Option<Bytes>,
    pub frame_stats: Option<Bytes>,
    pub all_valid: Option<Bytes>,
}

/// `aggregate_window` — evaluate one output row over its sub-frames.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateWindowRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub partition_id: i64,
    pub rid: i64,
    pub frame_starts: Vec<i64>,
    pub frame_ends: Vec<i64>,
}

/// `aggregate_window_batch` — evaluate `count` consecutive output rows.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateWindowBatchRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub partition_id: i64,
    pub row_idx: i64,
    pub count: i64,
    pub frames_per_row: Vec<i64>,
    pub frame_starts: Vec<i64>,
    pub frame_ends: Vec<i64>,
}

/// `aggregate_window` / `aggregate_window_batch` result.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateWindowResponse {
    pub result_batch: Bytes,
}

/// `aggregate_window_destructor` — drops a cached partition.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateWindowDestructorRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub partition_id: i64,
}

/// `table_function_dynamic_to_string` — post-execution profiling info.
#[derive(Debug, Clone, VgiArrow)]
pub struct DynamicToStringRequest {
    pub bind_call: Bytes,
    pub bind_opaque_data: Option<Bytes>,
    pub global_execution_id: Bytes,
}

/// `table_function_dynamic_to_string` result — ordered key/value pairs
/// surfaced as Extra Info under EXPLAIN ANALYZE.
#[derive(Debug, Clone, VgiArrow)]
pub struct DynamicToStringResponse {
    pub keys: Vec<String>,
    pub values: Vec<String>,
}

/// `aggregate_streaming_open` — start a streaming-partitioned session.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateStreamingOpenRequest {
    pub function_name: String,
    pub arguments: Bytes,
    pub input_schema: Bytes,
    pub partition_key_count: i64,
    pub order_key_count: i64,
    pub output_schema: Bytes,
    pub settings: Option<Bytes>,
    pub secrets: Option<Bytes>,
    pub attach_opaque_data: Option<Bytes>,
}

/// `aggregate_streaming_open` result — the session token.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateStreamingOpenResponse {
    pub execution_id: Bytes,
}

/// `aggregate_streaming_chunk` — process one input chunk in a session.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateStreamingChunkRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub input_batch: Bytes,
    pub attach_opaque_data: Option<Bytes>,
}

/// `aggregate_streaming_chunk` result — a same-length output batch.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateStreamingChunkResponse {
    pub result_batch: Bytes,
}

/// `aggregate_streaming_close` — end a session and free its state.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateStreamingCloseRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub attach_opaque_data: Option<Bytes>,
}

/// One physical source backing a (possibly multi-branch) table scan.
///
/// A branch is either a *function* branch (`function_name` names a table
/// function) or a *catalog-table* branch (`function_name` is empty and
/// `source_table` is set — it scans `source_catalog.source_schema.source_table`
/// in a companion catalog).
#[derive(Debug, Clone, VgiArrow)]
pub struct ScanBranch {
    pub function_name: String,
    pub arguments: Bytes,
    pub branch_filter: Option<String>,
    pub writable: bool,
    pub source_catalog: Option<String>,
    pub source_schema: Option<String>,
    pub source_table: Option<String>,
}

/// Response for `catalog_table_scan_branches_get`. The `branches` list must be
/// non-empty (one entry per physical source; single-source tables return one).
#[derive(Debug, Clone, VgiArrow)]
pub struct ScanBranchesResult {
    pub branches: Vec<Bytes>,
    pub required_extensions: Vec<String>,
}

/// Secret-type registration entry, IPC-serialized into
/// `CatalogAttachResult.secret_types`.
#[derive(Debug, Clone, VgiArrow)]
pub struct SecretTypeWire {
    pub name: String,
    pub description: String,
    pub parameters_schema: Bytes,
}

/// A companion catalog the client should ATTACH when this VGI catalog attaches
/// (lakehouse federation). IPC-serialized into `CatalogAttachResult.attach_catalogs`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AttachCatalogInfo {
    pub alias: String,
    pub target: String,
    pub db_type: String,
    pub options: StrMap,
    pub hidden: bool,
    pub required: bool,
    pub secret_ref: String,
}

/// `CatalogAttachResult` — flat result of `catalog_attach`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogAttachResult {
    pub attach_opaque_data: Bytes,
    pub supports_transactions: bool,
    pub supports_time_travel: bool,
    pub catalog_version_frozen: bool,
    pub catalog_version: i64,
    pub attach_opaque_data_required: bool,
    pub default_schema: String,
    pub settings: Vec<Bytes>,
    pub secret_types: Vec<Bytes>,
    pub attach_catalogs: Vec<Bytes>,
    pub comment: Option<String>,
    pub tags: StrMap,
    pub supports_column_statistics: bool,
    pub resolved_data_version: Option<String>,
    pub resolved_implementation_version: Option<String>,
}

// ---------------------------------------------------------------------------
// catalog transactions / version / detach
// ---------------------------------------------------------------------------

/// Flat params: `catalog_transaction_begin`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogTransactionBeginParams {
    pub attach_opaque_data: Bytes,
}

/// Flat result: `catalog_transaction_begin`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogTransactionBeginResult {
    pub transaction_opaque_data: Option<Bytes>,
}

/// Flat params: `catalog_transaction_commit` / `_rollback`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogTransactionEndParams {
    pub attach_opaque_data: Bytes,
    pub transaction_opaque_data: Bytes,
}

/// Flat params: `catalog_detach`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogDetachParams {
    pub attach_opaque_data: Bytes,
}

/// Flat params: `catalog_version`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogVersionParams {
    pub attach_opaque_data: Bytes,
    pub transaction_opaque_data: Option<Bytes>,
}

/// Flat result: `catalog_version`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogVersionResult {
    pub version: i64,
}

// ---------------------------------------------------------------------------
// catalog schema discovery
// ---------------------------------------------------------------------------

/// Flat params: `catalog_schemas`.
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogSchemasParams {
    pub attach_opaque_data: Bytes,
    pub transaction_opaque_data: Option<Bytes>,
}

/// Flat params: `catalog_schema_get` and the `catalog_schema_contents_{tables,
/// views,indexes}` family (attach + name + optional tx).
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogSchemaNameParams {
    pub attach_opaque_data: Bytes,
    pub name: String,
    pub transaction_opaque_data: Option<Bytes>,
}

/// Flat params: `catalog_schema_contents_functions` (and `_macros`).
#[derive(Debug, Clone, VgiArrow)]
pub struct CatalogSchemaContentsFunctionsParams {
    pub attach_opaque_data: Bytes,
    pub name: String,
    #[allow(non_snake_case)]
    pub r#type: DictString,
    pub transaction_opaque_data: Option<Bytes>,
}

/// Shared flat result for every `catalog_*_get` / `_contents_*` method:
/// `items` is a list of IPC-serialized item structs.
#[derive(Debug, Clone, VgiArrow)]
pub struct ItemsResult {
    pub items: Vec<Bytes>,
}

// ---------------------------------------------------------------------------
// catalog item structs (IPC-serialized into ItemsResult.items)
// ---------------------------------------------------------------------------

/// `SchemaInfo` item.
#[derive(Debug, Clone, VgiArrow)]
pub struct SchemaInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    pub attach_opaque_data: Bytes,
    pub name: String,
    pub estimated_object_count: Option<IntMap>,
}

/// One `examples` entry of `FunctionInfo`.
#[derive(Debug, Clone, VgiArrow)]
pub struct FunctionExample {
    pub sql: String,
    pub description: String,
    pub expected_output: Option<String>,
}

/// One `required_secrets` entry of `FunctionInfo`.
#[derive(Debug, Clone, VgiArrow)]
pub struct RequiredSecret {
    pub secret_type: String,
    pub scope: Option<String>,
    pub secret_name: Option<String>,
}

/// `TableInfo` item — describes a catalog table to DuckDB.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    pub name: String,
    pub schema_name: String,
    /// IPC-serialized Arrow schema of the table columns.
    pub columns: Bytes,
    pub not_null_constraints: Vec<i32>,
    pub unique_constraints: Vec<Vec<i32>>,
    pub check_constraints: Vec<String>,
    pub primary_key_constraints: Vec<Vec<i32>>,
    pub foreign_key_constraints: Vec<Bytes>,
    pub supports_insert: bool,
    pub supports_update: bool,
    pub supports_delete: bool,
    pub supports_returning: bool,
    pub supports_column_statistics: bool,
    /// IPC `ScanFunctionResult`, empty if not inlined.
    pub scan_function: Bytes,
    pub insert_function: Bytes,
    pub update_function: Bytes,
    pub delete_function: Bytes,
    /// Inlined optimizer cardinality. `None` → NULL on the wire, which the C++
    /// reads as "not inlined" → it fires the per-bind `table_function_cardinality`
    /// RPC instead. Emitting a sentinel (e.g. -1) would be read as an inlined
    /// value and wrongly skip the RPC.
    pub cardinality_estimate: InlineI64,
    pub cardinality_max: InlineI64,
    pub column_statistics: Bytes,
    pub bind_result: Bytes,
    /// Required WHERE-filter groups in conjunctive normal form (CNF): an AND
    /// (outer list) of OR-groups (inner lists) of dotted-path column references.
    /// A group is satisfied when any one of its member paths has a WHERE filter;
    /// every group must be satisfied. Empty = no enforcement. Wire type
    /// `list<list<utf8>>`. Trailing field of the C++ protocol's TableInfo.
    pub required_filters: Vec<Vec<String>>,
}

/// `ViewInfo` item.
#[derive(Debug, Clone, VgiArrow)]
pub struct ViewInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    pub name: String,
    pub schema_name: String,
    pub definition: String,
    pub column_comments: StrMap,
}

/// `MacroInfo` item.
#[derive(Debug, Clone, VgiArrow)]
pub struct MacroInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    pub name: String,
    pub schema_name: String,
    pub macro_type: DictString,
    pub parameters: Vec<String>,
    pub parameter_default_values: Bytes,
    pub definition: String,
    /// Optional Arrow schema (IPC bytes) with one nullable field per parameter,
    /// in `parameters` order. Each field's type is the parameter's default value
    /// type when known (else null), and the `vgi_doc` field metadata key carries
    /// the parameter's description (UTF-8, presence-only — omitted when
    /// undocumented). Mirrors the per-argument doc channel functions expose via
    /// `FunctionInfo.arguments`. Empty means the worker supplied no per-parameter
    /// docs (older workers); readers fall back to `parameters` for names. Appended
    /// last for consistency with the reference wire contract.
    pub arguments_schema: Bytes,
}

/// `ScanFunctionResult` — names the table function that scans a catalog table.
#[derive(Debug, Clone, VgiArrow)]
pub struct ScanFunctionResult {
    pub function_name: String,
    pub arguments: Bytes,
    pub required_extensions: Vec<String>,
}

/// `FunctionInfo` item — describes a function to DuckDB.
#[derive(Debug, Clone, VgiArrow)]
pub struct FunctionInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    pub name: String,
    pub schema_name: String,
    pub function_type: DictString,
    pub arguments: Bytes,
    pub output_schema: Bytes,
    pub stability: Option<DictString>,
    pub null_handling: Option<DictString>,
    pub description: String,
    pub examples: Vec<FunctionExample>,
    pub categories: Vec<String>,
    pub projection_pushdown: Option<bool>,
    pub filter_pushdown: Option<bool>,
    pub sampling_pushdown: Option<bool>,
    pub late_materialization: Option<bool>,
    pub supported_expression_filters: Vec<String>,
    pub order_preservation: Option<DictString>,
    pub max_workers: i32,
    pub supports_batch_index: bool,
    pub partition_kind: DictString,
    pub order_dependent: DictString,
    pub distinct_dependent: DictString,
    pub supports_window: bool,
    pub streaming_partitioned: bool,
    pub has_finalize: bool,
    pub source_order_dependent: bool,
    pub sink_order_dependent: bool,
    pub requires_input_batch_index: bool,
    pub required_settings: Vec<String>,
    pub required_secrets: Vec<RequiredSecret>,
}

/// `CopyFromFormatInfo` item — a custom `COPY ... FROM` format advertised by a
/// catalog via `catalog_copy_from_formats`. Field order/names mirror the Python
/// `vgi.catalog.catalog_interface.CopyFromFormatInfo` (inherits `comment` /
/// `tags` from `CatalogObject`). The C++ extension reads each field by name.
#[derive(Debug, Clone, VgiArrow)]
pub struct CopyFromFormatInfo {
    pub comment: Option<String>,
    pub tags: StrMap,
    /// The `FORMAT` identifier users type (e.g. `example_lines`).
    pub format_name: String,
    /// Registered name of the worker function that performs the read.
    pub handler: String,
    /// IPC-serialized Arrow schema of the format's options (same encoding as
    /// `FunctionInfo.arguments`); each field's metadata carries the option
    /// type / `vgi_doc` description.
    pub options: Bytes,
    /// `"from"` | `"to"` | `"both"` — which COPY direction(s) this format
    /// serves. COPY-FROM readers advertise `"from"`; COPY-TO writers `"to"`.
    pub direction: String,
    /// Intrinsic documentation from the handler's metadata description.
    pub description: String,
    /// COPY ... TO only — when true the writer requires rows in source order, so
    /// the extension installs a single-thread sink. Set from a
    /// [`CopyToFunction::ordered`](crate::copy_to::CopyToFunction::ordered)
    /// (`Meta.sink_order_dependent` in the Python model). Always `false` for
    /// COPY-FROM readers. Field order (after `description`) matches the canonical
    /// wire schema the C++ extension validates positionally.
    pub ordered: bool,
}

// ---------------------------------------------------------------------------
// cardinality / statistics
// ---------------------------------------------------------------------------

/// Flat result: `table_function_cardinality`.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableCardinality {
    pub estimate: Option<i64>,
    pub max: Option<i64>,
}

// ---------------------------------------------------------------------------
// table buffering (boxed in `request: binary`)
// ---------------------------------------------------------------------------

/// `TableBufferingProcessRequest` — sink one batch.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableBufferingProcessRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub input_batch: Bytes,
    pub attach_opaque_data: Option<Bytes>,
    pub transaction_id: Option<Bytes>,
    pub batch_index: Option<i64>,
}

/// `TableBufferingProcessResponse`.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableBufferingProcessResponse {
    pub state_id: Bytes,
}

/// `TableBufferingCombineRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableBufferingCombineRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub state_ids: Vec<Bytes>,
    pub attach_opaque_data: Option<Bytes>,
    pub transaction_id: Option<Bytes>,
}

/// `TableBufferingCombineResponse`.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableBufferingCombineResponse {
    pub finalize_state_ids: Vec<Bytes>,
}

/// `TableBufferingDestructorRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct TableBufferingDestructorRequest {
    pub function_name: String,
    pub execution_id: Bytes,
}

// ---------------------------------------------------------------------------
// aggregate (boxed in `request: binary`)
// ---------------------------------------------------------------------------

/// `AggregateBindRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateBindRequest {
    pub function_name: String,
    pub arguments: Bytes,
    pub input_schema: Option<Bytes>,
    pub settings: Option<Bytes>,
    pub secrets: Option<Bytes>,
    pub attach_opaque_data: Option<Bytes>,
}

/// `AggregateBindResponse`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateBindResponse {
    pub output_schema: Bytes,
    pub execution_id: Bytes,
}

/// `AggregateUpdateRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateUpdateRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub input_batch: Bytes,
    pub attach_opaque_data: Option<Bytes>,
}

/// `AggregateCombineRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateCombineRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub merge_batch: Bytes,
    pub attach_opaque_data: Option<Bytes>,
}

/// `AggregateFinalizeRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateFinalizeRequest {
    pub function_name: String,
    pub execution_id: Bytes,
    pub group_ids_batch: Bytes,
    pub output_schema: Bytes,
    pub attach_opaque_data: Option<Bytes>,
}

/// `AggregateFinalizeResponse`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateFinalizeResponse {
    pub result_batch: Bytes,
}

/// `AggregateDestructorRequest`.
#[derive(Debug, Clone, VgiArrow)]
pub struct AggregateDestructorRequest {
    pub function_name: String,
    pub execution_id: Bytes,
}

#[cfg(test)]
mod federation_tests {
    use super::*;

    #[test]
    fn scan_branch_catalog_table_roundtrips() {
        let b = ScanBranch {
            function_name: String::new(),
            arguments: Bytes::from(Vec::<u8>::new()),
            branch_filter: Some("id < 100".to_string()),
            writable: false,
            source_catalog: Some("acme_lake".to_string()),
            source_schema: Some("main".to_string()),
            source_table: Some("events".to_string()),
        };
        let batch = crate::wire::to_batch(b).unwrap();
        let back: ScanBranch = crate::wire::from_batch(&batch).unwrap();
        assert_eq!(back.function_name, "");
        assert_eq!(back.branch_filter.as_deref(), Some("id < 100"));
        assert_eq!(back.source_catalog.as_deref(), Some("acme_lake"));
        assert_eq!(back.source_schema.as_deref(), Some("main"));
        assert_eq!(back.source_table.as_deref(), Some("events"));
    }

    #[test]
    fn attach_catalog_info_roundtrips() {
        let info = AttachCatalogInfo {
            alias: "acme_lake".to_string(),
            target: "ducklake:sqlite:/data/meta.sqlite".to_string(),
            db_type: String::new(),
            options: vec![("DATA_PATH".to_string(), "/data/".to_string())],
            hidden: true,
            required: true,
            secret_ref: "pg".to_string(),
        };
        let batch = crate::wire::to_batch(info).unwrap();
        let back: AttachCatalogInfo = crate::wire::from_batch(&batch).unwrap();
        assert_eq!(back.alias, "acme_lake");
        assert_eq!(back.target, "ducklake:sqlite:/data/meta.sqlite");
        assert_eq!(
            back.options,
            vec![("DATA_PATH".to_string(), "/data/".to_string())]
        );
        assert!(back.hidden);
        assert!(back.required);
        assert_eq!(back.secret_ref, "pg");
    }
}