Struct Snapshot 

pub struct Snapshot { /* private fields */ }

An immutable collection root opened independently of collection refs.

Implementations

impl Snapshot

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

Imports a materialized canonical tree into an isolated temporary object database and opens the computed root. The source directory is not changed.

pub fn root(&self) -> ObjectId

Returns the deterministic Git tree ID that identifies this snapshot.

Examples found in repository

examples/lancedb_git_vdb_profile.rs (line 133)

fn mutate(
    input: &Path,
    repository: &Path,
    build_report: &Path,
    fraction: f64,
    output: &Path,
    sample_stable: bool,
) -> 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 = if sample_stable {
        sample_stable_points(points, count)?
    } else {
        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,
            "sample_stable": sample_stable,
            "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 sample_stable_points(
    points: Vec<Point>,
    count: usize,
) -> Result<Vec<Point>, Box<dyn std::error::Error>> {
    let mut sample_order = points
        .iter()
        .map(|point| Ok((uint_id_digest(&point.id)?, point.id.clone())))
        .collect::<Result<Vec<_>, Box<dyn std::error::Error>>>()?;
    sample_order.sort();
    let sample_ids = sample_order
        .into_iter()
        .take(8_192.min(points.len()))
        .map(|(_, id)| id)
        .collect::<HashSet<_>>();
    let selected = points
        .into_iter()
        .filter(|point| !sample_ids.contains(&point.id))
        .take(count)
        .collect::<Vec<_>>();
    if selected.len() != count {
        return Err(format!(
            "sample-stable mutation requested {count} points but only {} are outside the training sample",
            selected.len()
        )
        .into());
    }
    Ok(selected)
}

fn uint_id_digest(id: &PointId) -> Result<[u8; 32], Box<dyn std::error::Error>> {
    let PointId::UInt(value) = id else {
        return Err("sample-stable profile expects generated unsigned IDs".into());
    };
    let mut bytes = [0_u8; 10];
    bytes[..2].copy_from_slice(b"u\0");
    bytes[2..].copy_from_slice(&value.to_be_bytes());
    Ok(Sha256::digest(bytes).into())
}

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(())
}

examples/lancedb_git_vdb_runner.rs (line 104)

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 info(&self) -> Result<SnapshotInfo>

Returns configuration and point-count metadata for this root.

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

Retrieves canonically ordered records from this immutable root.

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

Counts all points or those matching a filter at this immutable root.

pub fn query(&self, query: Query) -> Result<QueryResult>

Executes an exact or deterministic approximate query at this root.

The first exact query may populate a root-scoped immutable search view; this cache does not alter persisted objects or the root ID.

Examples found in repository

examples/lancedb_git_vdb_runner.rs (lines 269-278)

fn query_all(
    snapshot: &git_vdb::Snapshot,
    queries: &[Vec<f32>],
    limit: usize,
    exact: bool,
    filter: Option<Filter>,
) -> git_vdb::Result<QueryBatch> {
    let mut durations = Vec::with_capacity(queries.len());
    let mut results = Vec::with_capacity(queries.len());
    let mut vectors_scored = Vec::with_capacity(queries.len());
    for vector in queries {
        let started = Instant::now();
        let result = snapshot.query(Query {
            vector: vector.clone(),
            limit,
            filter: filter.clone(),
            params: QueryParams {
                exact: Some(exact),
                ..QueryParams::default()
            },
            ..Query::default()
        })?;
        durations.push(micros(started));
        vectors_scored.push(result.stats.vectors_scored);
        results.push(Value::Array(
            result
                .points
                .into_iter()
                .map(|point| json!({"id": point.id, "score": point.score}))
                .collect(),
        ));
    }
    Ok((durations, results, vectors_scored))
}

fn query_collection_all(
    collection: &git_vdb::Collection,
    queries: &[Vec<f32>],
    limit: usize,
    exact: bool,
    filter: Option<Filter>,
) -> git_vdb::Result<QueryBatch> {
    let mut durations = Vec::with_capacity(queries.len());
    let mut results = Vec::with_capacity(queries.len());
    let mut vectors_scored = Vec::with_capacity(queries.len());
    for vector in queries {
        let started = Instant::now();
        let result = collection.query(Query {
            vector: vector.clone(),
            limit,
            filter: filter.clone(),
            params: QueryParams {
                exact: Some(exact),
                ..QueryParams::default()
            },
            ..Query::default()
        })?;
        durations.push(micros(started));
        vectors_scored.push(result.stats.vectors_scored);
        results.push(Value::Array(
            result
                .points
                .into_iter()
                .map(|point| json!({"id": point.id, "score": point.score}))
                .collect(),
        ));
    }
    Ok((durations, results, vectors_scored))
}

fn query_snapshot_throughput(
    snapshot: &git_vdb::Snapshot,
    queries: &[Vec<f32>],
    limit: usize,
    exact: bool,
    concurrencies: &[usize],
) -> git_vdb::Result<Map<String, Value>> {
    let mut measurements = Map::new();
    for &workers in concurrencies {
        if workers == 0 {
            return Err(git_vdb::Error::Invalid(
                "benchmark concurrency must be positive".into(),
            ));
        }
        let started = Instant::now();
        let outcome: git_vdb::Result<()> = thread::scope(|scope| {
            let mut handles = Vec::with_capacity(workers);
            for worker in 0..workers {
                let snapshot = snapshot.clone();
                handles.push(scope.spawn(move || -> git_vdb::Result<()> {
                    for vector in queries.iter().skip(worker).step_by(workers) {
                        snapshot.query(Query {
                            vector: vector.clone(),
                            limit,
                            params: QueryParams {
                                exact: Some(exact),
                                ..QueryParams::default()
                            },
                            ..Query::default()
                        })?;
                    }
                    Ok(())
                }));
            }
            join_query_workers(handles)
        });
        outcome?;
        measurements.insert(
            workers.to_string(),
            throughput_measurement(queries.len(), micros(started)),
        );
    }
    Ok(measurements)
}

examples/lancedb_git_vdb_profile.rs (line 258)

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 apply(&self, mutations: Vec<SnapshotMutation>) -> Result<Snapshot>

Applies mutations using this snapshot’s object database without creating collection history or refs.

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

Validates this root without changing its object database.

Full validation recomputes every approximate-index bucket.

pub fn materialize(&self, target: impl AsRef<Path>) -> Result<()>

Writes this exact Git tree as ordinary files and directories.

The target must not already exist. A sibling staging directory is renamed into place only after every object has been read successfully.

Trait Implementations

impl Clone for Snapshot

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for Snapshot

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

Formats the value using the given formatter.