Struct SnapshotEngine 

pub struct SnapshotEngine { /* private fields */ }

A ref-free engine for deterministic immutable collection roots.

The engine writes and reads Git objects but never creates a commit, updates a ref, consults repository history, or reads the clock. Callers are responsible for retaining returned roots, for example through an external content store or the named collection adapter.

Implementations

impl SnapshotEngine

pub fn open(path: impl AsRef<Path>) -> Result<Self>

Opens an existing bare or non-bare Git object database without reading a collection ref.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 84)

fn mutate(
    input: &Path,
    repository: &Path,
    build_report: &Path,
    fraction: f64,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let spec = read_spec(input)?;
    if !(0.0..=1.0).contains(&fraction) || fraction == 0.0 {
        return Err("mutation fraction must be greater than zero and at most one".into());
    }
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let points = make_points(&vectors);
    let count = ((spec.point_count as f64 * fraction).round() as usize).clamp(1, spec.point_count);
    let mut changed = points[..count].to_vec();
    for point in &mut changed {
        point.vector[0] += 0.001;
    }
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;

    let upsert_started = Instant::now();
    let upserted = engine.apply(
        root,
        changed.into_iter().map(SnapshotMutation::upsert).collect(),
    )?;
    let upsert_us = micros(upsert_started);
    let delete_started = Instant::now();
    let deleted = engine.apply(
        root,
        vec![SnapshotMutation::delete_ids(
            (0..count).map(|id| PointId::from(id as u64)),
        )],
    )?;
    let delete_us = micros(delete_started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "fraction": fraction,
            "points": count,
            "upsert_us": upsert_us,
            "delete_us": delete_us,
            "upsert_root": upserted.root(),
            "delete_root": deleted.root(),
            "on_disk_bytes_after": directory_bytes(repository)?,
        }))?,
    )?;
    Ok(())
}

fn validate(
    repository: &Path,
    build_report: &Path,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;
    let started = Instant::now();
    let report = engine.validate(root, true)?;
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "validation_us": micros(started),
            "report": report,
        }))?,
    )?;
    Ok(())
}

fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
    if repository.exists() {
        return Err(format!("repository already exists: {}", repository.display()).into());
    }
    let spec = read_spec(input)?;
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let points = make_points(&vectors);
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };
    let engine = SnapshotEngine::init(repository)?;
    let started = Instant::now();
    let snapshot = engine.build(config, points)?;
    let build_us = micros(started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": snapshot.root(),
            "build_us": build_us,
            "on_disk_bytes": directory_bytes(repository)?,
        }))?,
    )?;
    Ok(())
}

fn query(
    input: &Path,
    repository: &Path,
    build_report: &Path,
    mode: &str,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let (exact, warm_exact) = match mode {
        "exact" => (true, false),
        "approximate" => (false, false),
        "approximate-after-exact" => (false, true),
        _ => return Err(format!("unsupported query mode: {mode}").into()),
    };
    let spec = read_spec(input)?;
    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;
    let snapshot = engine.open_snapshot(root)?;

    let cache_build_us = if warm_exact {
        let started = Instant::now();
        snapshot.query(make_query(&queries[0], maximum_k, true))?;
        Some(micros(started))
    } else {
        None
    };
    // Fill the immutable snapshot cache, construct an approximate lookup over an
    // already-warm exact view, or warm the unchanged approximate ODB path without
    // including that one-time work in the samples.
    let warmup_started = Instant::now();
    snapshot.query(make_query(&queries[0], maximum_k, exact))?;
    let warmup_us = micros(warmup_started);
    wait_for_profiler()?;

    let mut query_us = Vec::with_capacity(queries.len());
    let mut results = Vec::with_capacity(queries.len());
    let mut vectors_scored = Vec::with_capacity(queries.len());
    let batch_started = Instant::now();
    for vector in &queries {
        let started = Instant::now();
        let result = snapshot.query(make_query(vector, maximum_k, exact))?;
        query_us.push(micros(started));
        vectors_scored.push(result.stats.vectors_scored);
        results.push(result.points);
    }
    let batch_us = micros(batch_started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "mode": mode,
            "cache_build_us": cache_build_us,
            "warmup_us": warmup_us,
            "query_us": query_us,
            "batch_us": batch_us,
            "vectors_scored": vectors_scored,
            "results": results,
        }))?,
    )?;
    Ok(())
}

