genegraph_storage/traits/

backend.rs

use arrow::array::{Array as ArrowArray, FixedSizeListArray, Float64Array, UInt32Array};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use log::{debug, info, trace};
use smartcore::linalg::basic::arrays::Array;
use smartcore::linalg::basic::matrix::DenseMatrix;
use sprs::{CsMat, TriMat};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::graph::{GraphEdge, GraphWriteOptions, StoredGraph};
use crate::metadata::GeneMetadata;
use crate::{StorageError, StorageResult};

/// Async storage backend for Lance-based graph and embedding data.
///
/// This trait defines the minimal async API required to persist and reload
/// all artifacts used by Javelin:
///
/// - Dense matrices (embeddings, eigenmaps, energy maps)
/// - Sparse matrices in CSR form (e.g. Laplacians, adjacency)
/// - Scalar vectors (eigenvalues, norms, generic f64 sequences)
/// - Index-like vectors (usize mappings and cluster assignments)
/// - Clustering metadata (centroid maps, subcentroids, lambdas)
/// - Global metadata describing the dataset layout and dimensions
///
/// ## Initialization
///
/// Storage must be initialized before saving any data:
///
/// 1. Call `save_metadata()` once to write an initial `*_metadata.json`.
/// 2. Subsequent `save_*` calls validate that metadata exists and is consistent.
/// 3. `exists()` can be used to detect and reuse an existing initialized store.
///
/// Filenames are conventionally:
///
/// ```ignore
/// <base dir>/<instance name or name id>_<key>.lance
/// ```
///
/// ## Async usage
///
/// All I/O functions are async and intended to be called from a Tokio runtime.
/// Implementations (e.g. `LanceStorage`) must not create their own runtimes or
/// block on I/O internally.
///
/// ## High-level flow
///
/// - Dense data:
///   - `save_dense("raw_input", &matrix, md_path)`
///   - `load_dense("raw_input")`
///
/// - Sparse data:
///   - `save_sparse("laplacian", &csr, md_path)`
///   - `load_sparse("laplacian")`
///
/// - Scalars and indices:
///   - `save_lambdas`, `load_lambdas`
///   - `save_vector`, `load_vector`
///   - `save_index`, `load_index`
///
/// Clustering artifacts (centroid maps, subcentroids, item norms, cluster
/// assignments) live on the [`crate::traits::clustering::ClusteringArtifacts`]
/// sub-trait (#117): the base backend stays domain-neutral, and general-purpose
/// implementors do not stub ML-pipeline vocabulary.
///
/// Implementations are free to choose the on-disk layout as long as they honor
/// these logical keys and round-trip semantics.
///
/// ## Extension policy (#117-2)
///
/// The core trait carries only domain-neutral, generic columnar surface.
/// Consumer-specific vocabulary (clustering pipelines, ML pipelines) belongs
/// on extension sub-traits, so adding surface there never breaks external
/// `StorageBackend` implementors. Core-trait additions must provide a default
/// implementation whenever one can be derived from existing methods.
pub trait StorageBackend: Send + Sync {
    /// Base directory of the instance
    fn get_base(&self) -> String;
    /// Name of the instance
    fn get_name(&self) -> String;

    ///
    /// Returns `true` and the path to the metadata file if metadata file exists and is valid,
    /// `false` otherwise.
    /// This is used to avoid overwriting existing indexes.
    fn exists(path: &str) -> (bool, Option<PathBuf>) {
        let base_path = std::path::PathBuf::from(path);
        if !base_path.exists() {
            debug!("StorageBackend: path {:?} does not exist", base_path);
            return (false, None);
        }

        // Check for any _metadata.json file in the directory
        if let Ok(entries) = std::fs::read_dir(&base_path) {
            for entry in entries.flatten() {
                let path = entry.path();
                if let Some(name) = path.file_name().and_then(|n| n.to_str())
                    && name.ends_with("_metadata.json")
                {
                    debug!("StorageBackend::exists: found metadata file at {:?}", path);
                    return (true, Some(path));
                }
            }
        }
        (false, None)
    }

    /// Returns the base directory path.
    fn base_path(&self) -> PathBuf;
    /// Returns the metadata path.
    fn metadata_path(&self) -> PathBuf;
    /// return the base path as file:// string
    fn basepath_to_uri(&self) -> StorageResult<String>;

