polydat_core/library/

vectors.rs

// Copyright 2024-2026 Jonathan Shook
// SPDX-License-Identifier: Apache-2.0

//! Vector dataset access nodes via the `vectordata` crate.
//!
//! Each node takes a dataset source string (URL, local path, or
//! catalog name) as a const parameter and loads the dataset handle at
//! construction time.
//!
//! Source specifier formats:
//! - `"dataset"` — catalog lookup, uses default profile
//! - `"dataset:profile"` — catalog lookup with explicit profile
//! - `"https://..."` — direct URL
//! - `"/path/to/dir"` — local filesystem path
//!
//! ## Prebuffering
//!
//! Use `dataset_prebuffer("source")` to eagerly download all facets
//! for a dataset before workload execution. After prebuffering, data
//! access uses local mmap readers (zero HTTP overhead).
//!
//! ## Cache-aware loading
//!
//! For catalog-resolved datasets, the loader checks the local cache
//! at `~/.cache/vectordata/<dataset>/` before issuing HTTP requests.
//! Prebuffering populates this cache; subsequent loads are local.
//!
//! Feature-gated behind `vectordata`.
//!
//! ## Upstream API reference
//!
//! See the vectordata consumer API docs for the full dataset access,
//! catalog, caching, and prebuffer model:
//! <https://github.com/nosqlbench/vectordata-rs/blob/main/docs/sysref/02-api.md>

use std::sync::{Arc, LazyLock};

use crate::library::support::cache::OnceCache;

use crate::ast::{NodeMeta, PolydatNode, Port, PortType, Slot, Value};
use vectordata::TestDataGroup;
use vectordata::TestDataView;
use vectordata::catalog::resolver::Catalog;
use vectordata::catalog::sources::CatalogSources;
use vectordata::io::{VectorReader, VvecReader};

/// Global cache for loaded dataset groups keyed by source string.
/// Ensures each dataset is loaded exactly once regardless of how many
/// node functions reference it. The race-free init pattern lives in
/// [`crate::library::support::cache::OnceCache`].
static DATASET_CACHE: LazyLock<OnceCache<String, Arc<TestDataGroup>>> =
    LazyLock::new(OnceCache::new);

/// Type-erased facet cache: (source, profile, facet) → Arc<dyn Any + Send + Sync>.
/// Ensures each reader (of any element type) is opened exactly once
/// and shared across all node instances that reference the same data.
/// The concrete type inside is `Arc<UniformDataset<T>>` or `Arc<Ivvec32Dataset>`
/// or `Arc<GenericFacetDataset>`. The race-free init pattern lives
/// in [`OnceCache`].
type FacetCache = OnceCache<(String, String, String), Arc<dyn std::any::Any + Send + Sync>>;
static FACET_CACHE: LazyLock<FacetCache> = LazyLock::new(OnceCache::new);

/// Whole-dataset prebuffer cache keyed by source string. Wraps
/// [`do_dataset_prebuffer_inner`] so N fibers all calling
/// `dataset_prebuffer("ds:profile")` at init time serialize on
/// one OnceLock and share the resulting handle. Without this,
/// each fiber's own kernel-init pass invokes the real prebuffer
/// body and they all race into the manifest walk + per-facet
/// download — exactly the thundering herd we hit. The inner
/// caches (DATASET_CACHE, FACET_CACHE) only protect the
/// group-resolve and per-facet-reader steps, not the outer
/// "walk every facet and pull all chunks" work.
static PREBUFFER_CACHE: LazyLock<OnceCache<String, Arc<DatasetHandle>>> =
    LazyLock::new(OnceCache::new);

// =================================================================
// Dataset resolution — catalog-aware, cache-aware
// =================================================================

/// Parse a source specifier into (dataset_name, profile_name).
///
/// Supports `"dataset:profile"` syntax. If no colon is present,
/// the profile defaults to `"default"`.
fn parse_source_specifier(source: &str) -> (&str, &str) {
    // Don't split on colon in URLs
    if source.starts_with("http://") || source.starts_with("https://") {
        return (source, "default");
    }
    if let Some(pos) = source.find(':') {
        (&source[..pos], &source[pos + 1..])
    } else {
        (source, "default")
    }
}

/// Run a synchronous body that may internally drive
/// `reqwest::blocking` (and thus spin up a private tokio
/// runtime per HTTP request). If we're sitting on an outer
/// async runtime — we are, in every per-cycle and per-phase
/// path — the inner runtime panics on drop with "Cannot drop a
/// runtime in a context where blocking is not allowed".
/// `block_in_place` parks the outer multi-thread worker for
/// the duration of the call so the inner runtime sees a
/// non-async drop context. Falls back to a direct call when no
/// runtime is current (e.g. unit tests). Single helper used by
/// all three vectordata-facing entry points
/// ([`load_dataset_group`], [`load_uniform_facet`],
/// [`GenericFacetDataset::load`]).
fn run_blocking_io<R>(body: impl FnOnce() -> R) -> R {
    // tokio rides the `vectordata` feature: without the crate there is
    // no HTTP client to park a worker for.
    #[cfg(feature = "vectordata")]
    if tokio::runtime::Handle::try_current().is_ok() {
        return tokio::task::block_in_place(body);
    }
    body()
}

/// Load a dataset group by name.
///
/// Uses the vectordata catalog API: `catalog.open(name)` handles
/// catalog discovery, cache resolution, and download transparently.
pub(crate) fn load_dataset_group(source: &str) -> Result<Arc<TestDataGroup>, String> {
    let (dataset_name, _profile) = parse_source_specifier(source);
    // Both keys (dataset name and full source spec) point at the
    // same `Arc<TestDataGroup>`; the `OnceCache` slot for one is
    // primed by the other on first hit.
    DATASET_CACHE.get_or_init(dataset_name.to_string(), || {
        run_blocking_io(|| {
            let catalog = Catalog::of(&CatalogSources::new().configure_default());
            catalog
                .open(dataset_name)
                .map(Arc::new)
                .map_err(|e| format!("failed to load dataset '{dataset_name}': {e}"))
        })
    })
}

// =================================================================
// Dataset handles — loaded once at node construction, shared via Arc
// =================================================================

/// Generic handle to a loaded uniform vector facet. Thread-safe, random-access.
/// Supports any element type provided by the vectordata API (f32, f64,
/// i32, i16, u8, i8, u16, u32, u64, i64, f16).
pub(crate) struct UniformDataset<T: Send + Sync + 'static> {
    reader: Arc<dyn VectorReader<T>>,
    count: usize,
    dim: usize,
}

/// Cache-aware loader for uniform vector facets.
/// Returns a shared `Arc<UniformDataset<T>>`, creating and caching it
/// on first access. Subsequent loads for the same (source, profile, facet)
/// return the cached instance.
fn load_uniform_facet<T: Send + Sync + 'static>(
    source: &str,
    profile: &str,
    facet: &str,
    open_fn: impl FnOnce(
        &dyn TestDataView,
    ) -> std::result::Result<Arc<dyn VectorReader<T>>, vectordata::Error>,
) -> Result<Arc<UniformDataset<T>>, String> {
    let key = (source.to_string(), profile.to_string(), facet.to_string());
    let any = FACET_CACHE.get_or_init(key, || {
        let group = load_dataset_group(source)?;
        let view = group
            .profile(profile)
            .ok_or_else(|| format!("profile '{profile}' not found in '{source}'"))?;
        // Audit: log the open *before* `open_fn` runs so the
        // line appears even if the open errors. Inside
        // `get_or_init`'s closure, this fires exactly once per
        // (source, profile, facet) — the prior shape logged
        // once per concurrent miss (storms of N for N fibers).
        crate::library::support::audit::record_opened(source, profile, facet, "uniform");
        let reader = run_blocking_io(|| open_fn(view.as_ref()))
            .map_err(|e| format!("failed to access {facet} from '{source}': {e}"))?;
        let count = reader.count();
        let dim = reader.dim();
        let arc: Arc<UniformDataset<T>> = Arc::new(UniformDataset { reader, count, dim });
        Ok(arc as Arc<dyn std::any::Any + Send + Sync>)
    })?;
    any.downcast::<UniformDataset<T>>().map_err(|_| {
        format!(
            "facet cache type mismatch for '{source}:{profile}/{facet}' — \
             this should be impossible; please file a bug."
        )
    })
}

// Type aliases for backward compatibility
type F32Dataset = UniformDataset<f32>;
type I32Dataset = UniformDataset<i32>;

impl F32Dataset {
    fn load(source: &str, profile: &str, facet: &str) -> Result<Arc<Self>, String> {
        let facet_name = facet.to_string();
        load_uniform_facet(source, profile, facet, move |view| {
            match facet_name.as_str() {
                "base" => view.base_vectors(),
                "query" => view.query_vectors(),
                "neighbor_distances" => view.neighbor_distances(),
                "filtered_neighbor_distances" => view.prefiltered_neighbor_distances(),
                other => Err(vectordata::Error::MissingFacet(format!(
                    "unknown f32 facet: '{other}'"
                ))),
            }
        })
    }
}

impl I32Dataset {
    fn load(source: &str, profile: &str, facet: &str) -> Result<Arc<Self>, String> {
        let facet_name = facet.to_string();
        load_uniform_facet(source, profile, facet, move |view| {
            match facet_name.as_str() {
                "neighbor_indices" => view.neighbor_indices(),
                "filtered_neighbor_indices" => view.prefiltered_neighbor_indices(),
                other => Err(vectordata::Error::MissingFacet(format!(
                    "unknown i32 facet: '{other}'"
                ))),
            }
        })
    }
}

