polydat-core 0.6.1

// 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 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 type_system.md §1.8: `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: a facet accessor opens its facet from
    /// `<source>:<profile>` (`query_vector_at(prebuffered, q)`
    /// reads the `query` facet), and a group accessor reads the
    /// group. [`DatasetHandle::resolve_facet`] and
    /// [`DatasetHandle::resolve_group`] are those two readings.
    ///
    /// The `group` field also 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.
    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)
    }
}

/// `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.
///
/// A resolve failure panics with the underlying error rather than
/// yielding `Value::None`, so a catalog miss cannot degrade into a
/// garbage value downstream: 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
/// 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.
///
/// A resolve failure panics, as in `dataset_open` and for the
/// same reason.
#[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: a handle wire
// and an `index` wire (u64). A source string in the handle
// position is promoted to `dataset_open(source, "<facet>")` for
// the facet the accessor names, and the facet reader is opened
// once per (source, profile, facet) through `FACET_CACHE`, so
// the per-cycle work is the facet read alone.

// =================================================================
// Per-cycle indexed accessors
// =================================================================

/// 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),
        }
    }

    /// The dataset group a group accessor reads: the group of a
    /// `dataset_group_open` handle, or the group a prebuffered handle
    /// carries. A facet handle has no group, and asking one for it
    /// panics with the handle's kind.
    fn resolve_group(&self) -> &TestDataGroup {
        match self {
            DatasetHandle::Group(g) | DatasetHandle::Prebuffered { group: g, .. } => g.as_ref(),
            other => panic!(
                "expected a group or prebuffered handle, got {}",
                dataset_handle_kind(other)
            ),
        }
    }
}

/// The facet each accessor opens, as a type.
///
/// A node states its facet twice — in the auto-resolver that splices
/// `dataset_open(_, "<facet>")` when the source is a string, and in
/// the `resolve_facet` call that follows a prebuffered handle — and
/// both readings come from here, so the two cannot drift apart.
macro_rules! facet_kind {
    ($name:ident, $facet:literal) => {
        /// The `
        #[doc = $facet]
        /// ` facet of a dataset.
        pub struct $name;
        impl $name {
            /// The facet's name, as `dataset_open` takes it.
            pub const FACET: &'static str = $facet;
        }
        impl crate::derive_support::ResolverKind for $name {
            const RESOLVER: crate::dsl::registry::DefaultResolver =
                crate::dsl::registry::DefaultResolver::Facet($facet);
        }
    };
}

facet_kind!(BaseFacet, "base");
facet_kind!(QueryFacet, "query");
facet_kind!(NeighborIndicesFacet, "neighbor_indices");
facet_kind!(NeighborDistancesFacet, "neighbor_distances");
facet_kind!(FilteredNeighborIndicesFacet, "filtered_neighbor_indices");
facet_kind!(
    FilteredNeighborDistancesFacet,
    "filtered_neighbor_distances"
);
facet_kind!(MetadataResultsFacet, "metadata_results");
facet_kind!(MetadataContentFacet, "metadata_content");
facet_kind!(MetadataPredicatesFacet, "metadata_predicates");

/// A handle wire that carries its facet in its type.
type Facet<R> = crate::derive_support::Resolved<R, DatasetHandle>;

/// A handle wire on the dataset group rather than one of its facets.
type Group = crate::derive_support::Resolved<crate::derive_support::GroupResolver, DatasetHandle>;

