Skip to main content

SnapshotEngine

git_vdb::snapshot

Struct SnapshotEngine 

Source 
pub struct SnapshotEngine { /* private fields */ }
Expand description

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§

Source§

impl SnapshotEngine

Source

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 107)
79fn mutate(
80    input: &Path,
81    repository: &Path,
82    build_report: &Path,
83    fraction: f64,
84    output: &Path,
85    sample_stable: bool,
86) -> Result<(), Box<dyn std::error::Error>> {
87    let spec = read_spec(input)?;
88    if !(0.0..=1.0).contains(&fraction) || fraction == 0.0 {
89        return Err("mutation fraction must be greater than zero and at most one".into());
90    }
91    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
92    let points = make_points(&vectors);
93    let count = ((spec.point_count as f64 * fraction).round() as usize).clamp(1, spec.point_count);
94    let mut changed = if sample_stable {
95        sample_stable_points(points, count)?
96    } else {
97        points[..count].to_vec()
98    };
99    for point in &mut changed {
100        point.vector[0] += 0.001;
101    }
102    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
103    let root = build
104        .get("root")
105        .and_then(serde_json::Value::as_str)
106        .ok_or("build report root is missing")?;
107    let engine = SnapshotEngine::open(repository)?;
108
109    let upsert_started = Instant::now();
110    let upserted = engine.apply(
111        root,
112        changed.into_iter().map(SnapshotMutation::upsert).collect(),
113    )?;
114    let upsert_us = micros(upsert_started);
115    let delete_started = Instant::now();
116    let deleted = engine.apply(
117        root,
118        vec![SnapshotMutation::delete_ids(
119            (0..count).map(|id| PointId::from(id as u64)),
120        )],
121    )?;
122    let delete_us = micros(delete_started);
123    fs::write(
124        output,
125        serde_json::to_vec_pretty(&json!({
126            "schema_version": 1,
127            "root": root,
128            "fraction": fraction,
129            "points": count,
130            "sample_stable": sample_stable,
131            "upsert_us": upsert_us,
132            "delete_us": delete_us,
133            "upsert_root": upserted.root(),
134            "delete_root": deleted.root(),
135            "on_disk_bytes_after": directory_bytes(repository)?,
136        }))?,
137    )?;
138    Ok(())
139}
140
141fn sample_stable_points(
142    points: Vec<Point>,
143    count: usize,
144) -> Result<Vec<Point>, Box<dyn std::error::Error>> {
145    let mut sample_order = points
146        .iter()
147        .map(|point| Ok((uint_id_digest(&point.id)?, point.id.clone())))
148        .collect::<Result<Vec<_>, Box<dyn std::error::Error>>>()?;
149    sample_order.sort();
150    let sample_ids = sample_order
151        .into_iter()
152        .take(8_192.min(points.len()))
153        .map(|(_, id)| id)
154        .collect::<HashSet<_>>();
155    let selected = points
156        .into_iter()
157        .filter(|point| !sample_ids.contains(&point.id))
158        .take(count)
159        .collect::<Vec<_>>();
160    if selected.len() != count {
161        return Err(format!(
162            "sample-stable mutation requested {count} points but only {} are outside the training sample",
163            selected.len()
164        )
165        .into());
166    }
167    Ok(selected)
168}
169
170fn uint_id_digest(id: &PointId) -> Result<[u8; 32], Box<dyn std::error::Error>> {
171    let PointId::UInt(value) = id else {
172        return Err("sample-stable profile expects generated unsigned IDs".into());
173    };
174    let mut bytes = [0_u8; 10];
175    bytes[..2].copy_from_slice(b"u\0");
176    bytes[2..].copy_from_slice(&value.to_be_bytes());
177    Ok(Sha256::digest(bytes).into())
178}
179
180fn validate(
181    repository: &Path,
182    build_report: &Path,
183    output: &Path,
184) -> Result<(), Box<dyn std::error::Error>> {
185    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
186    let root = build
187        .get("root")
188        .and_then(serde_json::Value::as_str)
189        .ok_or("build report root is missing")?;
190    let engine = SnapshotEngine::open(repository)?;
191    let started = Instant::now();
192    let report = engine.validate(root, true)?;
193    fs::write(
194        output,
195        serde_json::to_vec_pretty(&json!({
196            "schema_version": 1,
197            "root": root,
198            "validation_us": micros(started),
199            "report": report,
200        }))?,
201    )?;
202    Ok(())
203}
204
205fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
206    if repository.exists() {
207        return Err(format!("repository already exists: {}", repository.display()).into());
208    }
209    let spec = read_spec(input)?;
210    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
211    let points = make_points(&vectors);
212    let config = CollectionConfig {
213        dimension: spec.dimension,
214        ..CollectionConfig::default()
215    };
216    let engine = SnapshotEngine::init(repository)?;
217    let started = Instant::now();
218    let snapshot = engine.build(config, points)?;
219    let build_us = micros(started);
220    fs::write(
221        output,
222        serde_json::to_vec_pretty(&json!({
223            "schema_version": 1,
224            "root": snapshot.root(),
225            "build_us": build_us,
226            "on_disk_bytes": directory_bytes(repository)?,
227        }))?,
228    )?;
229    Ok(())
230}
231
232fn query(
233    input: &Path,
234    repository: &Path,
235    build_report: &Path,
236    mode: &str,
237    output: &Path,
238) -> Result<(), Box<dyn std::error::Error>> {
239    let (exact, warm_exact) = match mode {
240        "exact" => (true, false),
241        "approximate" => (false, false),
242        "approximate-after-exact" => (false, true),
243        _ => return Err(format!("unsupported query mode: {mode}").into()),
244    };
245    let spec = read_spec(input)?;
246    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
247    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
248    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
249    let root = build
250        .get("root")
251        .and_then(serde_json::Value::as_str)
252        .ok_or("build report root is missing")?;
253    let engine = SnapshotEngine::open(repository)?;
254    let snapshot = engine.open_snapshot(root)?;
255
256    let cache_build_us = if warm_exact {
257        let started = Instant::now();
258        snapshot.query(make_query(&queries[0], maximum_k, true))?;
259        Some(micros(started))
260    } else {
261        None
262    };
263    // Fill the immutable snapshot cache, construct an approximate lookup over an
264    // already-warm exact view, or warm the unchanged approximate ODB path without
265    // including that one-time work in the samples.
266    let warmup_started = Instant::now();
267    snapshot.query(make_query(&queries[0], maximum_k, exact))?;
268    let warmup_us = micros(warmup_started);
269    wait_for_profiler()?;
270
271    let mut query_us = Vec::with_capacity(queries.len());
272    let mut results = Vec::with_capacity(queries.len());
273    let mut vectors_scored = Vec::with_capacity(queries.len());
274    let batch_started = Instant::now();
275    for vector in &queries {
276        let started = Instant::now();
277        let result = snapshot.query(make_query(vector, maximum_k, exact))?;
278        query_us.push(micros(started));
279        vectors_scored.push(result.stats.vectors_scored);
280        results.push(result.points);
281    }
282    let batch_us = micros(batch_started);
283    fs::write(
284        output,
285        serde_json::to_vec_pretty(&json!({
286            "schema_version": 1,
287            "root": root,
288            "mode": mode,
289            "cache_build_us": cache_build_us,
290            "warmup_us": warmup_us,
291            "query_us": query_us,
292            "batch_us": batch_us,
293            "vectors_scored": vectors_scored,
294            "results": results,
295        }))?,
296    )?;
297    Ok(())
298}
Source

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 216)
205fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
206    if repository.exists() {
207        return Err(format!("repository already exists: {}", repository.display()).into());
208    }
209    let spec = read_spec(input)?;
210    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
211    let points = make_points(&vectors);
212    let config = CollectionConfig {
213        dimension: spec.dimension,
214        ..CollectionConfig::default()
215    };
216    let engine = SnapshotEngine::init(repository)?;
217    let started = Instant::now();
218    let snapshot = engine.build(config, points)?;
219    let build_us = micros(started);
220    fs::write(
221        output,
222        serde_json::to_vec_pretty(&json!({
223            "schema_version": 1,
224            "root": snapshot.root(),
225            "build_us": build_us,
226            "on_disk_bytes": directory_bytes(repository)?,
227        }))?,
228    )?;
229    Ok(())
230}
More examples
Hide additional examples
examples/lancedb_git_vdb_runner.rs (line 50)
28fn main() -> Result<(), Box<dyn std::error::Error>> {
29    let mut args = env::args_os().skip(1);
30    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
31    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
32    if args.next().is_some() {
33        return Err("usage: runner INPUT.json OUTPUT.json".into());
34    }
35    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
36    if spec.schema_version != 1 {
37        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
38    }
39    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
40    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
41    let points = make_points(&vectors);
42    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
43    let config = CollectionConfig {
44        dimension: spec.dimension,
45        ..CollectionConfig::default()
46    };
47
48    let setup_started = Instant::now();
49    let core_dir = tempfile::TempDir::new()?;
50    let engine = SnapshotEngine::init(core_dir.path())?;
51    let setup_us = micros(setup_started);
52    let build_started = Instant::now();
53    let snapshot = engine.build(config.clone(), points.clone())?;
54    let build_us = micros(build_started);
55    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;
56
57    let (exact_query_us, exact_results, exact_vectors_scored) =
58        query_all(&snapshot, &queries, maximum_k, true, None)?;
59    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
60        query_all(&snapshot, &queries, maximum_k, false, None)?;
61    let snapshot_throughput = json!({
62        "exact": query_snapshot_throughput(
63            &snapshot,
64            &queries,
65            maximum_k,
66            true,
67            &spec.concurrency,
68        )?,
69        "approximate": query_snapshot_throughput(
70            &snapshot,
71            &queries,
72            maximum_k,
73            false,
74            &spec.concurrency,
75        )?,
76    });
77
78    let mut filtered = Map::new();
79    for selectivity in &spec.filter_selectivities {
80        let filter = selectivity_filter(*selectivity);
81        let (exact_us, exact, exact_scored) =
82            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
83        let (approximate_us, approximate, approximate_scored) =
84            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
85        filtered.insert(
86            selectivity.to_string(),
87            json!({
88                "exact_query_us": exact_us,
89                "approximate_query_us": approximate_us,
90                "exact_results": exact,
91                "approximate_results": approximate,
92                "exact_vectors_scored": exact_scored,
93                "approximate_vectors_scored": approximate_scored,
94            }),
95        );
96    }
97
98    let mut mutations = Map::new();
99    for fraction in &spec.mutation_fractions {
100        let count = fraction_count(spec.point_count, *fraction);
101        let changed = changed_points(&points[..count]);
102        let upsert_started = Instant::now();
103        let upserted = engine.apply(
104            snapshot.root(),
105            changed.into_iter().map(SnapshotMutation::upsert).collect(),
106        )?;
107        let upsert_us = micros(upsert_started);
108        let delete_started = Instant::now();
109        let deleted = engine.apply(
110            snapshot.root(),
111            vec![SnapshotMutation::delete_ids(
112                (0..count).map(|id| PointId::from(id as u64)),
113            )],
114        )?;
115        let delete_us = micros(delete_started);
116        mutations.insert(
117            fraction.to_string(),
118            json!({
119                "points": count,
120                "upsert_us": upsert_us,
121                "delete_us": delete_us,
122                "upsert_root": upserted.root(),
123                "delete_root": deleted.root(),
124            }),
125        );
126    }
127
128    let adapter_setup_started = Instant::now();
129    let adapter_dir = tempfile::TempDir::new()?;
130    let database = Database::init_bare(adapter_dir.path())?;
131    let collection = database.create_collection("benchmark", config)?;
132    let adapter_setup_us = micros(adapter_setup_started);
133    let adapter_build_started = Instant::now();
134    let adapter_root = collection.upsert(points)?.root;
135    let adapter_build_us = micros(adapter_build_started);
136    if adapter_root != snapshot.root() {
137        return Err("snapshot-core and named-adapter roots differ".into());
138    }
139    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
140    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
141    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
142        query_collection_all(&collection, &queries, maximum_k, true, None)?;
143    let (
144        adapter_approximate_query_us,
145        adapter_approximate_results,
146        adapter_approximate_vectors_scored,
147    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
148    let adapter_throughput = json!({
149        "exact": query_collection_throughput(
150            &collection,
151            &queries,
152            maximum_k,
153            true,
154            &spec.concurrency,
155        )?,
156        "approximate": query_collection_throughput(
157            &collection,
158            &queries,
159            maximum_k,
160            false,
161            &spec.concurrency,
162        )?,
163    });
164    let historical_started = Instant::now();
165    let historical = collection.at(&adapter_root)?;
166    let historical_count = historical.count(None)?.count;
167    let historical_read_us = micros(historical_started);
168
169    let report = json!({
170        "schema_version": 1,
171        "engine": "git-vdb",
172        "case_name": spec.case_name,
173        "point_count": spec.point_count,
174        "dimension": spec.dimension,
175        "query_count": spec.query_count,
176        "k": spec.k,
177        "root": snapshot.root(),
178        "setup_us": setup_us,
179        "snapshot_core": {
180            "build_us": build_us,
181            "exact_query_us": exact_query_us,
182            "approximate_query_us": approximate_query_us,
183            "exact_results": exact_results,
184            "approximate_results": approximate_results,
185            "exact_vectors_scored": exact_vectors_scored,
186            "approximate_vectors_scored": approximate_vectors_scored,
187            "throughput": snapshot_throughput,
188            "filtered": filtered,
189            "mutations": mutations,
190            "on_disk_bytes": baseline_on_disk_bytes,
191        },
192        "named_adapter": {
193            "setup_us": adapter_setup_us,
194            "build_us": adapter_build_us,
195            "exact_query_us": adapter_exact_query_us,
196            "approximate_query_us": adapter_approximate_query_us,
197            "exact_results": adapter_exact_results,
198            "approximate_results": adapter_approximate_results,
199            "exact_vectors_scored": adapter_exact_vectors_scored,
200            "approximate_vectors_scored": adapter_approximate_vectors_scored,
201            "throughput": adapter_throughput,
202            "historical_read_us": historical_read_us,
203            "historical_count": historical_count,
204            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
205        }
206    });
207    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
208    Ok(())
209}
Source

pub fn ephemeral() -> Result<Self>

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

Source

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 218)
205fn build(input: &Path, repository: &Path, output: &Path) -> Result<(), Box<dyn std::error::Error>> {
206    if repository.exists() {
207        return Err(format!("repository already exists: {}", repository.display()).into());
208    }
209    let spec = read_spec(input)?;
210    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
211    let points = make_points(&vectors);
212    let config = CollectionConfig {
213        dimension: spec.dimension,
214        ..CollectionConfig::default()
215    };
216    let engine = SnapshotEngine::init(repository)?;
217    let started = Instant::now();
218    let snapshot = engine.build(config, points)?;
219    let build_us = micros(started);
220    fs::write(
221        output,
222        serde_json::to_vec_pretty(&json!({
223            "schema_version": 1,
224            "root": snapshot.root(),
225            "build_us": build_us,
226            "on_disk_bytes": directory_bytes(repository)?,
227        }))?,
228    )?;
229    Ok(())
230}
More examples
Hide additional examples
examples/lancedb_git_vdb_runner.rs (line 53)
28fn main() -> Result<(), Box<dyn std::error::Error>> {
29    let mut args = env::args_os().skip(1);
30    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
31    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
32    if args.next().is_some() {
33        return Err("usage: runner INPUT.json OUTPUT.json".into());
34    }
35    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
36    if spec.schema_version != 1 {
37        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
38    }
39    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
40    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
41    let points = make_points(&vectors);
42    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
43    let config = CollectionConfig {
44        dimension: spec.dimension,
45        ..CollectionConfig::default()
46    };
47
48    let setup_started = Instant::now();
49    let core_dir = tempfile::TempDir::new()?;
50    let engine = SnapshotEngine::init(core_dir.path())?;
51    let setup_us = micros(setup_started);
52    let build_started = Instant::now();
53    let snapshot = engine.build(config.clone(), points.clone())?;
54    let build_us = micros(build_started);
55    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;
56
57    let (exact_query_us, exact_results, exact_vectors_scored) =
58        query_all(&snapshot, &queries, maximum_k, true, None)?;
59    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
60        query_all(&snapshot, &queries, maximum_k, false, None)?;
61    let snapshot_throughput = json!({
62        "exact": query_snapshot_throughput(
63            &snapshot,
64            &queries,
65            maximum_k,
66            true,
67            &spec.concurrency,
68        )?,
69        "approximate": query_snapshot_throughput(
70            &snapshot,
71            &queries,
72            maximum_k,
73            false,
74            &spec.concurrency,
75        )?,
76    });
77
78    let mut filtered = Map::new();
79    for selectivity in &spec.filter_selectivities {
80        let filter = selectivity_filter(*selectivity);
81        let (exact_us, exact, exact_scored) =
82            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
83        let (approximate_us, approximate, approximate_scored) =
84            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
85        filtered.insert(
86            selectivity.to_string(),
87            json!({
88                "exact_query_us": exact_us,
89                "approximate_query_us": approximate_us,
90                "exact_results": exact,
91                "approximate_results": approximate,
92                "exact_vectors_scored": exact_scored,
93                "approximate_vectors_scored": approximate_scored,
94            }),
95        );
96    }
97
98    let mut mutations = Map::new();
99    for fraction in &spec.mutation_fractions {
100        let count = fraction_count(spec.point_count, *fraction);
101        let changed = changed_points(&points[..count]);
102        let upsert_started = Instant::now();
103        let upserted = engine.apply(
104            snapshot.root(),
105            changed.into_iter().map(SnapshotMutation::upsert).collect(),
106        )?;
107        let upsert_us = micros(upsert_started);
108        let delete_started = Instant::now();
109        let deleted = engine.apply(
110            snapshot.root(),
111            vec![SnapshotMutation::delete_ids(
112                (0..count).map(|id| PointId::from(id as u64)),
113            )],
114        )?;
115        let delete_us = micros(delete_started);
116        mutations.insert(
117            fraction.to_string(),
118            json!({
119                "points": count,
120                "upsert_us": upsert_us,
121                "delete_us": delete_us,
122                "upsert_root": upserted.root(),
123                "delete_root": deleted.root(),
124            }),
125        );
126    }
127
128    let adapter_setup_started = Instant::now();
129    let adapter_dir = tempfile::TempDir::new()?;
130    let database = Database::init_bare(adapter_dir.path())?;
131    let collection = database.create_collection("benchmark", config)?;
132    let adapter_setup_us = micros(adapter_setup_started);
133    let adapter_build_started = Instant::now();
134    let adapter_root = collection.upsert(points)?.root;
135    let adapter_build_us = micros(adapter_build_started);
136    if adapter_root != snapshot.root() {
137        return Err("snapshot-core and named-adapter roots differ".into());
138    }
139    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
140    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
141    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
142        query_collection_all(&collection, &queries, maximum_k, true, None)?;
143    let (
144        adapter_approximate_query_us,
145        adapter_approximate_results,
146        adapter_approximate_vectors_scored,
147    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
148    let adapter_throughput = json!({
149        "exact": query_collection_throughput(
150            &collection,
151            &queries,
152            maximum_k,
153            true,
154            &spec.concurrency,
155        )?,
156        "approximate": query_collection_throughput(
157            &collection,
158            &queries,
159            maximum_k,
160            false,
161            &spec.concurrency,
162        )?,
163    });
164    let historical_started = Instant::now();
165    let historical = collection.at(&adapter_root)?;
166    let historical_count = historical.count(None)?.count;
167    let historical_read_us = micros(historical_started);
168
169    let report = json!({
170        "schema_version": 1,
171        "engine": "git-vdb",
172        "case_name": spec.case_name,
173        "point_count": spec.point_count,
174        "dimension": spec.dimension,
175        "query_count": spec.query_count,
176        "k": spec.k,
177        "root": snapshot.root(),
178        "setup_us": setup_us,
179        "snapshot_core": {
180            "build_us": build_us,
181            "exact_query_us": exact_query_us,
182            "approximate_query_us": approximate_query_us,
183            "exact_results": exact_results,
184            "approximate_results": approximate_results,
185            "exact_vectors_scored": exact_vectors_scored,
186            "approximate_vectors_scored": approximate_vectors_scored,
187            "throughput": snapshot_throughput,
188            "filtered": filtered,
189            "mutations": mutations,
190            "on_disk_bytes": baseline_on_disk_bytes,
191        },
192        "named_adapter": {
193            "setup_us": adapter_setup_us,
194            "build_us": adapter_build_us,
195            "exact_query_us": adapter_exact_query_us,
196            "approximate_query_us": adapter_approximate_query_us,
197            "exact_results": adapter_exact_results,
198            "approximate_results": adapter_approximate_results,
199            "exact_vectors_scored": adapter_exact_vectors_scored,
200            "approximate_vectors_scored": adapter_approximate_vectors_scored,
201            "throughput": adapter_throughput,
202            "historical_read_us": historical_read_us,
203            "historical_count": historical_count,
204            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
205        }
206    });
207    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
208    Ok(())
209}
Source

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 110-113)
79fn mutate(
80    input: &Path,
81    repository: &Path,
82    build_report: &Path,
83    fraction: f64,
84    output: &Path,
85    sample_stable: bool,
86) -> Result<(), Box<dyn std::error::Error>> {
87    let spec = read_spec(input)?;
88    if !(0.0..=1.0).contains(&fraction) || fraction == 0.0 {
89        return Err("mutation fraction must be greater than zero and at most one".into());
90    }
91    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
92    let points = make_points(&vectors);
93    let count = ((spec.point_count as f64 * fraction).round() as usize).clamp(1, spec.point_count);
94    let mut changed = if sample_stable {
95        sample_stable_points(points, count)?
96    } else {
97        points[..count].to_vec()
98    };
99    for point in &mut changed {
100        point.vector[0] += 0.001;
101    }
102    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
103    let root = build
104        .get("root")
105        .and_then(serde_json::Value::as_str)
106        .ok_or("build report root is missing")?;
107    let engine = SnapshotEngine::open(repository)?;
108
109    let upsert_started = Instant::now();
110    let upserted = engine.apply(
111        root,
112        changed.into_iter().map(SnapshotMutation::upsert).collect(),
113    )?;
114    let upsert_us = micros(upsert_started);
115    let delete_started = Instant::now();
116    let deleted = engine.apply(
117        root,
118        vec![SnapshotMutation::delete_ids(
119            (0..count).map(|id| PointId::from(id as u64)),
120        )],
121    )?;
122    let delete_us = micros(delete_started);
123    fs::write(
124        output,
125        serde_json::to_vec_pretty(&json!({
126            "schema_version": 1,
127            "root": root,
128            "fraction": fraction,
129            "points": count,
130            "sample_stable": sample_stable,
131            "upsert_us": upsert_us,
132            "delete_us": delete_us,
133            "upsert_root": upserted.root(),
134            "delete_root": deleted.root(),
135            "on_disk_bytes_after": directory_bytes(repository)?,
136        }))?,
137    )?;
138    Ok(())
139}
More examples
Hide additional examples
examples/lancedb_git_vdb_runner.rs (lines 103-106)
28fn main() -> Result<(), Box<dyn std::error::Error>> {
29    let mut args = env::args_os().skip(1);
30    let input = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
31    let output = args.next().ok_or("usage: runner INPUT.json OUTPUT.json")?;
32    if args.next().is_some() {
33        return Err("usage: runner INPUT.json OUTPUT.json".into());
34    }
35    let spec: RunSpec = serde_json::from_slice(&fs::read(input)?)?;
36    if spec.schema_version != 1 {
37        return Err(format!("unsupported harness schema version {}", spec.schema_version).into());
38    }
39    let vectors = read_vectors(&spec.points_path, spec.point_count, spec.dimension)?;
40    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
41    let points = make_points(&vectors);
42    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
43    let config = CollectionConfig {
44        dimension: spec.dimension,
45        ..CollectionConfig::default()
46    };
47
48    let setup_started = Instant::now();
49    let core_dir = tempfile::TempDir::new()?;
50    let engine = SnapshotEngine::init(core_dir.path())?;
51    let setup_us = micros(setup_started);
52    let build_started = Instant::now();
53    let snapshot = engine.build(config.clone(), points.clone())?;
54    let build_us = micros(build_started);
55    let baseline_on_disk_bytes = directory_bytes(core_dir.path())?;
56
57    let (exact_query_us, exact_results, exact_vectors_scored) =
58        query_all(&snapshot, &queries, maximum_k, true, None)?;
59    let (approximate_query_us, approximate_results, approximate_vectors_scored) =
60        query_all(&snapshot, &queries, maximum_k, false, None)?;
61    let snapshot_throughput = json!({
62        "exact": query_snapshot_throughput(
63            &snapshot,
64            &queries,
65            maximum_k,
66            true,
67            &spec.concurrency,
68        )?,
69        "approximate": query_snapshot_throughput(
70            &snapshot,
71            &queries,
72            maximum_k,
73            false,
74            &spec.concurrency,
75        )?,
76    });
77
78    let mut filtered = Map::new();
79    for selectivity in &spec.filter_selectivities {
80        let filter = selectivity_filter(*selectivity);
81        let (exact_us, exact, exact_scored) =
82            query_all(&snapshot, &queries, maximum_k, true, Some(filter.clone()))?;
83        let (approximate_us, approximate, approximate_scored) =
84            query_all(&snapshot, &queries, maximum_k, false, Some(filter))?;
85        filtered.insert(
86            selectivity.to_string(),
87            json!({
88                "exact_query_us": exact_us,
89                "approximate_query_us": approximate_us,
90                "exact_results": exact,
91                "approximate_results": approximate,
92                "exact_vectors_scored": exact_scored,
93                "approximate_vectors_scored": approximate_scored,
94            }),
95        );
96    }
97
98    let mut mutations = Map::new();
99    for fraction in &spec.mutation_fractions {
100        let count = fraction_count(spec.point_count, *fraction);
101        let changed = changed_points(&points[..count]);
102        let upsert_started = Instant::now();
103        let upserted = engine.apply(
104            snapshot.root(),
105            changed.into_iter().map(SnapshotMutation::upsert).collect(),
106        )?;
107        let upsert_us = micros(upsert_started);
108        let delete_started = Instant::now();
109        let deleted = engine.apply(
110            snapshot.root(),
111            vec![SnapshotMutation::delete_ids(
112                (0..count).map(|id| PointId::from(id as u64)),
113            )],
114        )?;
115        let delete_us = micros(delete_started);
116        mutations.insert(
117            fraction.to_string(),
118            json!({
119                "points": count,
120                "upsert_us": upsert_us,
121                "delete_us": delete_us,
122                "upsert_root": upserted.root(),
123                "delete_root": deleted.root(),
124            }),
125        );
126    }
127
128    let adapter_setup_started = Instant::now();
129    let adapter_dir = tempfile::TempDir::new()?;
130    let database = Database::init_bare(adapter_dir.path())?;
131    let collection = database.create_collection("benchmark", config)?;
132    let adapter_setup_us = micros(adapter_setup_started);
133    let adapter_build_started = Instant::now();
134    let adapter_root = collection.upsert(points)?.root;
135    let adapter_build_us = micros(adapter_build_started);
136    if adapter_root != snapshot.root() {
137        return Err("snapshot-core and named-adapter roots differ".into());
138    }
139    query_collection_all(&collection, &queries[..1], maximum_k, true, None)?;
140    query_collection_all(&collection, &queries[..1], maximum_k, false, None)?;
141    let (adapter_exact_query_us, adapter_exact_results, adapter_exact_vectors_scored) =
142        query_collection_all(&collection, &queries, maximum_k, true, None)?;
143    let (
144        adapter_approximate_query_us,
145        adapter_approximate_results,
146        adapter_approximate_vectors_scored,
147    ) = query_collection_all(&collection, &queries, maximum_k, false, None)?;
148    let adapter_throughput = json!({
149        "exact": query_collection_throughput(
150            &collection,
151            &queries,
152            maximum_k,
153            true,
154            &spec.concurrency,
155        )?,
156        "approximate": query_collection_throughput(
157            &collection,
158            &queries,
159            maximum_k,
160            false,
161            &spec.concurrency,
162        )?,
163    });
164    let historical_started = Instant::now();
165    let historical = collection.at(&adapter_root)?;
166    let historical_count = historical.count(None)?.count;
167    let historical_read_us = micros(historical_started);
168
169    let report = json!({
170        "schema_version": 1,
171        "engine": "git-vdb",
172        "case_name": spec.case_name,
173        "point_count": spec.point_count,
174        "dimension": spec.dimension,
175        "query_count": spec.query_count,
176        "k": spec.k,
177        "root": snapshot.root(),
178        "setup_us": setup_us,
179        "snapshot_core": {
180            "build_us": build_us,
181            "exact_query_us": exact_query_us,
182            "approximate_query_us": approximate_query_us,
183            "exact_results": exact_results,
184            "approximate_results": approximate_results,
185            "exact_vectors_scored": exact_vectors_scored,
186            "approximate_vectors_scored": approximate_vectors_scored,
187            "throughput": snapshot_throughput,
188            "filtered": filtered,
189            "mutations": mutations,
190            "on_disk_bytes": baseline_on_disk_bytes,
191        },
192        "named_adapter": {
193            "setup_us": adapter_setup_us,
194            "build_us": adapter_build_us,
195            "exact_query_us": adapter_exact_query_us,
196            "approximate_query_us": adapter_approximate_query_us,
197            "exact_results": adapter_exact_results,
198            "approximate_results": adapter_approximate_results,
199            "exact_vectors_scored": adapter_exact_vectors_scored,
200            "approximate_vectors_scored": adapter_approximate_vectors_scored,
201            "throughput": adapter_throughput,
202            "historical_read_us": historical_read_us,
203            "historical_count": historical_count,
204            "on_disk_bytes": directory_bytes(adapter_dir.path())?,
205        }
206    });
207    fs::write(output, serde_json::to_vec_pretty(&report)?)?;
208    Ok(())
209}
Source

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 254)
232fn query(
233    input: &Path,
234    repository: &Path,
235    build_report: &Path,
236    mode: &str,
237    output: &Path,
238) -> Result<(), Box<dyn std::error::Error>> {
239    let (exact, warm_exact) = match mode {
240        "exact" => (true, false),
241        "approximate" => (false, false),
242        "approximate-after-exact" => (false, true),
243        _ => return Err(format!("unsupported query mode: {mode}").into()),
244    };
245    let spec = read_spec(input)?;
246    let queries = read_vectors(&spec.queries_path, spec.query_count, spec.dimension)?;
247    let maximum_k = *spec.k.iter().max().ok_or("k must not be empty")?;
248    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
249    let root = build
250        .get("root")
251        .and_then(serde_json::Value::as_str)
252        .ok_or("build report root is missing")?;
253    let engine = SnapshotEngine::open(repository)?;
254    let snapshot = engine.open_snapshot(root)?;
255
256    let cache_build_us = if warm_exact {
257        let started = Instant::now();
258        snapshot.query(make_query(&queries[0], maximum_k, true))?;
259        Some(micros(started))
260    } else {
261        None
262    };
263    // Fill the immutable snapshot cache, construct an approximate lookup over an
264    // already-warm exact view, or warm the unchanged approximate ODB path without
265    // including that one-time work in the samples.
266    let warmup_started = Instant::now();
267    snapshot.query(make_query(&queries[0], maximum_k, exact))?;
268    let warmup_us = micros(warmup_started);
269    wait_for_profiler()?;
270
271    let mut query_us = Vec::with_capacity(queries.len());
272    let mut results = Vec::with_capacity(queries.len());
273    let mut vectors_scored = Vec::with_capacity(queries.len());
274    let batch_started = Instant::now();
275    for vector in &queries {
276        let started = Instant::now();
277        let result = snapshot.query(make_query(vector, maximum_k, exact))?;
278        query_us.push(micros(started));
279        vectors_scored.push(result.stats.vectors_scored);
280        results.push(result.points);
281    }
282    let batch_us = micros(batch_started);
283    fs::write(
284        output,
285        serde_json::to_vec_pretty(&json!({
286            "schema_version": 1,
287            "root": root,
288            "mode": mode,
289            "cache_build_us": cache_build_us,
290            "warmup_us": warmup_us,
291            "query_us": query_us,
292            "batch_us": batch_us,
293            "vectors_scored": vectors_scored,
294            "results": results,
295        }))?,
296    )?;
297    Ok(())
298}
Source

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

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