// =================================================================
// Dataset handle — typed enum wrapping all per-facet dataset shapes
// =================================================================
//
// Per SRD 53 §"Dataset Handles": `dataset_open(source, facet)` is
// the resolver node; per-cycle accessors take a handle wire and
// downcast to the concrete variant they expect. A single Value
// variant `Value::Handle(Arc<dyn Any>)` carries any of the
// concrete shapes — the enum below is what's actually inside the
// Arc, and accessors `match` on the variant.
//
// One handle may be opened against multiple facets at the
// `dataset_open` boundary (the user calls `dataset_open(spec,
// "base")` vs. `dataset_open(spec, "query")` and gets two
// distinct handles); the downstream accessor matches on whatever
// variant came back.

/// Typed wrapper around a resolved dataset facet or group. Held
/// inside `Value::Handle` as `Arc<DatasetHandle>` and downcast by
/// accessor nodes via [`Value::as_handle::<DatasetHandle>()`].
/// Cloning a `Value::Handle` is one `Arc::clone` — one atomic
/// increment, no allocation — which is the design contract this
/// enum exists to satisfy.
#[derive(Clone)]
pub(crate) enum DatasetHandle {
    /// Uniform `f32` vector facet (base, query, neighbor_distances,
    /// filtered_neighbor_distances).
    F32(Arc<F32Dataset>),
    /// Uniform `i32` vector facet (neighbor_indices, filtered_neighbor_indices).
    I32(Arc<I32Dataset>),
    /// Variable-length `i32` facet (metadata_results).
    Ivvec32(Arc<Ivvec32Dataset>),
    /// Type-erased generic-typed scalar facet (metadata_content,
    /// metadata_predicates, ...).
    Generic(Arc<GenericFacetDataset>),
    /// Dataset-group handle — used by group-level metadata
    /// accessors (`dataset_profile_count`, `dataset_facets`,
    /// `dataset_distance_function`, ...) that operate on the
    /// `TestDataGroup` before any profile/facet is chosen.
    Group(Arc<TestDataGroup>),
    /// Prebuffered-and-resident dataset, returned by
    /// `dataset_prebuffer(source)`. Carries both the group AND
    /// the source spec so per-facet accessors can re-resolve
    /// (`query_vector_at(prebuffered, q)` → resolves the
    /// `query` facet from `<source>:<profile>`). Distinct from
    /// `Group` so existing Group-only consumers stay typed.
    ///
    /// The `_group` field keeps the prebuffered `TestDataGroup`
    /// alive for the duration of the handle — vectordata's
    /// internal storage cache is keyed off the group instance,
    /// so dropping the group prematurely would force per-facet
    /// readers to re-open against transport. The field isn't
    /// read directly by accessors (they use `source` to re-open
    /// via `DATASET_CACHE`, which has the same group cached);
    /// the field's purpose is the lifetime extension.
    Prebuffered {
        _group: Arc<TestDataGroup>,
        source: String,
    },
}

impl DatasetHandle {
    fn open(source: &str, facet: &str) -> Result<Self, String> {
        let (_, profile) = parse_source_specifier(source);
        match facet {
            "base" | "query" | "neighbor_distances" | "filtered_neighbor_distances" => {
                F32Dataset::load(source, profile, facet).map(DatasetHandle::F32)
            }
            "neighbor_indices" | "filtered_neighbor_indices" => {
                I32Dataset::load(source, profile, facet).map(DatasetHandle::I32)
            }
            "metadata_results" => Ivvec32Dataset::load(source, profile).map(DatasetHandle::Ivvec32),
            // Anything else routes through GenericFacetDataset (typed
            // scalar reader), which covers metadata_content,
            // metadata_predicates, and any future scalar facet.
            _ => GenericFacetDataset::load(source, profile, facet).map(DatasetHandle::Generic),
        }
    }

    fn open_group(source: &str) -> Result<Self, String> {
        load_dataset_group(source).map(DatasetHandle::Group)
    }
}

/// Downcast a handle Value to the typed `DatasetHandle` enum.
///
/// Surfaces a clear, actionable panic when the upstream slot
/// holds `Value::None` — that's the canonical signal from a
/// failed `dataset_open` / `dataset_group_open` (catalog miss,
/// I/O failure, missing facet on disk, …). Without this
/// dedicated branch the user would see a generic
/// `expected Handle, got U64` panic from
/// [`Value::as_handle`] (None reports U64 as its placeholder
/// port type), which is six layers removed from the actual
/// fault. The engine's `enrich_eval_panic` adds node
/// provenance on top of whichever message we panic with here.
fn handle_of(v: &Value) -> &DatasetHandle {
    if matches!(v, Value::None) {
        panic!(
            "dataset handle is None — an upstream dataset_open / \
             dataset_group_open failed to resolve. Check the audit \
             log for the underlying open error (catalog miss, missing \
             facet on disk, or transport failure). The dataset's name \
             is in the audit message; this is the most common fault \
             when running a workload on a system whose vectordata \
             catalog isn't configured for the requested dataset."
        );
    }
    v.as_handle::<DatasetHandle>()
}

/// `dataset_open(source: str, facet: str) -> Handle`
///
/// The single resolver node. Provenance follows its two wire
/// inputs; when both are scope-extern constants (the iter-var
/// case), this evaluates exactly once at iteration entry and
/// stays cached for every cycle in that iteration. When source
/// or facet is cycle-time, it re-evaluates accordingly.
///
/// All per-cycle accessors take the resulting handle on a wire
/// — the catalog/HTTP/mmap path is never on the cycle hot path.
///
/// Resolve failures used to fall through silently as `Value::None`
/// so a downstream op wrapper could lift them. That pattern also
/// let comprehension clause evaluation degrade silently (catalog
/// miss → None → downstream `handle_of` panic → caught + swallowed
/// → literal-list fallback splits the spec on a comma → garbage
/// iter-var). We now panic with the underlying error: the engine's
/// `enrich_eval_panic` adds node provenance, `eval_const_expr`
/// traps the panic into `Result::Err`, and `evaluate_spec`
/// propagates it as a clean clause-level diagnostic.
#[crate::polydat_node(category = RealData)]
fn dataset_open(source: &str, facet: &str) -> Arc<DatasetHandle> {
    match DatasetHandle::open(source, facet) {
        Ok(h) => Arc::new(h),
        Err(e) => {
            let msg = format!("dataset_open: failed to resolve '{source}' facet='{facet}': {e}");
            crate::library::support::audit::error(&msg);
            panic!("{msg}");
        }
    }
}

/// `dataset_group_open(source: str) -> Handle`
///
/// Group-level resolver. Returns a handle wrapping `Arc<TestDataGroup>`
/// — used by group-level metadata accessors (`dataset_profile_count`,
/// `dataset_facets`, ...) that operate on the dataset as a whole
/// before any profile/facet is selected.
///
/// Hard-fail on resolve failure — see `dataset_open` for the
/// rationale. The Value::None pattern was a silent-degradation
/// source for comprehension clause evaluation.
#[crate::polydat_node(category = RealData)]
fn dataset_group_open(source: &str) -> Arc<DatasetHandle> {
    match DatasetHandle::open_group(source) {
        Ok(h) => Arc::new(h),
        Err(e) => {
            let msg = format!("dataset_group_open: failed to resolve '{source}': {e}");
            crate::library::support::audit::error(&msg);
            panic!("{msg}");
        }
    }
}

// =================================================================
// Base vector nodes
// =================================================================

// All indexed-accessor nodes share the same shape: an `index`
// wire (u64) and a `source` wire (Str), with the dataset
// resolved lazily on first eval per spec via `DATASET_CACHE`
// (inside `F32Dataset::load` / `I32Dataset::load`). The
// per-cycle hot path is one HashMap lookup on the cached spec
// plus the existing facet read; the spec doesn't change within
// an iteration scope so subsequent cycles hit the cache.

// =================================================================
// Per-cycle indexed accessors
// =================================================================
//
// All take `(handle: Handle, index: u64)`. The handle is resolved
// once per iteration by `dataset_open` (or by cursor sugar that
// emits an implicit one); per-cycle eval downcasts the handle's
// `Arc<DatasetHandle>` and reads at `index`. No HashMap, no Mutex,
// no String allocation on the source side. The `Bytes` outputs
// allocate a fresh per-cycle buffer — that's the remaining
// per-cycle alloc until SRD 46 (native vector binding).
//
// `expected_variant` documents which `DatasetHandle` variant is
// expected; a panic on wrong variant indicates the workload
// opened a handle for a different facet than the accessor expects.

macro_rules! handle_indexed_node {
    (
        $(#[$meta:meta])*
        $name:ident, $func_name:literal, $out_port:ident,
        facet = $facet:literal,
        eval = $eval_fn:expr
    ) => {
        $(#[$meta])*
        pub struct $name {
            meta: NodeMeta,
        }

        impl $name {
            /// A node of this kind.
            pub fn new() -> Self {
                Self {
                    meta: NodeMeta {
                        name: $func_name.into(),
                        outs: vec![Port::new("output", PortType::$out_port)],
                        ins: vec![
                            Slot::Wire(Port::handle("handle")),
                            Slot::Wire(Port::u64("index")),
                        ],
                    },
                }
            }
        }

        impl Default for $name {
            fn default() -> Self { Self::new() }
        }

        impl PolydatNode for $name {
            fn meta(&self) -> &NodeMeta { &self.meta }
            fn eval(&self, inputs: &[Value], outputs: &mut [Value]) {
                let handle = handle_of(&inputs[0]);
                // If the handle is `Prebuffered` (returned by
                // `dataset_prebuffer`), resolve to the specific
                // facet variant the accessor expects. For all
                // other handles this is a no-op borrow.
                let resolved = handle.resolve_facet($facet);
                let index = inputs[1].as_u64() as usize;
                outputs[0] = ($eval_fn)(resolved.as_ref(), index);
            }
        }
    };
}

/// Resolve a Prebuffered handle to a specific-facet handle by
/// opening the named facet via the existing FACET_CACHE path
/// (which is `OnceCache`-backed — concurrent callers serialize
/// per (source, profile, facet)). For non-Prebuffered handles
/// this is a borrow-through; the macro callers use the result as
/// `&DatasetHandle` regardless of which arm fires.
///
/// Lives on `DatasetHandle` so the resolution logic — and the
/// "what does Prebuffered mean to a per-facet accessor" question
/// — sits next to the variant declaration.
impl DatasetHandle {
    fn resolve_facet<'a>(&'a self, facet: &str) -> std::borrow::Cow<'a, DatasetHandle> {
        match self {
            DatasetHandle::Prebuffered { source, .. } => match DatasetHandle::open(source, facet) {
                Ok(opened) => std::borrow::Cow::Owned(opened),
                Err(e) => panic!(
                    "DatasetHandle::resolve_facet: failed to open \
                         '{facet}' from prebuffered '{source}': {e}"
                ),
            },
            _ => std::borrow::Cow::Borrowed(self),
        }
    }
}