    /// Load initial data using columnar format from a file path.
    /// Implementations may use this as a helper for async `load_dense`.
    ///
    /// Supported parquet layouts: `vector: FixedSizeList<Float64>` (as
    /// written by [`Self::save_dense_to_file`]) and the legacy wide layout
    /// with `Float64` columns named `col_0..col_N`; anything else is rejected
    /// with [`StorageError::Invalid`]. Supported lance layout: the vector
    /// layout, read from the dataset directory at `path`.
    async fn load_dense_from_file(&self, path: &Path) -> StorageResult<DenseMatrix<f64>>;

    /// Compute the full Lance/parquet file path for a logical filetype.
    fn file_path(&self, key: &str) -> PathBuf;

    /// Converts a full file path to a `file://` URI for Lance.
    ///
    /// The path must be absolute; relative paths are rejected instead of
    /// being silently joined against a guessed working directory.
    /// Non-existing paths are allowed (saves write to fresh locations) and
    /// fall back to the given absolute path; any other resolution failure
    /// (permissions, symlink loops, ...) is surfaced as an error instead of
    /// being silently replaced by the unresolved path.
    fn path_to_uri(path: &Path) -> StorageResult<String> {
        let resolved = match path.canonicalize() {
            Ok(canonical) => canonical,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => path.to_path_buf(),
            Err(e) => {
                return Err(StorageError::Io(format!(
                    "Failed to resolve path `{}`: {}",
                    path.display(),
                    e
                )));
            }
        };
        if !resolved.is_absolute() {
            return Err(StorageError::Invalid(format!(
                "cannot convert relative path {:?} to a file:// URI; pass an absolute path",
                path
            )));
        }
        url::Url::from_file_path(&resolved)
            .map(|u| u.to_string())
            .map_err(|_| {
                StorageError::Invalid(format!("cannot express {:?} as a file:// URI", resolved))
            })
    }

    /// Validates that the storage directory is properly initialized with metadata.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` if metadata file exists, otherwise returns an error.
    fn validate_initialized(&self, md_path: &Path) -> StorageResult<()> {
        let expected = self.metadata_path();
        if expected != *md_path {
            return Err(StorageError::InvalidState(format!(
                "metadata path mismatch: expected `{}`, found `{}`",
                expected.display(),
                md_path.display()
            )));
        }
        if !md_path.exists() {
            return Err(StorageError::Invalid(format!(
                "Storage not initialized: metadata file missing at {:?}. \
                 Call save_metadata() or save_eigenmaps_all()/save_energymaps_all() first.",
                md_path
            )));
        }
        Ok(())
    }

    // =========
    // ASYNC API
    // =========