Source

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

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

Source

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

Retrieves records from an exact root ID.

Source

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

Counts records in an exact root ID.

Source

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 192)
180fn validate(
181    repository: &Path,
182    build_report: &Path,
183    output: &Path,
184) -> Result<(), Box<dyn std::error::Error>> {
185    let build: serde_json::Value = serde_json::from_slice(&fs::read(build_report)?)?;
186    let root = build
187        .get("root")
188        .and_then(serde_json::Value::as_str)
189        .ok_or("build report root is missing")?;
190    let engine = SnapshotEngine::open(repository)?;
191    let started = Instant::now();
192    let report = engine.validate(root, true)?;
193    fs::write(
194        output,
195        serde_json::to_vec_pretty(&json!({
196            "schema_version": 1,
197            "root": root,
198            "validation_us": micros(started),
199            "report": report,
200        }))?,
201    )?;
202    Ok(())
203}
Source

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§

Source§

impl Clone for SnapshotEngine

Source§

fn clone(&self) -> Self

Returns a duplicate of the value. Read more
1.0.0 (const: unstable) · Source§

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

Performs copy-assignment from source. Read more
Source§

impl Debug for SnapshotEngine

Source§

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<ST, DT> CastableFrom<ST, Initialized, Initialized> for DT
where ST: ?Sized, DT: ?Sized,

Source§

impl<ST, DT> CastableFrom<ST, Uninit, Uninit> for DT
where ST: ?Sized, DT: ?Sized,

Source§

impl<T> CloneToUninit for T
where T: Clone,

Source§

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dest. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> IntoEither for T

Source§

fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ

Converts self into a Left variant of Either<Self, Self> if into_left is true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more
Source§

fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
where F: FnOnce(&Self) -> bool,

Converts self into a Left variant of Either<Self, Self> if into_left(&self) returns true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more
Source§

impl<T> Pointable for T

Source§

const ALIGN: usize

The alignment of pointer.
Source§

type Init = T

The type for initializers.
Source§

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more
Source§

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more
Source§

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more
Source§

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more
Source§

impl<T> Read<Exclusive, BecauseExclusive> for T
where T: ?Sized,

Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T> ToOwned for T
where T: Clone,

Source§

type Owned = T

The resulting type after obtaining ownership.
Source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
Source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = !

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, !>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source§

fn vzip(self) -> V