/// The record count of a facet handle, opening `facet` first when the
/// handle is prebuffered. `who` names the caller in the diagnostics,
/// which is what the reader needs to see when a workload asks a
/// facet for a count it cannot give.
fn facet_record_count(h: &DatasetHandle, who: &str, facet: &str) -> u64 {
    match h.resolve_facet(facet).as_ref() {
        DatasetHandle::F32(d) => d.count as u64,
        DatasetHandle::I32(d) => d.count as u64,
        DatasetHandle::Ivvec32(d) => d.count as u64,
        DatasetHandle::Generic(d) => d.count as u64,
        other => panic!(
            "{who}: expected a facet handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

/// The `f32` record at `index`, as the typed vector it is.
fn f32_vec_typed(h: &DatasetHandle, index: usize) -> crate::ast::SliceArc<f32> {
    match h {
        DatasetHandle::F32(d) => slice_arc_from_uniform(d, index),
        other => panic!(
            "expected F32 dataset handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

/// The `i32` record at `index`, as the typed vector it is.
fn i32_vec_typed(h: &DatasetHandle, index: usize) -> crate::ast::SliceArc<i32> {
    match h {
        DatasetHandle::I32(d) => slice_arc_from_uniform(d, index),
        other => panic!(
            "expected I32 dataset handle, got {}",
            dataset_handle_kind(other)
        ),
    }
}

fn ivvec32_vec_typed(h: &DatasetHandle, index: usize) -> crate::ast::SliceArc<i32> {
    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 crate::ast::SliceArc::from_vec(Vec::<i32>::new());
            }
            let v = d.reader.get(index % d.count).unwrap_or_default();
            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`. One generic helper serves every element type — 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",
    }
}

/// Access an `f32` vector by index, returning a typed `VecF32`.
/// Reads the base facet of a source string or prebuffered handle,
/// and whatever facet an opened F32 handle names.
///
/// Signature: `vector_at(handle, index: u64) -> VecF32`
#[crate::polydat_node(category = RealData)]
fn vector_at(handle: Facet<BaseFacet>, index: u64) -> crate::ast::SliceArc<f32> {
    f32_vec_typed(
        handle.resolve_facet(BaseFacet::FACET).as_ref(),
        index as usize,
    )
}

/// Access a query vector by index. Reads the query facet of a
/// source string or prebuffered handle, as `vector_at` reads the
/// base facet.
///
/// Signature: `query_vector_at(handle, index: u64) -> VecF32`
#[crate::polydat_node(category = RealData)]
fn query_vector_at(handle: Facet<QueryFacet>, index: u64) -> crate::ast::SliceArc<f32> {
    f32_vec_typed(
        handle.resolve_facet(QueryFacet::FACET).as_ref(),
        index as usize,
    )
}

/// Access ground-truth neighbor indices for a query. Expects an
/// I32 handle.
///
/// Signature: `neighbor_indices_at(handle, index: u64) -> VecI32`
#[crate::polydat_node(category = RealData)]
fn neighbor_indices_at(
    handle: Facet<NeighborIndicesFacet>,
    index: u64,
) -> crate::ast::SliceArc<i32> {
    i32_vec_typed(
        handle.resolve_facet(NeighborIndicesFacet::FACET).as_ref(),
        index as usize,
    )
}

/// Access ground-truth neighbor distances for a query. Expects an
/// F32 handle.
///
/// Signature: `neighbor_distances_at(handle, index: u64) -> VecF32`
#[crate::polydat_node(category = RealData)]
fn neighbor_distances_at(
    handle: Facet<NeighborDistancesFacet>,
    index: u64,
) -> crate::ast::SliceArc<f32> {
    f32_vec_typed(
        handle.resolve_facet(NeighborDistancesFacet::FACET).as_ref(),
        index as usize,
    )
}

/// Access filtered ground-truth neighbor indices. Expects an I32 handle.
#[crate::polydat_node(category = RealData)]
fn filtered_neighbor_indices_at(
    handle: Facet<FilteredNeighborIndicesFacet>,
    index: u64,
) -> crate::ast::SliceArc<i32> {
    i32_vec_typed(
        handle
            .resolve_facet(FilteredNeighborIndicesFacet::FACET)
            .as_ref(),
        index as usize,
    )
}

/// Access filtered ground-truth neighbor distances. Expects an F32 handle.
#[crate::polydat_node(category = RealData)]
fn filtered_neighbor_distances_at(
    handle: Facet<FilteredNeighborDistancesFacet>,
    index: u64,
) -> crate::ast::SliceArc<f32> {
    f32_vec_typed(
        handle
            .resolve_facet(FilteredNeighborDistancesFacet::FACET)
            .as_ref(),
        index as usize,
    )
}

// =================================================================
// 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.

/// Return the dimensionality (`f32` count per record) of a
/// vector facet handle.
///
/// Signature: `vector_dim(handle) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn vector_dim(handle: Facet<BaseFacet>) -> u64 {
    match handle.resolve_facet(BaseFacet::FACET).as_ref() {
        DatasetHandle::F32(d) => d.dim as u64,
        DatasetHandle::I32(d) => d.dim as u64,
        _ => 0,
    }
}

/// 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. A source string or prebuffered handle counts
/// the base facet.
///
/// Signature: `vector_count(handle) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn vector_count(handle: Facet<BaseFacet>) -> u64 {
    facet_record_count(&handle, "vector_count", BaseFacet::FACET)
}

/// Return the count of query vectors. Counts the query facet of a
/// source string or prebuffered handle, as `vector_count` counts
/// the base facet.
///
/// Signature: `query_count(handle) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn query_count(handle: Facet<QueryFacet>) -> u64 {
    // `query_count(prebuffered)` is a legitimate idiom in the
    // pvs_query workload (`cursor q = range(0, query_count(...))`),
    // so the prebuffered case resolves the query facet and reads
    // its count, through the same OnceCache-backed open the
    // per-cycle accessors use.
    facet_record_count(&handle, "query_count", QueryFacet::FACET)
}

/// Return the per-record neighbor count (k) for an I32
/// neighbor-indices handle.
///
/// Signature: `neighbor_count(handle) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn neighbor_count(handle: Facet<NeighborIndicesFacet>) -> u64 {
    match handle.resolve_facet(NeighborIndicesFacet::FACET).as_ref() {
        DatasetHandle::I32(d) => d.dim as u64,
        _ => 0,
    }
}

/// Return the dataset's distance function (e.g., "COSINE",
/// "EUCLIDEAN"). Operates on the dataset group; takes a group or
/// prebuffered handle.
///
/// Signature: `dataset_distance_function(group) -> (String)`
#[crate::polydat_node(category = RealData)]
fn dataset_distance_function(group: Group) -> String {
    let group = group.resolve_group();
    let raw = group
        .attribute("distance_function")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    match raw.to_uppercase().as_str() {
        "L2" | "EUCLIDEAN" => "EUCLIDEAN",
        "L1" | "MANHATTAN" => "MANHATTAN",
        "COSINE" => "COSINE",
        "DOT_PRODUCT" | "DOTPRODUCT" | "DOT" | "INNER_PRODUCT" | "IP" => "DOT_PRODUCT",
        _ => raw,
    }
    .to_string()
}

// =================================================================
// 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'")
        })
    }
}