pub fn init(path: impl AsRef<Path>) -> Result<Self>

Initializes a bare Git object database for immutable snapshots.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 153)

fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
    if repository.exists() {
        return Err(format!("repository already exists: {}", repository.display()).into());
    }
    let spec = read_spec(input)?;
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let points = make_points(&vectors);
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };
    let engine = SnapshotEngine::init(repository)?;
    let started = Instant::now();
    let snapshot = engine.build(config, points)?;
    let build_us = micros(started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": snapshot.root(),
            "build_us": build_us,
            "on_disk_bytes": directory_bytes(repository)?,
        }))?,
    )?;
    Ok(())
}

examples/lancedb_git_vdb_runner.rs (line 50)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args = env::args_os().skip(1);
    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    if args.next().is_some() {
        return Err("usage: runner INPUT.json OUTPUT.json".into());
    }
    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
    if spec.schema_version != 1 {
        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
    }
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
    let points = make_points(&vectors);
    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };

    let setup_started = Instant::now();
    let core_dir = tempfile::TempDir::new()?;
    let engine = SnapshotEngine::init(core_dir.path())?;
    let setup_us = micros(setup_started);
    let build_started = Instant::now();
    let snapshot = engine.build(config.clone(), points.clone())?;
    let build_us = micros(build_started);
    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;

    let (exact_query_us, exact_results, exact_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, true, None)?;
    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, false, None)?;
    let snapshot_throughput = json!({
        "exact": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });

    let mut filtered = Map::new();
    for selectivity in &spec.filter_selectivities {
        let filter = selectivity_filter(*selectivity);
        let (exact_us, exact, exact_scored) =
            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
        let (approximate_us, approximate, approximate_scored) =
            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
        filtered.insert(
            selectivity.to_string(),
            json!({
                "exact_query_us": exact_us,
                "approximate_query_us": approximate_us,
                "exact_results": exact,
                "approximate_results": approximate,
                "exact_vectors_scored": exact_scored,
                "approximate_vectors_scored": approximate_scored,
            }),
        );
    }

    let mut mutations = Map::new();
    for fraction in &spec.mutation_fractions {
        let count = fraction_count(spec.point_count, *fraction);
        let changed = changed_points(&points[..count]);
        let upsert_started = Instant::now();
        let upserted = engine.apply(
            snapshot.root(),
            changed.into_iter().map(SnapshotMutation::upsert).collect(),
        )?;
        let upsert_us = micros(upsert_started);
        let delete_started = Instant::now();
        let deleted = engine.apply(
            snapshot.root(),
            vec![SnapshotMutation::delete_ids(
                (0..count).map(|id| PointId::from(id as u64)),
            )],
        )?;
        let delete_us = micros(delete_started);
        mutations.insert(
            fraction.to_string(),
            json!({
                "points": count,
                "upsert_us": upsert_us,
                "delete_us": delete_us,
                "upsert_root": upserted.root(),
                "delete_root": deleted.root(),
            }),
        );
    }

    let adapter_setup_started = Instant::now();
    let adapter_dir = tempfile::TempDir::new()?;
    let database = Database::init_bare(adapter_dir.path())?;
    let collection = database.create_collection("benchmark", config)?;
    let adapter_setup_us = micros(adapter_setup_started);
    let adapter_build_started = Instant::now();
    let adapter_root = collection.upsert(points)?.root;
    let adapter_build_us = micros(adapter_build_started);
    if adapter_root != snapshot.root() {
        return Err("snapshot-core and named-adapter roots differ".into());
    }
    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
        query_collection_all(&collection, &queries, maximum_k, true, None)?;
    let (
        adapter_approximate_query_us,
        adapter_approximate_results,
        adapter_approximate_vectors_scored,
    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
    let adapter_throughput = json!({
        "exact": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });
    let historical_started = Instant::now();
    let historical = collection.at(&adapter_root)?;
    let historical_count = historical.count(None)?.count;
    let historical_read_us = micros(historical_started);

    let report = json!({
        "schema_version": 1,
        "engine": "git-vdb",
        "case_name": spec.case_name,
        "point_count": spec.point_count,
        "dimension": spec.dimension,
        "query_count": spec.query_count,
        "k": spec.k,
        "root": snapshot.root(),
        "setup_us": setup_us,
        "snapshot_core": {
            "build_us": build_us,
            "exact_query_us": exact_query_us,
            "approximate_query_us": approximate_query_us,
            "exact_results": exact_results,
            "approximate_results": approximate_results,
            "exact_vectors_scored": exact_vectors_scored,
            "approximate_vectors_scored": approximate_vectors_scored,
            "throughput": snapshot_throughput,
            "filtered": filtered,
            "mutations": mutations,
            "on_disk_bytes": baseline_on_disk_bytes,
        },
        "named_adapter": {
            "setup_us": adapter_setup_us,
            "build_us": adapter_build_us,
            "exact_query_us": adapter_exact_query_us,
            "approximate_query_us": adapter_approximate_query_us,
            "exact_results": adapter_exact_results,
            "approximate_results": adapter_approximate_results,
            "exact_vectors_scored": adapter_exact_vectors_scored,
            "approximate_vectors_scored": adapter_approximate_vectors_scored,
            "throughput": adapter_throughput,
            "historical_read_us": historical_read_us,
            "historical_count": historical_count,
            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
        }
    });
    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
    Ok(())
}