fn f32_vec_at(h: &DatasetHandle, index: usize) -> Value {
    match h {
        DatasetHandle::F32(d) => Value::VecF32(slice_arc_from_uniform(d, index)),
        other => panic!(
            "expected F32 dataset handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

fn i32_vec_at(h: &DatasetHandle, index: usize) -> Value {
    match h {
        DatasetHandle::I32(d) => Value::VecI32(slice_arc_from_uniform(d, index)),
        other => panic!(
            "expected I32 dataset handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

fn ivvec32_vec_at(h: &DatasetHandle, index: usize) -> Value {
    match h {
        // Variable-length records — vectordata's trait doesn't expose
        // a zero-copy slice path for these (per-record dim is read
        // from the file), so we always allocate. One Vec<i32> per
        // cycle; same bound as the upstream trait API.
        DatasetHandle::Ivvec32(d) => {
            if d.count == 0 {
                return Value::VecI32(crate::ast::SliceArc::from_vec(Vec::<i32>::new()));
            }
            let v = d.reader.get(index % d.count).unwrap_or_default();
            Value::VecI32(crate::ast::SliceArc::from_vec(v))
        }
        other => panic!(
            "expected Ivvec32 dataset handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

/// Build a [`SliceArc<T>`] from any uniform-stride dataset at
/// `index`. Single generic helper consolidating what used to be
/// per-element-type slice-arc constructors — element type is
/// erased by the upstream `VectorReader<T>` trait, and `SliceArc`
/// is element-type-generic, so one body suffices for `f32`,
/// `i32`, and any other element type the trait supports.
///
/// Tries the zero-copy `VectorReader::get_slice` path first —
/// that returns a borrow into the mmap'd file pages, kept alive
/// by the `Arc<UniformDataset<T>>` that we move into the
/// `SliceArc` as owner. No allocation, no element decoding, no
/// copy.
///
/// Falls back to `VectorReader::get(idx) -> Vec<T>` for readers
/// that don't support zero-copy (HTTP-backed, or merkle-cached
/// storage that hasn't been promoted to mmap yet); that path
/// allocates one `Vec<T>` per cycle.
fn slice_arc_from_uniform<T>(d: &Arc<UniformDataset<T>>, index: usize) -> crate::ast::SliceArc<T>
where
    T: Send + Sync + Copy + 'static,
{
    if d.count == 0 {
        return crate::ast::SliceArc::from_vec(Vec::<T>::new());
    }
    let idx = index % d.count;
    if let Some(slice) = d.reader.get_slice(idx) {
        // SAFETY: the slice points into the mmap pages owned by
        // `d`'s reader; cloning `d` into the `SliceArc`'s owner
        // keeps the mmap alive for as long as the slice is held.
        let ptr_len = (slice.as_ptr(), slice.len());
        let owner = d.clone();
        let owner_dyn: Arc<dyn std::any::Any + Send + Sync> = owner;
        return unsafe {
            crate::ast::SliceArc::from_borrowed(
                owner_dyn,
                std::slice::from_raw_parts(ptr_len.0, ptr_len.1),
            )
        };
    }
    crate::ast::SliceArc::from_vec(d.reader.get(idx).unwrap_or_default())
}

fn dataset_handle_kind(h: &DatasetHandle) -> &'static str {
    match h {
        DatasetHandle::F32(_) => "F32",
        DatasetHandle::I32(_) => "I32",
        DatasetHandle::Ivvec32(_) => "Ivvec32",
        DatasetHandle::Generic(_) => "Generic",
        DatasetHandle::Group(_) => "Group",
        DatasetHandle::Prebuffered { .. } => "Prebuffered",
    }
}

/// Helper for group-level accessors: extract the `TestDataGroup`
/// from a `DatasetHandle::Group` variant. Panics on mismatch.
fn group_of(handle: &DatasetHandle) -> &TestDataGroup {
    match handle {
        DatasetHandle::Group(g) => g.as_ref(),
        other => panic!("expected Group handle, got {}", dataset_handle_kind(other)),
    }
}

handle_indexed_node!(
    /// Access an `f32` vector by index, returning a typed `VecF32`.
    /// Works on any F32 handle (base or query facet).
    ///
    /// Signature: `vector_at(handle, index: u64) -> VecF32`
    VectorAt, "vector_at", VecF32, facet = "base", eval = f32_vec_at
);

handle_indexed_node!(
    /// Access a query vector by index. Alias for [`VectorAt`] kept
    /// for clarity in workloads that distinguish base and query
    /// handles by name.
    ///
    /// Signature: `query_vector_at(handle, index: u64) -> VecF32`
    QueryVectorAt, "query_vector_at", VecF32, facet = "query", eval = f32_vec_at
);

handle_indexed_node!(
    /// Access ground-truth neighbor indices for a query. Expects an
    /// I32 handle.
    ///
    /// Signature: `neighbor_indices_at(handle, index: u64) -> VecI32`
    NeighborIndicesAt, "neighbor_indices_at", VecI32, facet = "neighbor_indices", eval = i32_vec_at
);

handle_indexed_node!(
    /// Access ground-truth neighbor distances for a query. Expects an
    /// F32 handle.
    ///
    /// Signature: `neighbor_distances_at(handle, index: u64) -> VecF32`
    NeighborDistancesAt, "neighbor_distances_at", VecF32, facet = "neighbor_distances", eval = f32_vec_at
);

handle_indexed_node!(
    /// Access filtered ground-truth neighbor indices. Expects an I32 handle.
    FilteredNeighborIndicesAt, "filtered_neighbor_indices_at", VecI32, facet = "filtered_neighbor_indices", eval = i32_vec_at
);

handle_indexed_node!(
    /// Access filtered ground-truth neighbor distances. Expects an F32 handle.
    FilteredNeighborDistancesAt, "filtered_neighbor_distances_at", VecF32, facet = "filtered_neighbor_distances", eval = f32_vec_at
);

// =================================================================
// Metadata nodes (constant per dataset)
// =================================================================

// =================================================================
// Per-handle metadata nodes
// =================================================================
//
// Take `(handle: Handle)` and read shape/metadata directly from the
// resolved dataset. Provenance bounded by the handle, which itself
// is bounded by the resolver's externs — so these collapse to a
// single per-iteration eval chain via the standard provenance
// caching. No per-cycle work.

macro_rules! handle_metadata_node {
    (
        $(#[$meta:meta])*
        $name:ident, $func_name:literal, $out_port:ident,
        eval = $eval_fn:expr
    ) => {
        $(#[$meta])*
        pub struct $name {
            meta: NodeMeta,
        }

        impl $name {
            /// A node of this kind.
            pub fn new() -> Self {
                Self {
                    meta: NodeMeta {
                        name: $func_name.into(),
                        outs: vec![Port::new("output", PortType::$out_port)],
                        ins: vec![Slot::Wire(Port::handle("handle"))],
                    },
                }
            }
        }

        impl Default for $name {
            fn default() -> Self { Self::new() }
        }

        impl PolydatNode for $name {
            fn meta(&self) -> &NodeMeta { &self.meta }
            fn eval(&self, inputs: &[Value], outputs: &mut [Value]) {
                let handle = handle_of(&inputs[0]);
                outputs[0] = ($eval_fn)(handle);
            }
        }
    };
}

handle_metadata_node!(
    /// Return the dimensionality (`f32` count per record) of a
    /// vector facet handle.
    ///
    /// Signature: `vector_dim(handle) -> (u64)`
    VectorDim, "vector_dim", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::F32(d) => Value::U64(d.dim as u64),
        DatasetHandle::I32(d) => Value::U64(d.dim as u64),
        _ => Value::U64(0),
    }
);

/// Return the dataset's distance function (e.g., "cosine", "euclidean").
///
/// Signature: `dataset_distance_function(source) -> (String)`
// Source-only nodes (just take a `source` String wire) all
// share the new shape: declare one Wire slot, read inputs[0]
// at eval, look up via the global cache, return the property.
macro_rules! source_only_node {
    (
        $(#[$meta:meta])*
        $name:ident, $func_name:literal,
        out_port = $out:ident,
        eval = |$src:ident| $body:expr
    ) => {
        $(#[$meta])*
        pub struct $name {
            meta: NodeMeta,
        }

        impl $name {
            /// A node of this kind.
            pub fn new() -> Self {
                Self {
                    meta: NodeMeta {
                        name: $func_name.into(),
                        outs: vec![Port::new("output", PortType::$out)],
                        ins: vec![Slot::Wire(Port::str("source"))],
                    },
                }
            }
        }

        impl Default for $name {
            fn default() -> Self { Self::new() }
        }

        impl PolydatNode for $name {
            fn meta(&self) -> &NodeMeta { &self.meta }
            fn eval(&self, inputs: &[Value], outputs: &mut [Value]) {
                let $src = inputs[0].as_str();
                outputs[0] = $body;
            }
        }
    };
}

// Helper: read a dataset-group attribute via a handle's underlying
// `TestDataGroup`. Used by `dataset_distance_function`. We cache
// the source string on the dataset handle itself to keep
// dataset-group attributes accessible without an extra lookup —
// but for now the simpler path is to require a separate
// dataset-group-level node that takes `source: str` (kept below).

handle_metadata_node!(
    /// Return the count of records in the facet a handle was opened
    /// against. This is the canonical "how many vectors / queries /
    /// neighbor-rows" accessor — `vector_count(base_handle)` for
    /// base vectors, `vector_count(query_handle)` for query
    /// vectors, etc.
    ///
    /// Signature: `vector_count(handle) -> (u64)`
    VectorCount, "vector_count", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::F32(d) => Value::U64(d.count as u64),
        DatasetHandle::I32(d) => Value::U64(d.count as u64),
        DatasetHandle::Ivvec32(d) => Value::U64(d.count as u64),
        DatasetHandle::Generic(d) => Value::U64(d.count as u64),
        DatasetHandle::Group(_) => panic!("vector_count: expected facet handle, got Group"),
        DatasetHandle::Prebuffered { source, .. } => {
            // Resolve the `base` facet (canonical interpretation
            // of `vector_count(prebuffered)` — base vectors).
            match DatasetHandle::open(source, "base") {
                Ok(DatasetHandle::F32(d)) => Value::U64(d.count as u64),
                Ok(other) => panic!(
                    "vector_count: expected F32 base facet, got {}",
                    dataset_handle_kind(&other)),
                Err(e) => panic!(
                    "vector_count: failed to open 'base' from prebuffered '{source}': {e}"),
            }
        }
    }
);

handle_metadata_node!(
    /// Alias for [`VectorCount`] kept for clarity in workloads that
    /// distinguish base/query handles by name.
    ///
    /// Signature: `query_count(handle) -> (u64)`
    QueryCount, "query_count", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::F32(d) => Value::U64(d.count as u64),
        DatasetHandle::I32(d) => Value::U64(d.count as u64),
        DatasetHandle::Ivvec32(d) => Value::U64(d.count as u64),
        DatasetHandle::Generic(d) => Value::U64(d.count as u64),
        DatasetHandle::Group(_) => panic!("query_count: expected facet handle, got Group"),
        DatasetHandle::Prebuffered { source, .. } => {
            // `query_count(prebuffered)` is a legitimate idiom in the
            // pvs_query workload (`cursor q = range(0, query_count(...))`),
            // so resolve to the query-facet handle and read its count.
            // Same OnceCache-backed open as the per-cycle accessors.
            match DatasetHandle::open(source, "query") {
                Ok(DatasetHandle::F32(d)) => Value::U64(d.count as u64),
                Ok(other) => panic!(
                    "query_count: expected F32 query facet, got {}",
                    dataset_handle_kind(&other)),
                Err(e) => panic!(
                    "query_count: failed to open 'query' from prebuffered '{source}': {e}"),
            }
        }
    }
);

handle_metadata_node!(
    /// Return the per-record neighbor count (k) for an I32
    /// neighbor-indices handle.
    ///
    /// Signature: `neighbor_count(handle) -> (u64)`
    NeighborCount, "neighbor_count", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::I32(d) => Value::U64(d.dim as u64),
        _ => Value::U64(0),
    }
);

handle_metadata_node!(
    /// Return the dataset's distance function (e.g., "COSINE",
    /// "EUCLIDEAN"). Operates on the dataset group; takes a Group
    /// handle.
    ///
    /// Signature: `dataset_distance_function(group) -> (String)`
    DatasetDistanceFunction, "dataset_distance_function", Str,
    eval = |h: &DatasetHandle| {
        let group = group_of(h);
        let raw = group
            .attribute("distance_function")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown");
        let df = match raw.to_uppercase().as_str() {
            "L2" | "EUCLIDEAN" => "EUCLIDEAN",
            "L1" | "MANHATTAN" => "MANHATTAN",
            "COSINE" => "COSINE",
            "DOT_PRODUCT" | "DOTPRODUCT" | "DOT" | "INNER_PRODUCT" | "IP" => "DOT_PRODUCT",
            _ => raw,
        };
        Value::Str(df.to_string().into())
    }
);

// =================================================================
// Metadata facet nodes
// =================================================================

/// Handle to a loaded variable-length i32 facet (metadata_results).
pub(crate) struct Ivvec32Dataset {
    reader: Arc<dyn VvecReader<i32>>,
    count: usize,
}

impl Ivvec32Dataset {
    fn load(source: &str, profile: &str) -> Result<Arc<Self>, String> {
        let key = (
            source.to_string(),
            profile.to_string(),
            "metadata_results".to_string(),
        );
        let any = FACET_CACHE.get_or_init(key, || {
            let group = load_dataset_group(source)?;
            let view = group
                .profile(profile)
                .ok_or_else(|| format!("profile '{profile}' not found in '{source}'"))?;
            crate::library::support::audit::record_opened(
                source,
                profile,
                "metadata_results",
                "ivvec32",
            );
            let reader = run_blocking_io(|| view.metadata_results())
                .map_err(|e| format!("failed to access metadata_results from '{source}': {e}"))?;
            let count = reader.count();
            let arc: Arc<Self> = Arc::new(Self { reader, count });
            Ok(arc as Arc<dyn std::any::Any + Send + Sync>)
        })?;
        any.downcast::<Self>().map_err(|_| {
            format!("facet cache type mismatch for '{source}:{profile}/metadata_results'")
        })
    }
}

handle_indexed_node!(
    /// Access metadata indices (variable-length matching base ordinals
    /// per query) at an index. Expects an Ivvec32 handle.
    ///
    /// Signature: `metadata_results_at(handle, index: u64) -> VecI32`
    MetadataResultsAt, "metadata_results_at", VecI32, facet = "metadata_results", eval = ivvec32_vec_at
);

handle_indexed_node!(
    /// Return the length of one metadata_results record without
    /// loading the data (reads only the 4-byte header). Expects an
    /// Ivvec32 handle.
    ///
    /// Signature: `metadata_results_len_at(handle, index: u64) -> (u64)`
    MetadataResultsLenAt, "metadata_results_len_at", U64, facet = "metadata_results",
    eval = |h: &DatasetHandle, idx: usize| match h {
        DatasetHandle::Ivvec32(d) if d.count > 0 => {
            let len = d.reader.dim_at(idx % d.count).unwrap_or(0);
            Value::U64(len as u64)
        }
        _ => Value::U64(0),
    }
);

handle_metadata_node!(
    /// Return the metadata indices count (number of predicate result
    /// sets). Expects an Ivvec32 handle.
    MetadataResultsCount, "metadata_results_count", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::Ivvec32(d) => Value::U64(d.count as u64),
        _ => Value::U64(0),
    }
);

handle_metadata_node!(
    /// Report which facets are available for the default profile of
    /// a dataset group as a comma-separated list. Expects a Group
    /// handle.
    ///
    /// Signature: `dataset_facets(group) -> (String)`
    DatasetFacets, "dataset_facets", Str,
    eval = |h: &DatasetHandle| {
        let group = group_of(h);
        // Use the default profile — group-level callers want a
        // dataset-wide manifest; specific profile facets come via
        // `profile_facets(group, idx)`.
        let names = group.profile_names();
        if let Some(first) = names.first()
            && let Some(view) = group.profile(first) {
                let manifest = view.facet_manifest();
                let mut names: Vec<&String> = manifest.keys().collect();
                names.sort();
                return Value::Str(
                    names.iter().map(|s| s.as_str()).collect::<Vec<_>>().join(", ").into()
                );
            }
        Value::Str(String::new().into())
    }
);

// =================================================================
// Profile enumeration — discover and iterate over dataset profiles
// =================================================================
//
// Group-level: take a Group handle, read the dataset's sorted profile
// list. Sort order is canonical (by base_count via `profile_sort_by_size`)
// and computed once per group via the underlying TestDataGroup.

handle_metadata_node!(
    /// Total number of profiles in a dataset group.
    ///
    /// Signature: `dataset_profile_count(group) -> (u64)`
    DatasetProfileCount, "dataset_profile_count", U64,
    eval = |h: &DatasetHandle| Value::U64(group_of(h).profile_names().len() as u64)
);

handle_metadata_node!(
    /// Comma-separated list of all profile names in canonical sort
    /// order (by base_count).
    ///
    /// Signature: `dataset_profile_names(group) -> (String)`
    DatasetProfileNames, "dataset_profile_names", Str,
    eval = |h: &DatasetHandle| Value::Str(group_of(h).profile_names().join(", ").into())
);

/// Return profile names matching a prefix, comma-separated.
///
/// Signature: `matching_profiles(group, prefix: str) -> (String)`
///
/// If prefix is empty, returns all profiles. Used by `for_each:`
/// phase templates to discover profiles dynamically.
///
/// `group` declares its source-string auto-resolver via the
/// `Resolved<GroupResolver, _>` marker wrapper — the macro reads
/// `<Resolved<GroupResolver, DatasetHandle> as Wire>::RESOLVER` at
/// codegen and emits the matching `FuncSig.default_resolver`
/// (`DefaultResolver::Group`). The spliced `dataset_group_open` yields
/// the canonical `Value::Handle(Arc<DatasetHandle>)`, so the wire is
/// resolved as `DatasetHandle` and the group is taken via `group_of`
/// (downcasting straight to `TestDataGroup` would fail — the handle is
/// always the unified `DatasetHandle` enum).
#[crate::polydat_node(category = RealData)]
fn matching_profiles(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    prefix: &str,
) -> String {
    let group: &TestDataGroup = group_of(&group);
    let all = group.profile_names();
    let mut matched: Vec<&str> = if prefix.is_empty() {
        all.iter().map(|s| s.as_str()).collect()
    } else {
        all.iter()
            .filter(|s| s.starts_with(prefix))
            .map(|s| s.as_str())
            .collect()
    };
    // Natural-order sort: alphabetic with numeric runs
    // compared as numbers so `label_03` sorts before
    // `label_10`. The upstream `group.profile_names()`
    // orders by `base_count` (vectordata's choice — useful
    // for index-based lookups), but `for_each` iteration
    // wants stable, human-natural order so users see
    // label_01, label_02, label_03 instead of whatever the
    // size-sort happens to produce.
    matched.sort_by(|a, b| natural_cmp(a, b));
    matched.join(",")
}

/// Natural ordering: split each string into alternating text
/// and numeric runs and compare run-by-run, comparing numeric
/// runs as integers. Beats lexicographic on `label_03` vs
/// `label_10` (lex: "10" < "3"; natural: 3 < 10).
fn natural_cmp(a: &str, b: &str) -> std::cmp::Ordering {
    let mut ai = a.chars().peekable();
    let mut bi = b.chars().peekable();
    loop {
        match (ai.peek().copied(), bi.peek().copied()) {
            (None, None) => return std::cmp::Ordering::Equal,
            (None, _) => return std::cmp::Ordering::Less,
            (_, None) => return std::cmp::Ordering::Greater,
            (Some(ac), Some(bc)) => {
                if ac.is_ascii_digit() && bc.is_ascii_digit() {
                    let mut na: u64 = 0;
                    while let Some(c) = ai.peek().copied()
                        && c.is_ascii_digit()
                    {
                        na = na
                            .saturating_mul(10)
                            .saturating_add((c as u8 - b'0') as u64);
                        ai.next();
                    }
                    let mut nb: u64 = 0;
                    while let Some(c) = bi.peek().copied()
                        && c.is_ascii_digit()
                    {
                        nb = nb
                            .saturating_mul(10)
                            .saturating_add((c as u8 - b'0') as u64);
                        bi.next();
                    }
                    match na.cmp(&nb) {
                        std::cmp::Ordering::Equal => continue,
                        non_eq => return non_eq,
                    }
                } else {
                    match ac.cmp(&bc) {
                        std::cmp::Ordering::Equal => {
                            ai.next();
                            bi.next();
                        }
                        non_eq => return non_eq,
                    }
                }
            }
        }
    }
}

/// Look up a profile name by index from the canonical sorted list.
///
/// Signature: `dataset_profile_name_at(group, index: u64) -> (String)`
///
/// Index wraps modulo the number of profiles.
#[crate::polydat_node(category = RealData)]
fn dataset_profile_name_at(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    index: u64,
) -> String {
    let group: &TestDataGroup = group_of(&group);
    let names = group.profile_names();
    if names.is_empty() {
        String::new()
    } else {
        names[(index as usize) % names.len()].clone()
    }
}

/// Return the base vector count for the profile at a given index.
///
/// Signature: `profile_base_count(group, index: u64) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn profile_base_count(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    index: u64,
) -> u64 {
    let group: &TestDataGroup = group_of(&group);
    let names = group.profile_names();
    if names.is_empty() {
        0
    } else {
        let name = &names[(index as usize) % names.len()];
        group
            .profile(name)
            .and_then(|view| view.base_count())
            .unwrap_or(0)
    }
}

/// Return the comma-separated facet list for the profile at a given index.
///
/// Signature: `profile_facets(group, index: u64) -> (String)`
#[crate::polydat_node(category = RealData)]
fn profile_facets(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    index: u64,
) -> String {
    let group: &TestDataGroup = group_of(&group);
    let names = group.profile_names();
    if names.is_empty() {
        String::new()
    } else {
        let name = &names[(index as usize) % names.len()];
        match group.profile(name) {
            Some(view) => {
                let manifest = view.facet_manifest();
                let mut fnames: Vec<&String> = manifest.keys().collect();
                fnames.sort();
                fnames
                    .iter()
                    .map(|s| s.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            }
            None => String::new(),
        }
    }
}

/// Partition a dataset's vector space by its **profiles matching a
/// pattern**, treated as cumulative size tiers (an SRD-71 partition
/// source). One partition per masked profile, in canonical
/// (base-count-ascending) order: partition `k` spans
/// `[prev_masked_base_count, this_masked_base_count)` — exactly the
/// vectors added at that tier — so a sweep's "load only the increment
/// since the previously loaded set" is just the partition's
/// `[start_of(p), end_of(p))`, and a partition inherently knows its
/// start (no cross-iteration carry needed). `idx_of(p)` is the 0-based
/// masked position (pairs with `matching_profile_name_at` to address
/// the tier's own ground-truth facets); `count_of(p)` is the number of
/// masked tiers; `base_extent` is the largest masked tier's size.
///
/// `pattern` follows the literal / glob / regex promotion of
/// [`crate::library::support::pattern::compile_pattern`]; `*` (or any pattern that
/// matches all names) selects every profile.
///
/// Signature: `profile_partitions(group, pattern: str) -> (PartitionList)`
#[crate::polydat_node(category = RealData)]
fn profile_partitions(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    pattern: &str,
) -> crate::derive_support::Ext<crate::iteration::cursor_partition::PartitionList> {
    let group: &TestDataGroup = group_of(&group);
    let parts = build_profile_partitions(group, pattern);
    crate::derive_support::Ext(crate::iteration::cursor_partition::PartitionList::new(
        parts,
    ))
}

/// Build the cumulative size-tier partitions for the profiles of `group`
/// matching `pattern` (literal/glob/regex promotion). Shared by the
/// [`profile_partitions`] node and the comprehension-source desugaring
/// in `iteration::comprehension::eval` so a `for: "p in
/// profile_partitions(...)"` sweep produces identical partitions either
/// way. Partition `k` spans `[prev masked base_count, this base_count)`.
/// The masked profile size-tiers for `group` matching `pattern`, in
/// canonical (base-count-ascending) order: `(name, cumulative_base_count)`
/// for each matching profile WITH a non-zero base count. Profiles with no
/// base facet, or a zero count, are skipped — they would otherwise emit a
/// degenerate empty `[prev, prev)` partition (and a query against an empty
/// tier). Shared by [`build_profile_partitions`] and
/// [`matching_profile_name_at`] so a partition's `idx_of(p)` and the tier
/// name resolve against the SAME masked sequence.
fn masked_profile_tiers(group: &TestDataGroup, pattern: &str) -> Vec<(String, u64)> {
    let (re, _) = crate::library::support::pattern::compile_pattern(pattern)
        .unwrap_or_else(|e| panic!("profile pattern: {e}"));
    group
        .profile_names()
        .iter()
        .filter(|n| re.is_match(n))
        .filter_map(|n| {
            group
                .profile(n)
                .and_then(|v| v.base_count())
                .filter(|&c| c > 0)
                .map(|c| (n.clone(), c))
        })
        .collect()
}

pub(crate) fn build_profile_partitions(
    group: &TestDataGroup,
    pattern: &str,
) -> Vec<crate::iteration::cursor_partition::Partition> {
    // Masked size-tiers (matching profiles with a non-zero base count),
    // canonical (ascending) order — skipping zero-count profiles avoids a
    // degenerate empty first partition.
    let masked: Vec<u64> = masked_profile_tiers(group, pattern)
        .into_iter()
        .map(|(_, c)| c)
        .collect();
    let base_extent = masked.last().copied().unwrap_or(0);
    let count = masked.len() as u64;
    let pct = |o: u64| {
        if base_extent == 0 {
            0.0
        } else {
            (o as f64 / base_extent as f64) * 100.0
        }
    };
    let mut parts: Vec<crate::iteration::cursor_partition::Partition> =
        Vec::with_capacity(masked.len());
    let mut prev: u64 = 0;
    for (k, &this) in masked.iter().enumerate() {
        parts.push(crate::iteration::cursor_partition::Partition {
            idx: k as u64,
            count,
            start_ord: prev,
            end_ord: this,
            start_pct: pct(prev),
            end_pct: pct(this),
            base_extent,
        });
        prev = this;
    }
    parts
}

/// Name of the `index`-th profile **matching `pattern`**, in canonical
/// (base-count-ascending) order. Pairs with `profile_partitions`'s
/// masked-position `idx_of` so a sweep can prebuffer the active tier's
/// own ground-truth facets (`dataset_prebuffer(str_concat("ds:", name))`).
/// `pattern` follows the literal / glob / regex promotion; the index
/// wraps modulo the number of matching profiles; empty string if none
/// match.
///
/// Signature: `matching_profile_name_at(group, pattern: str, index: u64) -> (String)`
#[crate::polydat_node(category = RealData)]
fn matching_profile_name_at(
    group: crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>,
    pattern: &str,
    index: u64,
) -> String {
    let group: &TestDataGroup = group_of(&group);
    // Same masked sequence `profile_partitions` uses (matching profiles
    // with a non-zero base count, canonical order), so `idx_of(p)` from a
    // partition resolves to the right tier name.
    let tiers = masked_profile_tiers(group, pattern);
    if tiers.is_empty() {
        String::new()
    } else {
        tiers[(index as usize) % tiers.len()].0.clone()
    }
}

// =================================================================
// Prebuffering — eagerly download dataset facets before workload run
// =================================================================

source_only_node!(
    /// Eagerly download all facets for a dataset profile into the
    /// local cache, returning a `DatasetHandle::Group` handle
    /// that downstream facet accessors take as their first
    /// argument. After this returns, every subsequent facet read
    /// served by [`vectordata::TestDataView`] hits the merkle-
    /// verified mmap fast path with no further network traffic.
    ///
    /// Signature: `dataset_prebuffer(source) -> Handle`
    ///
    /// **Why a handle, not a count.** Returning a value the
    /// downstream accessors *consume* makes prebuffer part of
    /// the dataflow graph: DCE keeps the chain alive because
    /// `vector_at(prebuffered, q)` needs `prebuffered` to be
    /// resolvable, which forces evaluation. Bindings such as
    /// `init prebuffered = dataset_prebuffer(...)` whose result
    /// nothing reads would still be pruned (TODO: rationalise
    /// dangling-init dataflow as a separate followup; the
    /// established pattern is to thread the handle through).
    ///
    /// `source` is the canonical `dataset:profile` string. The
    /// returned handle is a `Group` handle — accessors that take
    /// it route through the same `DatasetHandle::open(...)`
    /// resolver path used by other group-aware nodes
    /// (`dataset_facets`, `dataset_distance_function`, ...).
    ///
    /// Errors during prebuffer surface via stderr; the node
    /// still returns the group handle so downstream binds don't
    /// fail with a "missing value" cascade — the operator's
    /// intent is "best effort warm-up", and a workload that
    /// wants strict guarantees can wrap this in a `required(...)`
    /// predicate or check facet readiness explicitly.
    DatasetPrebuffer, "dataset_prebuffer",
    out_port = Handle,
    eval = |source| do_dataset_prebuffer(source)
);

/// Implementation behind the `dataset_prebuffer` node — kept
/// out of the macro body so we can use `?`/early return
/// without fighting the closure-vs-eval-fn return type
/// mismatch the `source_only_node!` macro embeds.
fn do_dataset_prebuffer(source: &str) -> Value {
    // The init-binding contract (SRD 11) means this function is
    // expected to fire exactly once per scope activation: Plan B
    // pulls the binding on the activation kernel; OpBuilder's
    // init_overrides propagate the result to every fiber. The
    // PREBUFFER_CACHE below is belt-and-suspenders: it serializes
    // anyone who ends up calling here through a non-init path,
    // and makes per-source thundering-herd structurally
    // impossible regardless of upstream caller behavior.
    match PREBUFFER_CACHE.get_or_init(source.to_string(), || {
        // Only the first concurrent caller for this source runs
        // the inner body — the audit "entered" event reflects
        // that single download, not per-caller noise.
        crate::library::support::audit::record_prebuffer_entered(source);
        do_dataset_prebuffer_inner(source)
    }) {
        Ok(handle) => Value::handle(handle),
        Err(_) => Value::None,
    }
}

/// Inner body of [`do_dataset_prebuffer`]. Runs **at most once
/// per (source, process)** under the [`PREBUFFER_CACHE`] OnceLock.
fn do_dataset_prebuffer_inner(source: &str) -> Result<Arc<DatasetHandle>, String> {
    // Return a Group handle in every exit (success or error) so
    // the downstream `*_at(prebuffered, q)` accessors can resolve.
    // Failed prebuffer still hands back the group handle — the
    // accessors will then HTTP-fall-through, and the operator
    // sees the prebuffer error in the audit log.
    let group = match load_dataset_group(source) {
        Ok(g) => g,
        Err(e) => {
            let msg = format!("dataset_prebuffer: cannot resolve '{source}': {e}");
            crate::library::support::audit::error(&msg);
            // No group → no handle to hand back. Sticky error in
            // the cache; downstream accessors will produce a
            // diagnostic when they fail to downcast.
            return Err(msg);
        }
    };
    let group_for_handle = group.clone();
    let (_, profile) = parse_source_specifier(source);
    let view = match group.profile(profile) {
        Some(v) => v,
        None => {
            crate::library::support::audit::error(&format!(
                "dataset_prebuffer: profile '{profile}' not found in '{source}'"
            ));
            return Ok(Arc::new(DatasetHandle::Prebuffered {
                _group: group_for_handle,
                source: source.to_string(),
            }));
        }
    };
    // vectordata's default `prebuffer_all_with_progress`
    // walks the manifest and calls `FacetStorage::prebuffer`
    // per facet — works for both record-shaped (xvec) and
    // scalar (typed) facets after the storage-transport
    // refactor in vectordata 1.0.0.
    //
    // Audit instrumentation: log every facet the prebuffer
    // *covers* with its key (`source:profile/facet`). Compared
    // against the `vectordata: opened …` lines emitted by
    // [`load_uniform_facet`] / [`GenericFacetDataset::load`],
    // this surfaces any facet the workload reads at cycle time
    // that prebuffer did NOT pull — the typical cause of a
    // "still hitting HTTP after prebuffer" symptom.
    // Manual facet walk so we can hook the per-chunk
    // `DownloadProgress` callback (vectordata's
    // `view.prebuffer_all_with_progress` only fires its outer cb
    // once per facet *completion*; the per-chunk cb is exposed
    // via `FacetStorage::prebuffer_with_progress`). Every facet's
    // chunk-level progress is throttled per facet: ~1 Hz for the first
    // 10s of its download (fine detail while it ramps), then one line
    // per 10s for the long tail so a multi-gigabyte facet doesn't spew
    // thousands of lines into session.log; a final per-facet `covered`
    // line lands when the download finishes.
    let mut facet_count: u64 = 0;
    for (name, _descriptor) in view.facet_manifest() {
        // Skip facets with unrecognised element types (vectordata's
        // own default impl skips these — they're not data facets the
        // typed reader would touch).
        if view.facet_element_type(&name).is_err() {
            continue;
        }

        let storage = match run_blocking_io(|| view.open_facet_storage(&name)) {
            Ok(s) => s,
            Err(e) => {
                crate::library::support::audit::warn(&format!(
                    "dataset_prebuffer: open '{name}' for prebuffer failed: {e}"
                ));
                continue;
            }
        };

        // Inline the closure at the call site so Rust's type
        // inference can read `&DownloadProgress` straight from
        // the trait bound on `prebuffer_with_progress`.
        // The `DownloadProgress` type is `pub(crate)` upstream
        // (vectordata 1.0.2) so we can't name it ourselves.
        let prebuf_source = source.to_string();
        let prebuf_profile = profile.to_string();
        let prebuf_facet = name.clone();
        let mut last_done: u64 = 0;
        let facet_start = std::time::Instant::now();
        let mut last_log_at = facet_start;
        // `prebuffer_with_progress` drives `reqwest::blocking`,
        // which spins up a private tokio runtime per request.
        // Without parking the outer worker via `block_in_place`,
        // dropping that inner runtime inside an async context
        // panics with "Cannot drop a runtime in a context where
        // blocking is not allowed". Same pattern as
        // `load_dataset_group` and `load_uniform_facet`.
        let prebuf_result = run_blocking_io(|| {
            storage.prebuffer_with_progress(|p| {
                // Per-facet throttle: ~1 Hz for the first 10s of this
                // facet's download, then once per 10s, so a gigabyte-sized
                // facet doesn't spew thousands of progress lines.
                let now = std::time::Instant::now();
                let interval =
                    if now.duration_since(facet_start) < std::time::Duration::from_secs(10) {
                        std::time::Duration::from_secs(1)
                    } else {
                        std::time::Duration::from_secs(10)
                    };
                let since_last = now.duration_since(last_log_at);
                if since_last < interval {
                    return;
                }
                last_log_at = now;
                let total_b = p.total_bytes();
                let done_b = p.downloaded_bytes();
                let total_c = p.total_chunks();
                let done_c = p.completed_chunks();
                let pct = p.fraction() * 100.0;
                // Throughput over the actual gap between log lines, so the
                // rate reads correctly whichever interval is in force.
                let delta_mb = (done_b.saturating_sub(last_done)) as f64 / (1024.0 * 1024.0);
                let rate_mb_s = delta_mb / since_last.as_secs_f64().max(0.001);
                last_done = done_b;
                crate::library::support::audit::info(&format!(
                    "prebuffer: progress {prebuf_source}:{prebuf_profile}/{prebuf_facet} \
                 {pct:5.1}% ({done_c}/{total_c} chunks, \
                 {done_mb:.1}/{total_mb:.1} MB, {rate_mb_s:.1} MB/s)",
                    done_mb = done_b as f64 / (1024.0 * 1024.0),
                    total_mb = total_b as f64 / (1024.0 * 1024.0),
                ));
            })
        });
        if let Err(e) = prebuf_result {
            crate::library::support::audit::warn(&format!(
                "dataset_prebuffer: download error for '{source}' facet '{name}': {e}"
            ));
            continue;
        }
        facet_count = facet_count.saturating_add(1);
        crate::library::support::audit::record_prebuffered(source, profile, &name);
    }
    crate::library::support::audit::log_prebuffer_summary(source, profile, facet_count);
    Ok(Arc::new(DatasetHandle::Prebuffered {
        _group: group_for_handle,
        source: source.to_string(),
    }))
}

// =================================================================
// Generic facet access — type-aware scalar/vector readers
// =================================================================

/// A type-erased facet reader that stores values as i64.
/// Uses `generic_view().open_facet_typed::<i64>()` which handles all
/// element types (u8, i32, etc.), caching, and local/remote access.
pub(crate) struct GenericFacetDataset {
    reader: vectordata::typed_access::TypedReader<i64>,
    count: usize,
}

impl GenericFacetDataset {
    fn load(source: &str, profile: &str, facet: &str) -> Result<Arc<Self>, String> {
        let key = (source.to_string(), profile.to_string(), facet.to_string());
        let any = FACET_CACHE.get_or_init(key, || {
            let group = load_dataset_group(source)?;
            let gv = group
                .generic_view(profile)
                .ok_or_else(|| format!("profile '{profile}' not found in '{source}'"))?;
            crate::library::support::audit::record_opened(source, profile, facet, "generic-typed");
            let reader = run_blocking_io(|| gv.open_facet_typed::<i64>(facet))
                .map_err(|e| format!("failed to open {facet} from '{source}:{profile}': {e}"))?;
            let count = reader.count();
            let arc: Arc<Self> = Arc::new(Self { reader, count });
            Ok(arc as Arc<dyn std::any::Any + Send + Sync>)
        })?;
        any.downcast::<Self>()
            .map_err(|_| format!("facet cache type mismatch for '{source}:{profile}/{facet}'"))
    }

    fn get_scalar(&self, index: usize) -> i64 {
        if self.count == 0 {
            return 0;
        }
        self.reader.get_value(index % self.count).unwrap_or(0)
    }

    fn format_scalar(&self, index: usize) -> String {
        self.get_scalar(index).to_string()
    }
}

// Generic-facet readers (typed scalar). Take a Generic handle and
// read the i64-cast value at the requested index.

fn generic_str_at(h: &DatasetHandle, idx: usize) -> Value {
    match h {
        DatasetHandle::Generic(d) => Value::Str(d.format_scalar(idx).into()),
        _ => Value::Str(String::new().into()),
    }
}

handle_indexed_node!(
    /// Access a metadata value per base ordinal. Expects a Generic
    /// handle opened against the `metadata_content` facet.
    ///
    /// Signature: `metadata_value_at(handle, index: u64) -> (String)`
    MetadataValueAt, "metadata_value_at", Str, facet = "metadata_content", eval = generic_str_at
);

handle_indexed_node!(
    /// Access a predicate value per query ordinal. Expects a Generic
    /// handle opened against the `metadata_predicates` facet.
    ///
    /// Signature: `predicate_value_at(handle, index: u64) -> (String)`
    PredicateValueAt, "predicate_value_at", Str, facet = "metadata_predicates", eval = generic_str_at
);

handle_metadata_node!(
    /// Count of metadata content records. Expects a Generic handle.
    ///
    /// Signature: `metadata_content_count(handle) -> (u64)`
    MetadataContentCount, "metadata_content_count", U64,
    eval = |h: &DatasetHandle| match h {
        DatasetHandle::Generic(d) => Value::U64(d.count as u64),
        _ => Value::U64(0),
    }
);

// ---------------------------------------------------------------------------
// Signature declarations for the DSL registry
// ---------------------------------------------------------------------------

use crate::ast::SlotType;
use crate::dsl::registry::{Arity, DefaultResolver, FuncCategory, FuncSig, ParamSpec};

// Macros to keep the bulk of the registry compact and consistent.
// Per SRD 53 §"Source-string call-site sugar": each handle-taking
// accessor declares a `default_resolver` so the binding compiler
// can promote a string source into the right resolver call.

macro_rules! sig_handle_indexed {
    ($name:literal, $resolver:expr, $desc:literal, $help:literal) => {
        FuncSig {
            name: $name,
            category: FuncCategory::RealData,
            outputs: 1,
            description: $desc,
            help: $help,
            identity: None,
            variadic_ctor: None,
            params: &[
                ParamSpec {
                    name: "handle",
                    slot_type: SlotType::Wire,
                    required: true,
                    example: "base",
                    constraint: None,
                },
                ParamSpec {
                    name: "index",
                    slot_type: SlotType::Wire,
                    required: true,
                    example: "cycle",
                    constraint: None,
                },
            ],
            arity: Arity::Fixed,
            commutativity: crate::ast::Commutativity::Positional,
            default_resolver: Some($resolver),
            output_type: crate::dsl::registry::OutputType::Fixed,
            // Hand registration: no static return-port declaration;
            // type inference falls back to the name heuristic.
            output_port: None,
        }
    };
}

macro_rules! sig_handle_metadata {
    ($name:literal, $resolver:expr, $desc:literal, $help:literal) => {
        FuncSig {
            name: $name,
            category: FuncCategory::RealData,
            outputs: 1,
            description: $desc,
            help: $help,
            identity: None,
            variadic_ctor: None,
            params: &[ParamSpec {
                name: "handle",
                slot_type: SlotType::Wire,
                required: true,
                example: "base",
                constraint: None,
            }],
            arity: Arity::Fixed,
            commutativity: crate::ast::Commutativity::Positional,
            default_resolver: Some($resolver),
            output_type: crate::dsl::registry::OutputType::Fixed,
            // Hand registration: no static return-port declaration;
            // type inference falls back to the name heuristic.
            output_port: None,
        }
    };
}

/// Signatures for vector dataset access nodes (feature-gated).
///
/// `dataset_open` and `dataset_group_open` register themselves via the
/// `#[polydat_node]` macro's own `NodeRegistration` channel.
pub fn signatures() -> &'static [FuncSig] {
    use FuncCategory as C;
    &[
        // ===== Per-cycle facet accessors (typed-vector outputs) =====
        sig_handle_indexed!(
            "vector_at",
            DefaultResolver::Facet("base"),
            "access f32 vector by index",
            "Read an f32 vector from a facet handle as a typed VecF32.\nAuto-promotes a string source via dataset_open(_,\"base\").\nExample: vector_at(base, cycle)"
        ),
        sig_handle_indexed!(
            "query_vector_at",
            DefaultResolver::Facet("query"),
            "access query vector by index",
            "Alias for vector_at over a query-facet handle.\nAuto-promotes a string source via dataset_open(_,\"query\")."
        ),
        sig_handle_indexed!(
            "neighbor_indices_at",
            DefaultResolver::Facet("neighbor_indices"),
            "ground-truth neighbor indices for a query",
            "Read ground-truth k-nearest neighbor indices for a query as a\ntyped VecI32. Auto-promotes a string source via\ndataset_open(_,\"neighbor_indices\")."
        ),
        sig_handle_indexed!(
            "neighbor_distances_at",
            DefaultResolver::Facet("neighbor_distances"),
            "ground-truth neighbor distances for a query",
            "Read ground-truth distances for a query's k-nearest neighbors\nas a typed VecF32."
        ),
        sig_handle_indexed!(
            "filtered_neighbor_indices_at",
            DefaultResolver::Facet("filtered_neighbor_indices"),
            "filtered ground-truth neighbor indices",
            "Read filtered ground-truth indices for a query as a typed VecI32.\nUsed for filtered-ANN recall verification."
        ),
        sig_handle_indexed!(
            "filtered_neighbor_distances_at",
            DefaultResolver::Facet("filtered_neighbor_distances"),
            "filtered ground-truth neighbor distances",
            "Read filtered ground-truth distances for a query as a typed VecF32."
        ),
        sig_handle_indexed!(
            "metadata_results_len_at",
            DefaultResolver::Facet("metadata_results"),
            "length of metadata indices for a query",
            "Return the per-record matching-base count for a query without\nloading the full index list (reads only the 4-byte header)."
        ),
        sig_handle_indexed!(
            "metadata_results_at",
            DefaultResolver::Facet("metadata_results"),
            "matching base ordinals for a query predicate",
            "Variable-length list of base vector ordinals matching a query's\npredicate."
        ),
        sig_handle_indexed!(
            "metadata_value_at",
            DefaultResolver::Facet("metadata_content"),
            "scalar metadata value per base vector",
            "Read a metadata value for a base vector by ordinal.\nReads from the metadata_content facet."
        ),
        sig_handle_indexed!(
            "predicate_value_at",
            DefaultResolver::Facet("metadata_predicates"),
            "scalar predicate value per query",
            "Read a predicate value for a query by ordinal.\nReads from the metadata_predicates facet."
        ),
        // ===== Per-handle metadata =====
        sig_handle_metadata!(
            "vector_dim",
            DefaultResolver::Facet("base"),
            "vector dimensionality of a facet handle",
            "Return the per-record element count (dimension) of a vector facet."
        ),
        sig_handle_metadata!(
            "vector_count",
            DefaultResolver::Facet("base"),
            "record count of a facet handle",
            "Return the number of records in a facet handle (base vectors,\nquery vectors, ...). Auto-promotes a string source via\ndataset_open(_,\"base\")."
        ),
        sig_handle_metadata!(
            "query_count",
            DefaultResolver::Facet("query"),
            "record count of a query-facet handle",
            "Same as vector_count but defaults to dataset_open(_,\"query\")\nfor string sources."
        ),
        sig_handle_metadata!(
            "neighbor_count",
            DefaultResolver::Facet("neighbor_indices"),
            "ground-truth neighbors per query (maxk)",
            "Return the per-record neighbor count (k) of a neighbor-indices handle."
        ),
        sig_handle_metadata!(
            "metadata_results_count",
            DefaultResolver::Facet("metadata_results"),
            "number of predicate result sets",
            "Return the record count of a metadata-indices handle."
        ),
        sig_handle_metadata!(
            "metadata_content_count",
            DefaultResolver::Facet("metadata_content"),
            "number of metadata content records",
            "Return the record count of a metadata-content handle."
        ),
        // ===== Group-level (Group handle) =====
        sig_handle_metadata!(
            "dataset_distance_function",
            DefaultResolver::Group,
            "dataset distance/similarity function name",
            "Return the distance function declared in the dataset metadata\n('COSINE','EUCLIDEAN','DOT_PRODUCT','MANHATTAN'). Group-level."
        ),
        sig_handle_metadata!(
            "dataset_facets",
            DefaultResolver::Group,
            "list available facets in default profile",
            "Comma-separated facet names available in the group's default profile."
        ),
        sig_handle_metadata!(
            "dataset_profile_count",
            DefaultResolver::Group,
            "total number of profiles in a dataset",
            "Number of profiles defined in the dataset group."
        ),
        sig_handle_metadata!(
            "dataset_profile_names",
            DefaultResolver::Group,
            "comma-separated list of profile names",
            "All profile names in canonical sort order (by base_count)."
        ),
        // `matching_profiles`, `dataset_profile_name_at`,
        // `profile_base_count`, and `profile_facets` are now
        // `#[polydat_node]`-emitted; they register their FuncSig
        // via the macro's NodeRegistration entry. Their
        // `Resolved<GroupResolver, TestDataGroup>` arg supplies
        // the `default_resolver: Some(DefaultResolver::Group)`
        // value to the emitted FuncSig automatically (Wire-trait
        // RESOLVER projection — SRD-80b).

        // ===== Side-effect resolver =====
        FuncSig {
            name: "dataset_prebuffer",
            category: C::RealData,
            outputs: 1,
            description: "eagerly download dataset facets to local cache",
            help: "Downloads all facets for a dataset to the local cache. Returns 0\n(side-effect resolver). Subsequent loads use fast local mmap access.\nKept on a string source — typically called once per workload at\ninit time.\nExample: const _pb := dataset_prebuffer(\"example\")",
            identity: None,
            variadic_ctor: None,
            params: &[ParamSpec {
                name: "source",
                slot_type: SlotType::Wire,
                required: true,
                example: "\"test\"",
                constraint: None,
            }],
            arity: Arity::Fixed,
            commutativity: crate::ast::Commutativity::Positional,
            default_resolver: None,
            output_type: crate::dsl::registry::OutputType::Fixed,
            // Hand registration: no static return-port declaration;
            // type inference falls back to the name heuristic.
            output_port: None,
        },
    ]
}

/// Try to build a vector dataset node from a function name and const args.
///
/// Returns `None` if the name is not handled by this module.
/// All functions in this module are feature-gated on `vectordata`.
#[cfg(feature = "vectordata")]
pub(crate) fn build_node(
    name: &str,
    _wires: &[crate::compile::assembly::WireRef],
    _wire_types: &[crate::ast::PortType],
    _consts: &[crate::dsl::factory::ConstArg],
) -> Option<Result<Box<dyn crate::ast::PolydatNode>, String>> {
    // Every dataset function in this module now takes its
    // `source` (and any other previously-const string params)
    // as a Wire input, so `consts` is unused — the spec arrives
    // through `inputs[i]` at eval time. Literal-string args
    // are auto-lifted to anonymous `ConstStr` wire nodes by the
    // binding compiler; non-literal args (e.g. `printf` from a
    // string-interpolated source spec) wire directly.
    match name {
        // `dataset_open` and `dataset_group_open` register through
        // their `#[polydat_node]`-emitted NodeRegistration entries.
        "vector_at" => Some(Ok(
            Box::new(VectorAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "query_vector_at" => Some(Ok(
            Box::new(QueryVectorAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "neighbor_indices_at" => Some(Ok(
            Box::new(NeighborIndicesAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "neighbor_distances_at" => Some(Ok(
            Box::new(NeighborDistancesAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "filtered_neighbor_indices_at" => Some(Ok(
            Box::new(FilteredNeighborIndicesAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "filtered_neighbor_distances_at" => Some(Ok(
            Box::new(FilteredNeighborDistancesAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "dataset_distance_function" => Some(Ok(
            Box::new(DatasetDistanceFunction::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "vector_dim" => Some(Ok(
            Box::new(VectorDim::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "vector_count" => Some(Ok(
            Box::new(VectorCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "query_count" => Some(Ok(
            Box::new(QueryCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "neighbor_count" => Some(Ok(
            Box::new(NeighborCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "metadata_results_len_at" => Some(Ok(
            Box::new(MetadataResultsLenAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "metadata_results_at" => Some(Ok(
            Box::new(MetadataResultsAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "metadata_results_count" => Some(Ok(
            Box::new(MetadataResultsCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "dataset_facets" => Some(Ok(
            Box::new(DatasetFacets::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "dataset_profile_count" => Some(Ok(
            Box::new(DatasetProfileCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "dataset_profile_names" => Some(Ok(
            Box::new(DatasetProfileNames::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        // `matching_profiles`, `dataset_profile_name_at`,
        // `profile_base_count`, `profile_facets` register
        // through the `#[polydat_node]`-emitted NodeRegistration.
        "dataset_prebuffer" => Some(Ok(
            Box::new(DatasetPrebuffer::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "metadata_value_at" => Some(Ok(
            Box::new(MetadataValueAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "predicate_value_at" => Some(Ok(
            Box::new(PredicateValueAt::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        "metadata_content_count" => Some(Ok(
            Box::new(MetadataContentCount::new()) as Box<dyn crate::ast::PolydatNode>
        )),
        _ => None,
    }
}

#[cfg(feature = "vectordata")]
crate::register_nodes!(signatures, build_node);

// =========================================================================
// Cursor-sugar handlers (SRD 18 §"Source-driven workloads")
// =========================================================================
//
// Three sugar forms recognized by this module — none of them
// known to the core compiler, which dispatches generically
// through `dsl::cursor_sugar`. Each desugars to a synthetic
// `range(0, vector_count|query_count(...))` constructor plus
// auxiliary bindings:
//
//   - `__<cursor>_prebuffer := dataset_prebuffer("ds:profile")`
//     loaded once at init time so cycle-time accessor calls hit
//     prefetched memory.
//   - `<cursor>__vector := vector_at("ds:profile", <cursor>__ordinal)`
//     (or `query_vector_at` for the query
//     facet) — published as the cursor's `vector` projection so
//     workloads can reference `<cursor>.vector`.
//
// Forms:
//   - `vectordata_source(dataset, profile, facet)` — explicit facet
//   - `vectordata_base(dataset, profile)`           — facet = "base"
//   - `vectordata_query(dataset, profile)`          — facet = "query"
//
// Facet-specific projections like `metadata` / `ground_truth` /
// `predicate` stay explicit (the user writes
// `meta := metadata_value_at(<cursor>.ordinal, "ds:profile")` by
// hand) because their existence is dataset-conditional — not
// every dataset declares a metadata column or a predicate facet.

#[cfg(feature = "vectordata")]
fn vectordata_sugar(
    source_name: &str,
    constructor: &crate::dsl::ast::Expr,
) -> Result<Option<crate::dsl::cursor_sugar::CursorSugar>, String> {
    use crate::dsl::ast::{Arg, CallExpr, Expr};
    use crate::dsl::compile::positional_str_lit;
    use crate::dsl::cursor_sugar::{AuxBinding, CursorSugar};

    let Expr::Call(call) = constructor else {
        return Ok(None);
    };

    let (dataset, profile, facet) = match call.func.as_str() {
        "vectordata_source" => {
            let d = positional_str_lit(call.args.first()).ok_or_else(|| format!(
                "cursor '{source_name}': vectordata_source(dataset, profile, facet) — first arg must be a string literal"
            ))?;
            let p = positional_str_lit(call.args.get(1)).ok_or_else(|| format!(
                "cursor '{source_name}': vectordata_source(dataset, profile, facet) — second arg must be a string literal"
            ))?;
            let f = positional_str_lit(call.args.get(2)).ok_or_else(|| format!(
                "cursor '{source_name}': vectordata_source(dataset, profile, facet) — third arg must be a string literal (\"base\" or \"query\")"
            ))?;
            (d, p, f)
        }
        "vectordata_base" | "vectordata_query" => {
            let f = call.func.strip_prefix("vectordata_").unwrap().to_string();
            let d = positional_str_lit(call.args.first()).ok_or_else(|| format!(
                "cursor '{source_name}': {}(dataset, profile) — first arg must be a string literal",
                call.func,
            ))?;
            let p = positional_str_lit(call.args.get(1)).ok_or_else(|| format!(
                "cursor '{source_name}': {}(dataset, profile) — second arg must be a string literal",
                call.func,
            ))?;
            (d, p, f)
        }
        _ => return Ok(None),
    };

    if facet != "base" && facet != "query" {
        return Err(format!(
            "cursor '{source_name}': vectordata facet must be \"base\" or \"query\", got \"{facet}\""
        ));
    }
    let (count_func, vector_func) = match facet.as_str() {
        "base" => ("vector_count", "vector_at"),
        "query" => ("query_count", "query_vector_at"),
        _ => unreachable!(),
    };

    let combined = format!("{dataset}:{profile}");
    let span = call.span;
    let lit = |s: String| Expr::StringLit(s, span);
    let positional = |e: Expr| Arg::Positional(e);

    let effective_constructor = Expr::Call(CallExpr {
        func: "range".into(),
        args: vec![
            positional(Expr::IntLit(0, span)),
            positional(Expr::Call(CallExpr {
                func: count_func.into(),
                args: vec![positional(lit(combined.clone()))],
                span,
            })),
        ],
        span,
    });

    let prebuffer_binding = AuxBinding {
        name: format!("__{source_name}_prebuffer"),
        value: Expr::Call(CallExpr {
            func: "dataset_prebuffer".into(),
            args: vec![positional(lit(combined.clone()))],
            span,
        }),
        projection: None,
    };

    // Cursor sugar emits the call with the new (handle, index) order.
    // The combined source string auto-promotes to the right facet
    // handle via the binding compiler's call-site sugar (SRD 53).
    let vector_binding = AuxBinding {
        name: format!("{source_name}__vector"),
        value: Expr::Call(CallExpr {
            func: vector_func.into(),
            args: vec![
                positional(lit(combined)),
                positional(Expr::Ident(format!("{source_name}__ordinal"), span)),
            ],
            span,
        }),
        projection: Some(("vector".into(), crate::ast::PortType::VecF32)),
    };

    Ok(Some(CursorSugar {
        effective_constructor,
        aux_bindings: vec![prebuffer_binding, vector_binding],
    }))
}

#[cfg(feature = "vectordata")]
inventory::submit! {
    crate::dsl::cursor_sugar::CursorSugarRegistration {
        handler: vectordata_sugar,
        name: "vectordata",
    }
}