/// 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`
#[crate::polydat_node(category = RealData)]
fn metadata_results_at(
    handle: Facet<MetadataResultsFacet>,
    index: u64,
) -> crate::ast::SliceArc<i32> {
    ivvec32_vec_typed(
        handle.resolve_facet(MetadataResultsFacet::FACET).as_ref(),
        index as usize,
    )
}

/// 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)`
#[crate::polydat_node(category = RealData)]
fn metadata_results_len_at(handle: Facet<MetadataResultsFacet>, index: u64) -> u64 {
    match handle.resolve_facet(MetadataResultsFacet::FACET).as_ref() {
        DatasetHandle::Ivvec32(d) if d.count > 0 => {
            d.reader.dim_at(index as usize % d.count).unwrap_or(0) as u64
        }
        _ => 0,
    }
}

/// Return the metadata indices count (number of predicate result
/// sets). Expects an Ivvec32 handle.
#[crate::polydat_node(category = RealData)]
fn metadata_results_count(handle: Facet<MetadataResultsFacet>) -> u64 {
    match handle.resolve_facet(MetadataResultsFacet::FACET).as_ref() {
        DatasetHandle::Ivvec32(d) => d.count as u64,
        _ => 0,
    }
}

/// Report which facets are available for the default profile of
/// a dataset group as a comma-separated list. Expects a group or
/// prebuffered handle.
///
/// Signature: `dataset_facets(group) -> (String)`
#[crate::polydat_node(category = RealData)]
fn dataset_facets(group: Group) -> String {
    let group = group.resolve_group();
    // 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 names
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>()
            .join(", ");
    }
    String::new()
}

// =================================================================
// 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.

/// Total number of profiles in a dataset group.
///
/// Signature: `dataset_profile_count(group) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn dataset_profile_count(group: Group) -> u64 {
    group.resolve_group().profile_names().len() as u64
}

/// Comma-separated list of all profile names in canonical sort
/// order (by base_count).
///
/// Signature: `dataset_profile_names(group) -> (String)`
#[crate::polydat_node(category = RealData)]
fn dataset_profile_names(group: Group) -> String {
    group.resolve_group().profile_names().join(", ")
}

/// 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
/// [`DatasetHandle::resolve_group`] (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.resolve_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.resolve_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.resolve_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.resolve_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 (a partition source,
/// cursor_partitions.md §7.1). 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.resolve_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.resolve_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
// =================================================================