pub fn ephemeral() -> Result<Self>

Creates an isolated temporary object database whose lifetime is retained by the engine and snapshots returned from it.

pub fn build( &self, config: CollectionConfig, points: Vec<Point>, ) -> Result<Snapshot>

Builds a canonical root from a complete point set.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 155)

fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
    if repository.exists() {
        return Err(format!("repository already exists: {}", repository.display()).into());
    }
    let spec = read_spec(input)?;
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let points = make_points(&vectors);
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };
    let engine = SnapshotEngine::init(repository)?;
    let started = Instant::now();
    let snapshot = engine.build(config, points)?;
    let build_us = micros(started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": snapshot.root(),
            "build_us": build_us,
            "on_disk_bytes": directory_bytes(repository)?,
        }))?,
    )?;
    Ok(())
}

examples/lancedb_git_vdb_runner.rs (line 53)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args = env::args_os().skip(1);
    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    if args.next().is_some() {
        return Err("usage: runner INPUT.json OUTPUT.json".into());
    }
    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
    if spec.schema_version != 1 {
        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
    }
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
    let points = make_points(&vectors);
    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };

    let setup_started = Instant::now();
    let core_dir = tempfile::TempDir::new()?;
    let engine = SnapshotEngine::init(core_dir.path())?;
    let setup_us = micros(setup_started);
    let build_started = Instant::now();
    let snapshot = engine.build(config.clone(), points.clone())?;
    let build_us = micros(build_started);
    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;

    let (exact_query_us, exact_results, exact_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, true, None)?;
    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, false, None)?;
    let snapshot_throughput = json!({
        "exact": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });

    let mut filtered = Map::new();
    for selectivity in &spec.filter_selectivities {
        let filter = selectivity_filter(*selectivity);
        let (exact_us, exact, exact_scored) =
            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
        let (approximate_us, approximate, approximate_scored) =
            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
        filtered.insert(
            selectivity.to_string(),
            json!({
                "exact_query_us": exact_us,
                "approximate_query_us": approximate_us,
                "exact_results": exact,
                "approximate_results": approximate,
                "exact_vectors_scored": exact_scored,
                "approximate_vectors_scored": approximate_scored,
            }),
        );
    }

    let mut mutations = Map::new();
    for fraction in &spec.mutation_fractions {
        let count = fraction_count(spec.point_count, *fraction);
        let changed = changed_points(&points[..count]);
        let upsert_started = Instant::now();
        let upserted = engine.apply(
            snapshot.root(),
            changed.into_iter().map(SnapshotMutation::upsert).collect(),
        )?;
        let upsert_us = micros(upsert_started);
        let delete_started = Instant::now();
        let deleted = engine.apply(
            snapshot.root(),
            vec![SnapshotMutation::delete_ids(
                (0..count).map(|id| PointId::from(id as u64)),
            )],
        )?;
        let delete_us = micros(delete_started);
        mutations.insert(
            fraction.to_string(),
            json!({
                "points": count,
                "upsert_us": upsert_us,
                "delete_us": delete_us,
                "upsert_root": upserted.root(),
                "delete_root": deleted.root(),
            }),
        );
    }

    let adapter_setup_started = Instant::now();
    let adapter_dir = tempfile::TempDir::new()?;
    let database = Database::init_bare(adapter_dir.path())?;
    let collection = database.create_collection("benchmark", config)?;
    let adapter_setup_us = micros(adapter_setup_started);
    let adapter_build_started = Instant::now();
    let adapter_root = collection.upsert(points)?.root;
    let adapter_build_us = micros(adapter_build_started);
    if adapter_root != snapshot.root() {
        return Err("snapshot-core and named-adapter roots differ".into());
    }
    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
        query_collection_all(&collection, &queries, maximum_k, true, None)?;
    let (
        adapter_approximate_query_us,
        adapter_approximate_results,
        adapter_approximate_vectors_scored,
    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
    let adapter_throughput = json!({
        "exact": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });
    let historical_started = Instant::now();
    let historical = collection.at(&adapter_root)?;
    let historical_count = historical.count(None)?.count;
    let historical_read_us = micros(historical_started);

    let report = json!({
        "schema_version": 1,
        "engine": "git-vdb",
        "case_name": spec.case_name,
        "point_count": spec.point_count,
        "dimension": spec.dimension,
        "query_count": spec.query_count,
        "k": spec.k,
        "root": snapshot.root(),
        "setup_us": setup_us,
        "snapshot_core": {
            "build_us": build_us,
            "exact_query_us": exact_query_us,
            "approximate_query_us": approximate_query_us,
            "exact_results": exact_results,
            "approximate_results": approximate_results,
            "exact_vectors_scored": exact_vectors_scored,
            "approximate_vectors_scored": approximate_vectors_scored,
            "throughput": snapshot_throughput,
            "filtered": filtered,
            "mutations": mutations,
            "on_disk_bytes": baseline_on_disk_bytes,
        },
        "named_adapter": {
            "setup_us": adapter_setup_us,
            "build_us": adapter_build_us,
            "exact_query_us": adapter_exact_query_us,
            "approximate_query_us": adapter_approximate_query_us,
            "exact_results": adapter_exact_results,
            "approximate_results": adapter_approximate_results,
            "exact_vectors_scored": adapter_exact_vectors_scored,
            "approximate_vectors_scored": adapter_approximate_vectors_scored,
            "throughput": adapter_throughput,
            "historical_read_us": historical_read_us,
            "historical_count": historical_count,
            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
        }
    });
    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
    Ok(())
}