    /// Converts a dense matrix to a RecordBatch in vector format (Lance-optimized).
    /// Each row of the matrix becomes a single FixedSizeList entry.
    ///
    /// Arguments:
    /// * matrix - Dense matrix to convert (N rows × F cols)
    ///
    /// Returns:
    /// RecordBatch with schema: { vector: FixedSizeList<Float64>[F] }
    fn to_dense_record_batch(
        &self,
        matrix: &DenseMatrix<f64>,
    ) -> Result<RecordBatch, StorageError> {
        let (rows, cols) = (matrix.shape().0, matrix.shape().1);

        debug!(
            "Converting dense matrix to RecordBatch (vector format): {}x{}",
            rows, cols
        );

        if rows == 0 || cols == 0 {
            return Err(StorageError::Invalid(
                "Cannot convert empty matrix to RecordBatch".to_string(),
            ));
        }

        // Flatten matrix row-by-row into a single Vec<f64>
        let mut values: Vec<f64> = Vec::with_capacity(rows * cols);
        for r in 0..rows {
            for c in 0..cols {
                values.push(*matrix.get((r, c)));
            }
        }

        // Create FixedSizeList field: each entry is a vector of length cols
        let value_field = Field::new("item", DataType::Float64, false);
        let list_field = Field::new(
            "vector",
            DataType::FixedSizeList(Arc::new(value_field), cols as i32),
            false,
        );

        let schema = Schema::new(vec![list_field]);

        // Build the FixedSizeList array
        let values_array = Float64Array::from(values);
        let list_array = FixedSizeListArray::new(
            Arc::new(Field::new("item", DataType::Float64, false)),
            cols as i32,
            Arc::new(values_array),
            None, // No nulls
        );

        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(list_array)])
            .map_err(|e| StorageError::Lance(e.to_string()))?;

        trace!(
            "RecordBatch created with {} rows (vectors of length {})",
            batch.num_rows(),
            cols
        );

        Ok(batch)
    }

    /// Reconstructs a dense matrix from a RecordBatch in vector format.
    ///
    /// Arguments:
    /// * batch - RecordBatch containing FixedSizeList<Float64> vectors
    ///
    /// Returns:
    /// DenseMatrix in column-major format (smartcore convention)
    #[allow(clippy::wrong_self_convention)]
    fn from_dense_record_batch(
        &self,
        batch: &RecordBatch,
    ) -> Result<DenseMatrix<f64>, StorageError> {
        use std::mem;

        debug!("Reconstructing dense matrix from RecordBatch (vector format)");
        debug!("Batch has {} columns", batch.num_columns());

        if batch.num_columns() != 1 {
            return Err(StorageError::Invalid(format!(
                "Expected Lance row-major format with 1 FixedSizeList<Float64> column, but found {} columns. \
                  This parquet file appears to be in wide format (feature-per-column). \
                  Convert it first using: \
                  `python -c \"import pyarrow.parquet as pq; import pyarrow.compute as pc; \
                  tbl = pq.read_table('input.parquet'); \
                  import pyarrow as pa; \
                  vectors = pa.array([row.as_py() for row in tbl.to_pylist()], type=pa.list_(pa.float64(), len(tbl.column_names))); \
                  new_tbl = pa.table({{'vector': vectors}}); \
                  pq.write_table(new_tbl, 'output.parquet')\"` \
                  or use a Lance-native writer in your data pipeline.",
                batch.num_columns()
            )));
        }

        debug!("Extracting FixedSizeList column");
        let column = batch.column(0);
        let list_array = column
            .as_any()
            .downcast_ref::<FixedSizeListArray>()
            .ok_or_else(|| {
                StorageError::Invalid(format!(
                    "Column 0 is not FixedSizeList (found type: {:?}). \
                      Expected Lance row-major format with a single FixedSizeList<Float64> column.",
                    column.data_type()
                ))
            })?;

        let rows = list_array.len();
        let cols = list_array.value_length() as usize;

        debug!("Matrix dimensions: {}x{}", rows, cols);

        // Guard against excessive allocations
        let total = rows
            .checked_mul(cols)
            .ok_or_else(|| StorageError::Invalid("Matrix size overflow (rows*cols)".to_string()))?;
        let bytes = total
            .checked_mul(mem::size_of::<f64>())
            .ok_or_else(|| StorageError::Invalid("Byte size overflow".to_string()))?;

        const MAX_BYTES: usize = 4usize * 1024 * 1024 * 1024; // 4 GiB
        if bytes > MAX_BYTES {
            return Err(StorageError::Invalid(format!(
                "Dense load would allocate {} bytes for {}x{} matrix; exceeds 4GiB cap. \
                  Enable --reduce-dim or shard your input data.",
                bytes, rows, cols
            )));
        }

        // Extract Float64 values
        let values_array = list_array
            .values()
            .as_any()
            .downcast_ref::<Float64Array>()
            .ok_or_else(|| {
                StorageError::Invalid("FixedSizeList values are not Float64Array".to_string())
            })?;

        debug!("Converting row-major to column-major");
        let mut data = vec![0.0f64; total];
        for r in 0..rows {
            for c in 0..cols {
                let row_major_idx = r * cols + c;
                let col_major_idx = c * rows + r;
                data[col_major_idx] = values_array.value(row_major_idx);
            }
        }

        debug!("Creating DenseMatrix");
        DenseMatrix::new(rows, cols, data, true).map_err(|e| StorageError::Invalid(e.to_string()))
    }

    /// Converts a sparse CSR matrix to a RecordBatch in columnar format.
    ///
    /// Only non-zero entries are stored.
    fn to_sparse_record_batch(&self, m: &CsMat<f64>) -> StorageResult<RecordBatch> {
        debug!(
            "Converting sparse matrix to RecordBatch: {} x {}, nnz={}",
            m.rows(),
            m.cols(),
            m.nnz()
        );

        let mut row_idx = Vec::with_capacity(m.nnz());
        let mut col_idx = Vec::with_capacity(m.nnz());
        let mut vals = Vec::with_capacity(m.nnz());

        for (v, (r, c)) in m.iter() {
            row_idx.push(r as u32);
            col_idx.push(c as u32);
            vals.push(*v);
        }

        // Store actual dimensions in schema metadata
        let mut schema_metadata = std::collections::HashMap::new();
        schema_metadata.insert("rows".to_string(), m.rows().to_string());
        schema_metadata.insert("cols".to_string(), m.cols().to_string());
        schema_metadata.insert("nnz".to_string(), m.nnz().to_string());

        let schema = Schema::new(vec![
            Field::new("row", DataType::UInt32, false),
            Field::new("col", DataType::UInt32, false),
            Field::new("value", DataType::Float64, false),
        ])
        .with_metadata(schema_metadata);

        let batch = RecordBatch::try_new(
            Arc::new(schema),
            vec![
                Arc::new(UInt32Array::from(row_idx)) as _,
                Arc::new(UInt32Array::from(col_idx)) as _,
                Arc::new(Float64Array::from(vals)) as _,
            ],
        )
        .map_err(|e| StorageError::Lance(e.to_string()))?;

        trace!(
            "Sparse RecordBatch created with {} entries",
            batch.num_rows()
        );
        Ok(batch)
    }

    /// Reconstructs a sparse CSR matrix from a RecordBatch in columnar format.
    ///
    /// * `batch` - RecordBatch containing (`row`, `col`, `value`) triplets
    /// * `expected_rows` / `expected_cols` - dimensions taken from metadata
    #[allow(clippy::wrong_self_convention)]
    fn from_sparse_record_batch(
        &self,
        batch: RecordBatch,
        expected_rows: usize,
        expected_cols: usize,
    ) -> StorageResult<CsMat<f64>> {
        debug!("Reconstructing sparse matrix from RecordBatch");

        let row_arr = batch
            .column(0)
            .as_any()
            .downcast_ref::<UInt32Array>()
            .ok_or_else(|| StorageError::Invalid("row column type mismatch".into()))?;
        let col_arr = batch
            .column(1)
            .as_any()
            .downcast_ref::<UInt32Array>()
            .ok_or_else(|| StorageError::Invalid("col column type mismatch".into()))?;
        let val_arr = batch
            .column(2)
            .as_any()
            .downcast_ref::<Float64Array>()
            .ok_or_else(|| StorageError::Invalid("value column type mismatch".into()))?;

        let n = row_arr.len();
        if n == 0 {
            debug!(
                "Empty RecordBatch, returning {}x{} sparse matrix",
                expected_rows, expected_cols
            );
            return Ok(CsMat::zero((expected_rows, expected_cols)));
        }

        // Try to read dimensions from schema metadata (for validation)
        let schema = batch.schema();
        let schema_metadata = schema.metadata();
        if let (Some(rows_str), Some(cols_str)) =
            (schema_metadata.get("rows"), schema_metadata.get("cols"))
        {
            let schema_rows = rows_str.parse::<usize>().ok();
            let schema_cols = cols_str.parse::<usize>().ok();
            if schema_rows != Some(expected_rows) || schema_cols != Some(expected_cols) {
                return Err(StorageError::DimensionMismatch {
                    expected: format!("{}x{}", expected_rows, expected_cols),
                    found: match (schema_rows, schema_cols) {
                        (Some(r), Some(c)) => format!("{}x{}", r, c),
                        _ => format!(
                            "unparseable schema metadata {:?}x{:?}",
                            schema_rows, schema_cols
                        ),
                    },
                });
            } else {
                debug!(
                    "Schema metadata matches storage metadata: {}x{}",
                    expected_rows, expected_cols
                );
            }
        }

        let rows = expected_rows;
        let cols = expected_cols;
        debug!(
            "Reconstructing {}x{} sparse matrix from {} entries",
            rows, cols, n
        );

        let mut trimat = TriMat::new((rows, cols));
        for i in 0..n {
            let r = row_arr.value(i) as usize;
            let c = col_arr.value(i) as usize;
            let v = val_arr.value(i);

            if r >= rows || c >= cols {
                return Err(StorageError::Invalid(format!(
                    "Index out of bounds: ({}, {}) in {}x{} matrix",
                    r, c, rows, cols
                )));
            }
            trimat.add_triplet(r, c, v);
        }

        let result = trimat.to_csr();
        if result.rows() != rows || result.cols() != cols {
            return Err(StorageError::Invalid(format!(
                "Dimension mismatch after reconstruction: expected {}x{}, got {}x{}",
                rows,
                cols,
                result.rows(),
                result.cols()
            )));
        }

        Ok(result)
    }

    /// Saves a dense matrix. Requires metadata to exist.
    async fn save_dense(
        &self,
        key: &str,
        matrix: &DenseMatrix<f64>,
        md_path: &Path,
    ) -> StorageResult<()>;

    /// Loads a dense matrix from storage.
    async fn load_dense(&self, key: &str) -> StorageResult<DenseMatrix<f64>>;

    /// Saves a sparse matrix. Requires metadata to exist.
    async fn save_sparse(
        &self,
        key: &str,
        matrix: &CsMat<f64>,
        md_path: &Path,
    ) -> StorageResult<()>;

    /// Loads a sparse matrix from storage.
    async fn load_sparse(&self, key: &str) -> StorageResult<CsMat<f64>>;

    /// Saves lambda eigenvalues. Requires metadata to exist.
    async fn save_lambdas(&self, lambdas: &[f64], md_path: &Path) -> StorageResult<()>;

    /// Loads lambda eigenvalues from storage.
    async fn load_lambdas(&self) -> StorageResult<Vec<f64>>;

    /// Initializes storage by saving metadata. Must be called first.
    async fn save_metadata(&self, metadata: &GeneMetadata) -> StorageResult<PathBuf> {
        let path = self.metadata_path();
        info!("Saving metadata to {:?}", path);
        let s = serde_json::to_string_pretty(metadata).map_err(StorageError::Serde)?;
        // Atomic publish (#93): tmp + fsync + rename — the file is the
        // single commit pointer and must never be observed half-written.
        crate::generations::write_json_atomic(&path, &s)?;
        info!("Metadata saved successfully");
        Ok(path)
    }

    /// Loads metadata from storage.
    async fn load_metadata(&self) -> StorageResult<GeneMetadata> {
        let filename = self.metadata_path();
        info!("Loading metadata from {:?}", filename);
        let s = tokio::fs::read_to_string(filename)
            .await
            .map_err(|e| StorageError::Io(e.to_string()))?;
        let md: GeneMetadata = serde_json::from_str(&s).map_err(StorageError::Serde)?;
        info!("Metadata loaded successfully");
        Ok(md)
    }

    /// Save vectors that are not lambdas but indices.
    #[allow(dead_code)]
    async fn save_index(&self, key: &str, vector: &[usize], md_path: &Path) -> StorageResult<()>;

    /// save a generic f64 sequence
    async fn save_vector(&self, key: &str, vector: &[f64], md_path: &Path) -> StorageResult<()>;

    /// Load index or generic usize vector from storage.
    #[allow(dead_code)]
    async fn load_index(&self, key: &str) -> StorageResult<Vec<usize>>;

    async fn load_vector(&self, key: &str) -> StorageResult<Vec<f64>>;

    /// Writes a dense matrix to `path` (`.lance` dataset directory or
    /// `.parquet` file in the `vector: FixedSizeList<Float64>` layout, the
    /// counterpart of [`Self::load_dense_from_file`]). Parent directories are
    /// created as needed.
    async fn save_dense_to_file(data: &DenseMatrix<f64>, path: &Path) -> StorageResult<()>;

    // =========
    // NAMED COLLECTIONS (RFC #81)
    // =========

    /// Saves a schema-driven vector-space collection (RFC #81-P2) without
    /// user properties; see [`Self::save_vectors_with`].
    ///
    /// `save_dense` remains as the fixed-key `f64` wrapper for compat.
    async fn save_vectors(
        &self,
        name: &str,
        batch: &RecordBatch,
        md_path: &Path,
    ) -> StorageResult<()> {
        self.save_vectors_with(name, batch, &Default::default(), md_path)
            .await
    }

    /// Saves a vector-space collection with user properties (e.g. the
    /// `graph` linkage of RFC #81-P4: `properties.graph = <name>`).
    async fn save_vectors_with(
        &self,
        name: &str,
        batch: &RecordBatch,
        properties: &std::collections::BTreeMap<String, String>,
        md_path: &Path,
    ) -> StorageResult<()>;

    /// Loads a vector-space collection back as a `RecordBatch` with its
    /// original schema (including collection metadata).
    async fn load_vectors(&self, name: &str) -> StorageResult<RecordBatch>;

    /// Saves an edge-list graph collection (RFC #81-P3) with `u32` node ids
    /// (the default width) and `f64` weights (the default width, mirroring
    /// the sparse-matrix `value` column). Ids above `u32::MAX` surface
    /// [`StorageError::Overflow`] instead of being truncated. Weights are
    /// persisted faithfully (normalization transforms belong to the
    /// producer); the opt-in `GraphWriteOptions::weight_range` asserts
    /// compliance with a declared closed interval, and weights that cannot
    /// be stored exactly at an explicitly declared
    /// [`crate::graph::WeightType::F32`] width are rejected instead of
    /// being silently narrowed.
    ///
    /// The collection is either fully weighted (`weight: Float64|Float32`
    /// column, schema-declared via `GraphWriteOptions::weight_type`) or
    /// topology-only (`src`/`dst` only); mixed edges are rejected.
    async fn save_graph(
        &self,
        name: &str,
        edges: &[GraphEdge],
        md_path: &Path,
    ) -> StorageResult<()> {
        self.save_graph_with(name, edges, &GraphWriteOptions::default(), md_path)
            .await
    }

    /// Saves a graph collection with explicit options (node-id and weight
    /// widths, node count, user properties for vector-space linkage,
    /// RFC #81-P4).
    async fn save_graph_with(
        &self,
        name: &str,
        edges: &[GraphEdge],
        options: &GraphWriteOptions,
        md_path: &Path,
    ) -> StorageResult<()>;

    /// Loads a graph collection back as an edge list; convert to CSR at the
    /// API boundary with [`StoredGraph::to_csr`].
    async fn load_graph(&self, name: &str) -> StorageResult<StoredGraph>;

    // =========
    // REGISTRY-FREE COLLECTION I/O (#106)
    // =========

    /// Saves a vector-space collection dataset at an explicit `path`
    /// without touching the metadata registry (#106): the batch is
    /// validated, stamped with dataset-level collection metadata
    /// (including user properties) and written; parent directories are
    /// created as needed. Registry ownership stays with the caller — the
    /// counterpart of [`Self::save_vectors_with`] for consumers that run
    /// their own metadata registry (e.g. an ArrowSpaceMetadata commit
    /// pointer at the instance metadata path). No `GeneMetadata` read or
    /// write occurs.
    async fn save_vectors_to_path(
        &self,
        path: &Path,
        batch: &RecordBatch,
        properties: &std::collections::BTreeMap<String, String>,
    ) -> StorageResult<()>;

    /// Loads a vector-space collection dataset from an explicit `path`
    /// (registry-free): the path-based counterpart of
    /// [`Self::load_vectors`].
    async fn load_vectors_from_path(&self, path: &Path) -> StorageResult<RecordBatch>;

    /// Saves a graph collection dataset at an explicit `path` without
    /// touching the metadata registry (#106); see
    /// [`Self::save_graph_with`] for the options and
    /// [`Self::save_vectors_to_path`] for the registry-free contract.
    async fn save_graph_to_path(
        &self,
        path: &Path,
        edges: &[GraphEdge],
        options: &GraphWriteOptions,
    ) -> StorageResult<()>;

    /// Loads a graph collection dataset from an explicit `path`
    /// (registry-free): the path-based counterpart of [`Self::load_graph`].
    async fn load_graph_from_path(&self, path: &Path) -> StorageResult<StoredGraph>;

    /// Loads a scalar collection — a single `Float64|Float32` column, e.g.
    /// lambdas or norms — as `Vec<f64>` (#106); `f32` values are upcast
    /// losslessly. The stamped dataset kind must be `vector-space`
    /// (missing or mismatched kinds are rejected: physical shape
    /// determines decodability, the logical kind determines semantic
    /// validity), making every `kind=vector-space` collection uniformly
    /// loadable instead of requiring the legacy fixed-key readers.
    async fn load_scalars(&self, name: &str) -> StorageResult<Vec<f64>>;

    // =========
    // #107: registry-free scalar collections
    // =========

    /// Saves a scalar collection — a single non-nullable `Float64` column
    /// (e.g. lambdas or norms) — at an explicit `path` without touching the
    /// metadata registry (#107): the dataset is stamped `kind=vector-space`
    /// and parent directories are created as needed. Registry ownership
    /// stays with the caller; no `GeneMetadata` read or write occurs. The
    /// counterpart of [`Self::save_vectors_to_path`] for scalar-only
    /// collections (which `save_vectors_to_path` cannot write, as its
    /// schema validation requires a `FixedSizeList` column).
    async fn save_scalars_to_path(&self, path: &Path, values: &[f64]) -> StorageResult<()>;

    /// Loads a scalar collection from an explicit `path` (registry-free):
    /// the path-based counterpart of [`Self::load_scalars`], with the same
    /// kind gate. `load_scalars` delegates to this.
    async fn load_scalars_from_path(&self, path: &Path) -> StorageResult<Vec<f64>>;

    // =========
    // #108: strict graph read mode
    // =========

    /// Loads a graph collection from an explicit `path` with read options
    /// (#108): strict mode rejects datasets without a `num_nodes` stamp
    /// (and, optionally, validates column names) instead of silently
    /// resizing to `max_id + 1`. The tolerant default is the compat shim.
    async fn load_graph_from_path_with_options(
        &self,
        path: &Path,
        options: &crate::graph::GraphReadOptions,
    ) -> StorageResult<StoredGraph>;

    /// Strict variant of [`Self::load_graph_from_path`] (#108): rejects
    /// datasets without a `num_nodes` stamp with a typed
    /// [`StorageError::Invalid`] and validates the `src`/`dst`/`weight`
    /// column names, so pre-collections triplet artifacts are positively
    /// identified rather than positionally coerced.
    ///
    /// Provided default (#117-2): derivable from
    /// [`Self::load_graph_from_path_with_options`], so implementors only
    /// need the options-based method.
    async fn load_graph_from_path_strict(&self, path: &Path) -> StorageResult<StoredGraph> {
        self.load_graph_from_path_with_options(path, &crate::graph::GraphReadOptions::strict())
            .await
    }

    // =========
    // #109: metadata-only schema read
    // =========

    /// Reads the dataset schema (fields + KV metadata) of a collection at
    /// an explicit `path` without decoding any column buffers (#109): the
    /// backend surface of [`crate::lancefmt::read_schema`], so consumers
    /// doing kind/stamp gating don't pay a full columnar scan.
    async fn collection_schema_from_path(&self, path: &Path) -> StorageResult<Schema>;

    /// Reads the dataset schema of a named collection (registry-free,
    /// path-resolved): the name-based counterpart of
    /// [`Self::collection_schema_from_path`].
    async fn collection_schema(&self, name: &str) -> StorageResult<Schema> {
        self.collection_schema_from_path(&self.file_path(name))
            .await
    }

    // =========
    // #115: dataset verification
    // =========

    /// Verifies the stamped dataset schema of a collection at `path`
    /// against `expected` (#115): reads the schema footer via
    /// [`crate::lancefmt::read_schema`] — no column buffers are decoded —
    /// and cross-checks the facts (`kind`, `rows`/`cols`, `num_nodes`,
    /// weight width). Mismatches surface as typed errors naming the fact.
    ///
    /// Provided (#117-2 policy): derivable from the lancefmt primitive, so
    /// no implementor burden. Expectations derive from the registry via
    /// [`crate::lancefmt::verify::DatasetFacts::from_file_info`] or
    /// `from_descriptor`.
    async fn verify_collection_from_path(
        &self,
        path: &Path,
        expected: &crate::lancefmt::verify::DatasetFacts,
    ) -> StorageResult<()> {
        let path = path.to_path_buf();
        let expected = expected.clone();
        tokio::task::spawn_blocking(move || {
            crate::lancefmt::verify_dataset_schema(&path, &expected)
        })
        .await
        .map_err(|e| StorageError::Io(format!("verify task failed: {e}")))?
    }

    /// Name-based counterpart of [`Self::verify_collection_from_path`],
    /// resolving through [`Self::file_path`].
    async fn verify_collection(
        &self,
        name: &str,
        expected: &crate::lancefmt::verify::DatasetFacts,
    ) -> StorageResult<()> {
        self.verify_collection_from_path(&self.file_path(name), expected)
            .await
    }
}