/// Eagerly download all facets for a dataset profile into the
/// local cache, returning a dataset handle that every facet
/// accessor and every group accessor takes as its 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. A binding such as
/// `init prebuffered = dataset_prebuffer(...)` whose result
/// nothing reads is pruned, so a workload threads the handle
/// through its accessors.
///
/// `source` is the canonical `dataset:profile` string. The
/// returned handle names the dataset rather than one facet: a
/// facet accessor opens its own facet of `source` from it
/// ([`DatasetHandle::resolve_facet`]), and a group accessor
/// (`dataset_facets`, `dataset_distance_function`, ...) reads
/// the dataset's group ([`DatasetHandle::resolve_group`]).
///
/// A dataset that does not resolve yields no handle, and the
/// resolve error goes to the audit log. A profile the dataset
/// lacks, or a facet whose download fails, is logged there too
/// and still yields the handle: prebuffering is a best-effort
/// warm-up, and an accessor that then reads a missing facet
/// fails with that facet's own error.
#[crate::polydat_node(category = RealData)]
fn dataset_prebuffer(source: &str) -> Option<Arc<dyn std::any::Any + Send + Sync>> {
    // The const binding contract (evaluation_model.md §"Const
    // Binding Contract") means this runs exactly once per scope
    // activation: the activation kernel pulls the binding and the host propagates the result to every
    // fiber. PREBUFFER_CACHE is belt-and-suspenders: it serializes
    // anyone who reaches here by another path, so a per-source
    // thundering herd is structurally impossible whatever the caller
    // does.
    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 one
        // download, not per-caller noise.
        crate::library::support::audit::record_prebuffer_entered(source);
        do_dataset_prebuffer_inner(source)
    }) {
        Ok(handle) => Some(handle as Arc<dyn std::any::Any + Send + Sync>),
        Err(_) => None,
    }
}

/// Inner body of [`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> {
    // Every exit after the group resolves returns a prebuffered
    // handle, so the downstream `*_at(prebuffered, q)` accessors can
    // resolve. A failed download still hands back the handle — the
    // accessors then read through the transport, 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.
/// The scalar of a Generic facet at `idx`, rendered.
fn generic_str_typed(h: &DatasetHandle, idx: usize) -> String {
    match h {
        DatasetHandle::Generic(d) => d.format_scalar(idx),
        _ => String::new(),
    }
}

/// 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)`
#[crate::polydat_node(category = RealData)]
fn metadata_value_at(handle: Facet<MetadataContentFacet>, index: u64) -> String {
    generic_str_typed(
        handle.resolve_facet(MetadataContentFacet::FACET).as_ref(),
        index as usize,
    )
}

/// 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)`
#[crate::polydat_node(category = RealData)]
fn predicate_value_at(handle: Facet<MetadataPredicatesFacet>, index: u64) -> String {
    generic_str_typed(
        handle
            .resolve_facet(MetadataPredicatesFacet::FACET)
            .as_ref(),
        index as usize,
    )
}

/// Count of metadata content records. Expects a Generic handle.
///
/// Signature: `metadata_content_count(handle) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn metadata_content_count(handle: Facet<MetadataContentFacet>) -> u64 {
    match handle.resolve_facet(MetadataContentFacet::FACET).as_ref() {
        DatasetHandle::Generic(d) => d.count as u64,
        _ => 0,
    }
}

// ── Counting a facet's matching records ─────────────────────────────

/// How many scalars of a Generic facet equal `value`.
///
/// A linear read of the facet, which is what the question is: the
/// reader holds the values and nothing indexes them by content. Both
/// callers below pass a handle and a value that are fixed for a scope,
/// so the count is computed once at scope init and reused, the way the
/// facet itself is loaded once.
/// How many scalars of a Generic facet equal `value`.
fn generic_count_typed(h: &DatasetHandle, value: i64) -> u64 {
    match h {
        DatasetHandle::Generic(d) => {
            (0..d.count).filter(|i| d.get_scalar(*i) == value).count() as u64
        }
        _ => 0,
    }
}