pub fn apply( &self, previous_root: impl AsRef<str>, mutations: Vec<SnapshotMutation>, ) -> Result<Snapshot>

Applies an ordered mutation batch to a root and returns the new root.

Mutation order is significant. Duplicate upserts for the same typed ID within one call are rejected. No ref or commit is created.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (lines 87-90)

fn mutate(
    input: &Path,
    repository: &Path,
    build_report: &Path,
    fraction: f64,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let spec = read_spec(input)?;
    if !(0.0..=1.0).contains(&fraction) || fraction == 0.0 {
        return Err("mutation fraction must be greater than zero and at most one".into());
    }
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let points = make_points(&vectors);
    let count = ((spec.point_count as f64 * fraction).round() as usize).clamp(1, spec.point_count);
    let mut changed = points[..count].to_vec();
    for point in &mut changed {
        point.vector[0] += 0.001;
    }
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;

    let upsert_started = Instant::now();
    let upserted = engine.apply(
        root,
        changed.into_iter().map(SnapshotMutation::upsert).collect(),
    )?;
    let upsert_us = micros(upsert_started);
    let delete_started = Instant::now();
    let deleted = engine.apply(
        root,
        vec![SnapshotMutation::delete_ids(
            (0..count).map(|id| PointId::from(id as u64)),
        )],
    )?;
    let delete_us = micros(delete_started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "fraction": fraction,
            "points": count,
            "upsert_us": upsert_us,
            "delete_us": delete_us,
            "upsert_root": upserted.root(),
            "delete_root": deleted.root(),
            "on_disk_bytes_after": directory_bytes(repository)?,
        }))?,
    )?;
    Ok(())
}

examples/lancedb_git_vdb_runner.rs (lines 103-106)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args = env::args_os().skip(1);
    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
    if args.next().is_some() {
        return Err("usage: runner INPUT.json OUTPUT.json".into());
    }
    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
    if spec.schema_version != 1 {
        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
    }
    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
    let points = make_points(&vectors);
    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
    let config = CollectionConfig {
        dimension: spec.dimension,
        ..CollectionConfig::default()
    };

    let setup_started = Instant::now();
    let core_dir = tempfile::TempDir::new()?;
    let engine = SnapshotEngine::init(core_dir.path())?;
    let setup_us = micros(setup_started);
    let build_started = Instant::now();
    let snapshot = engine.build(config.clone(), points.clone())?;
    let build_us = micros(build_started);
    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;

    let (exact_query_us, exact_results, exact_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, true, None)?;
    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
        query_all(&snapshot, &queries, maximum_k, false, None)?;
    let snapshot_throughput = json!({
        "exact": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_snapshot_throughput(
            &snapshot,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });

    let mut filtered = Map::new();
    for selectivity in &spec.filter_selectivities {
        let filter = selectivity_filter(*selectivity);
        let (exact_us, exact, exact_scored) =
            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
        let (approximate_us, approximate, approximate_scored) =
            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
        filtered.insert(
            selectivity.to_string(),
            json!({
                "exact_query_us": exact_us,
                "approximate_query_us": approximate_us,
                "exact_results": exact,
                "approximate_results": approximate,
                "exact_vectors_scored": exact_scored,
                "approximate_vectors_scored": approximate_scored,
            }),
        );
    }

    let mut mutations = Map::new();
    for fraction in &spec.mutation_fractions {
        let count = fraction_count(spec.point_count, *fraction);
        let changed = changed_points(&points[..count]);
        let upsert_started = Instant::now();
        let upserted = engine.apply(
            snapshot.root(),
            changed.into_iter().map(SnapshotMutation::upsert).collect(),
        )?;
        let upsert_us = micros(upsert_started);
        let delete_started = Instant::now();
        let deleted = engine.apply(
            snapshot.root(),
            vec![SnapshotMutation::delete_ids(
                (0..count).map(|id| PointId::from(id as u64)),
            )],
        )?;
        let delete_us = micros(delete_started);
        mutations.insert(
            fraction.to_string(),
            json!({
                "points": count,
                "upsert_us": upsert_us,
                "delete_us": delete_us,
                "upsert_root": upserted.root(),
                "delete_root": deleted.root(),
            }),
        );
    }

    let adapter_setup_started = Instant::now();
    let adapter_dir = tempfile::TempDir::new()?;
    let database = Database::init_bare(adapter_dir.path())?;
    let collection = database.create_collection("benchmark", config)?;
    let adapter_setup_us = micros(adapter_setup_started);
    let adapter_build_started = Instant::now();
    let adapter_root = collection.upsert(points)?.root;
    let adapter_build_us = micros(adapter_build_started);
    if adapter_root != snapshot.root() {
        return Err("snapshot-core and named-adapter roots differ".into());
    }
    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
        query_collection_all(&collection, &queries, maximum_k, true, None)?;
    let (
        adapter_approximate_query_us,
        adapter_approximate_results,
        adapter_approximate_vectors_scored,
    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
    let adapter_throughput = json!({
        "exact": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            true,
            &spec.concurrency,
        )?,
        "approximate": query_collection_throughput(
            &collection,
            &queries,
            maximum_k,
            false,
            &spec.concurrency,
        )?,
    });
    let historical_started = Instant::now();
    let historical = collection.at(&adapter_root)?;
    let historical_count = historical.count(None)?.count;
    let historical_read_us = micros(historical_started);

    let report = json!({
        "schema_version": 1,
        "engine": "git-vdb",
        "case_name": spec.case_name,
        "point_count": spec.point_count,
        "dimension": spec.dimension,
        "query_count": spec.query_count,
        "k": spec.k,
        "root": snapshot.root(),
        "setup_us": setup_us,
        "snapshot_core": {
            "build_us": build_us,
            "exact_query_us": exact_query_us,
            "approximate_query_us": approximate_query_us,
            "exact_results": exact_results,
            "approximate_results": approximate_results,
            "exact_vectors_scored": exact_vectors_scored,
            "approximate_vectors_scored": approximate_vectors_scored,
            "throughput": snapshot_throughput,
            "filtered": filtered,
            "mutations": mutations,
            "on_disk_bytes": baseline_on_disk_bytes,
        },
        "named_adapter": {
            "setup_us": adapter_setup_us,
            "build_us": adapter_build_us,
            "exact_query_us": adapter_exact_query_us,
            "approximate_query_us": adapter_approximate_query_us,
            "exact_results": adapter_exact_results,
            "approximate_results": adapter_approximate_results,
            "exact_vectors_scored": adapter_exact_vectors_scored,
            "approximate_vectors_scored": adapter_approximate_vectors_scored,
            "throughput": adapter_throughput,
            "historical_read_us": historical_read_us,
            "historical_count": historical_count,
            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
        }
    });
    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
    Ok(())
}