/// How many records carry the metadata value `value`.
///
/// The per-label cardinality the recall audit compares against a
/// profile's declared count: not a counter advancing as records are
/// visited, but a property of the facet and the value, so it
/// replays and agrees on every engine and every fiber
/// (docs/design/runtime_model.md §9.1).
///
/// `metadata_value_at(handle, i)` formats the same scalar as text,
/// so a host comparing labels as strings and one counting them
/// here are reading one facet.
#[crate::polydat_node(category = RealData)]
fn metadata_count_of(handle: Facet<MetadataContentFacet>, value: i64) -> u64 {
    generic_count_typed(
        handle.resolve_facet(MetadataContentFacet::FACET).as_ref(),
        value,
    )
}

/// How many queries carry the predicate value `value`.
///
/// The denominator of a per-predicate recall figure, and the bound
/// on a rank within one predicate's queries.
///
/// Signature: `predicate_count_of(handle, value: i64) -> (u64)`
#[crate::polydat_node(category = RealData)]
fn predicate_count_of(handle: Facet<MetadataPredicatesFacet>, value: i64) -> u64 {
    generic_count_typed(
        handle
            .resolve_facet(MetadataPredicatesFacet::FACET)
            .as_ref(),
        value,
    )
}

// =========================================================================
// Cursor-sugar handlers
// =========================================================================
//
// 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 in (handle, index) order. The
    // combined source string auto-promotes to the right facet handle
    // through the wire's default resolver (library_catalog.md
    // "Shapes", `Resolved<R, T>`).
    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",
    }
}

#[cfg(test)]
mod count_of_tests {
    use super::*;
    use crate::ast::{PolydatNode, PortType, Slot};
    use crate::dsl::registry::DefaultResolver;

    /// The ports the compiler types a call against: a handle and a
    /// signed value in, a count out. The value is `i64` because that is
    /// what the facet stores; a caller with a literal reaches it
    /// through `to_i64`, since an integer literal is a `u64` and the
    /// adapter catalog will not heal that pair.
    #[test]
    fn the_count_of_nodes_take_a_handle_and_a_signed_value() {
        for node in [
            Box::new(MetadataCountOf::new()) as Box<dyn PolydatNode>,
            Box::new(PredicateCountOf::new()) as Box<dyn PolydatNode>,
        ] {
            let meta = node.meta();
            assert_eq!(meta.outs.len(), 1);
            assert_eq!(meta.outs[0].typ, PortType::U64, "{}", meta.name);
            assert_eq!(meta.ins.len(), 2, "{}", meta.name);
            let Slot::Wire(handle) = &meta.ins[0] else {
                panic!("{}: the first input is a wire", meta.name)
            };
            let Slot::Wire(value) = &meta.ins[1] else {
                panic!("{}: the second input is a wire", meta.name)
            };
            assert_eq!(handle.typ, PortType::Handle, "{}", meta.name);
            assert_eq!(value.typ, PortType::I64, "{}", meta.name);
        }
    }

    /// A handle of the wrong shape answers zero rather than failing.
    /// The facets these read are scalar ones; a caller that opened a
    /// vector facet by mistake gets a count of nothing, which is the
    /// same answer the facet's own readers give for a shape they do not
    /// hold.
    #[test]
    fn a_handle_of_another_shape_counts_nothing() {
        let group_shaped = DatasetHandle::Prebuffered {
            group: match load_dataset_group("nonexistent:profile") {
                Ok(g) => g,
                // No dataset available in this environment: the
                // fallback is still exercised through the match arm
                // below, which is what this test is about.
                Err(_) => return,
            },
            source: "nonexistent:profile".into(),
        };
        assert_eq!(generic_count_typed(&group_shaped, 1), 0);
    }

    /// Both names resolve through the registry and carry a default
    /// resolver, so `metadata_count_of("ds:profile", v)` promotes the
    /// source string to the right facet the way its neighbours do.
    #[test]
    fn both_names_are_registered_with_a_facet_resolver() {
        for (name, facet) in [
            ("metadata_count_of", "metadata_content"),
            ("predicate_count_of", "metadata_predicates"),
        ] {
            let sig = crate::dsl::registry::lookup(name)
                .unwrap_or_else(|| panic!("{name} is registered"));
            assert_eq!(sig.outputs, 1, "{name}");
            assert_eq!(sig.params.len(), 2, "{name}");
            // The facet comes from the handle argument's type, so
            // this is the resolver the macro read off `Facet<_>`.
            match sig.default_resolver {
                Some(DefaultResolver::Facet(f)) => assert_eq!(f, facet, "{name}"),
                other => panic!("{name}: expected a facet resolver, got {other:?}"),
            }
        }
    }
}