pub fn open_snapshot(&self, root: impl AsRef<str>) -> Result<Snapshot>

Opens an exact tree object ID without resolving refs or commits.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 191)

fn query(
    input: &Path,
    repository: &Path,
    build_report: &Path,
    mode: &str,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let (exact, warm_exact) = match mode {
        "exact" => (true, false),
        "approximate" => (false, false),
        "approximate-after-exact" => (false, true),
        _ => return Err(format!("unsupported query mode: {mode}").into()),
    };
    let spec = read_spec(input)?;
    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;
    let snapshot = engine.open_snapshot(root)?;

    let cache_build_us = if warm_exact {
        let started = Instant::now();
        snapshot.query(make_query(&queries[0], maximum_k, true))?;
        Some(micros(started))
    } else {
        None
    };
    // Fill the immutable snapshot cache, construct an approximate lookup over an
    // already-warm exact view, or warm the unchanged approximate ODB path without
    // including that one-time work in the samples.
    let warmup_started = Instant::now();
    snapshot.query(make_query(&queries[0], maximum_k, exact))?;
    let warmup_us = micros(warmup_started);
    wait_for_profiler()?;

    let mut query_us = Vec::with_capacity(queries.len());
    let mut results = Vec::with_capacity(queries.len());
    let mut vectors_scored = Vec::with_capacity(queries.len());
    let batch_started = Instant::now();
    for vector in &queries {
        let started = Instant::now();
        let result = snapshot.query(make_query(vector, maximum_k, exact))?;
        query_us.push(micros(started));
        vectors_scored.push(result.stats.vectors_scored);
        results.push(result.points);
    }
    let batch_us = micros(batch_started);
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "mode": mode,
            "cache_build_us": cache_build_us,
            "warmup_us": warmup_us,
            "query_us": query_us,
            "batch_us": batch_us,
            "vectors_scored": vectors_scored,
            "results": results,
        }))?,
    )?;
    Ok(())
}

pub fn import_directory(&self, path: impl AsRef<Path>) -> Result<Snapshot>

Imports a materialized canonical tree into this engine’s object database.

pub fn query(&self, root: impl AsRef<str>, query: Query) -> Result<QueryResult>

Queries an exact root ID without first constructing a named collection.

pub fn get( &self, root: impl AsRef<str>, request: GetRequest, ) -> Result<GetResult>

Retrieves records from an exact root ID.

pub fn count( &self, root: impl AsRef<str>, filter: Option<Filter>, ) -> Result<CountResult>

Counts records in an exact root ID.

pub fn validate( &self, root: impl AsRef<str>, full: bool, ) -> Result<ValidationReport>

Validates an exact root ID.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 129)

fn validate(
    repository: &Path,
    build_report: &Path,
    output: &Path,
) -> Result<(), Box<dyn std::error::Error>> {
    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
    let root = build
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or("build report root is missing")?;
    let engine = SnapshotEngine::open(repository)?;
    let started = Instant::now();
    let report = engine.validate(root, true)?;
    fs::write(
        output,
        serde_json::to_vec_pretty(&json!({
            "schema_version": 1,
            "root": root,
            "validation_us": micros(started),
            "report": report,
        }))?,
    )?;
    Ok(())
}

pub fn build_directory( path: impl AsRef<Path>, config: CollectionConfig, points: Vec<Point>, ) -> Result<Snapshot>

Builds a snapshot without a caller-provided Git repository and writes its canonical files into a new materialized directory.

Trait Implementations

impl Clone for SnapshotEngine

fn clone(&self) -> SnapshotEngine

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for SnapshotEngine

fn fmt(&self, formatter: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.