Export and load graph snapshots through any file destination

3rdparty/notices/inventory.json

@@ -515,6 +515,7 @@
         "rust:cpp-demangle@0.4.5",
         "rust:cpufeatures@0.2.17",
         "rust:cpufeatures@0.3.0",
+        "rust:crc32fast@1.5.0",
         "rust:crossbeam-deque@0.8.6",
         "rust:crossbeam-epoch@0.9.21",
         "rust:crossbeam-utils@0.8.21",
@@ -3642,7 +3643,8 @@
     },
     {
       "artifact_ids": [
-        "diskgraph-extension"
+        "diskgraph-extension",
+        "diskgraph-ingest"
       ],
       "copyright": [
         "Copyright (c) 2018 Sam Rijs, Alex Crichton and contributors"

packages/diskgraph/diskgraph-ingest/Cargo.toml

@@ -55,6 +55,7 @@ epserde = { workspace = true }
 memmap2 = "0.9"
 libc = "0.2"
 rayon = "1"
+crc32fast = "1.5"
 jemallocator = { version = "0.5", optional = true, features = ["profiling"] }
 jemalloc-ctl = { version = "0.5", optional = true }
 sysinfo = "0.30"

packages/diskgraph/diskgraph-ingest/src/snapshot_mirror.rs

@@ -36,6 +36,7 @@ use anyhow::{Context, Result};
 use object_store::path::Path as ObjectPath;
 use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
 
 use diskgraph_fs::{CloudTransfer, CloudTransferTicket};
 
@@ -488,21 +489,9 @@ pub(crate) fn validate_manifest_rel_path(rel: &str) -> Result<(), &'static str>
     Ok(())
 }
 
-/// Verify the snapshot rooted at `output_dir` against its `_MANIFEST.json`.
-/// Used by the consumer side after a Python-driven dataplane→local copy
-/// completes, to catch lifecycle-mid-delete races and partial copies
-/// before the snapshot is handed to a workload.
-///
-/// Pass on success; [`CorruptionError`] on any deviation from the
-/// manifest contract. Errors are precise about what failed so the
-/// caller's alerting carries actionable detail (which file is missing,
-/// expected vs actual size, etc.).
-pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::Result<(), CorruptionError> {
-    let success_path = output_dir.join(SUCCESS_MARKER_FILENAME);
-    if !success_path.exists() {
-        return Err(CorruptionError::MissingSuccess(success_path));
-    }
-
+/// Read and validate `_MANIFEST.json` at `output_dir`: schema version and
+/// every rel_path's confinement to the snapshot root.
+fn read_manifest(output_dir: &std::path::Path) -> std::result::Result<Manifest, CorruptionError> {
     let manifest_path = output_dir.join(MANIFEST_FILENAME);
     let manifest_bytes = match std::fs::read(&manifest_path) {
         Ok(b) => b,
@@ -548,6 +537,25 @@ pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::R
             });
         }
     }
+    Ok(manifest)
+}
+
+/// Verify the snapshot rooted at `output_dir` against its `_MANIFEST.json`.
+/// Used by the consumer side after a Python-driven dataplane→local copy
+/// completes, to catch lifecycle-mid-delete races and partial copies
+/// before the snapshot is handed to a workload.
+///
+/// Pass on success; [`CorruptionError`] on any deviation from the
+/// manifest contract. Errors are precise about what failed so the
+/// caller's alerting carries actionable detail (which file is missing,
+/// expected vs actual size, etc.).
+pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::Result<(), CorruptionError> {
+    let success_path = output_dir.join(SUCCESS_MARKER_FILENAME);
+    if !success_path.exists() {
+        return Err(CorruptionError::MissingSuccess(success_path));
+    }
+
+    let manifest = read_manifest(output_dir)?;
 
     let mut missing: Vec<String> = Vec::new();
     let mut size_mismatches: Vec<(String, u64, u64)> = Vec::new();
@@ -584,6 +592,71 @@ pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::R
     Ok(())
 }
 
+/// CRC-32 of every `range_bytes` span of each file `_MANIFEST.json` lists,
+/// and of `_MANIFEST.json` itself, keyed by rel_path (an empty file has one
+/// zero entry). Lets a copy of the snapshot be checked for content as well
+/// as size; the manifest format is unchanged.
+pub fn snapshot_range_checksums(
+    output_dir: &std::path::Path,
+    range_bytes: u64,
+) -> std::result::Result<BTreeMap<String, Vec<u32>>, CorruptionError> {
+    use rayon::prelude::*;
+    use std::os::unix::fs::FileExt;
+
+    const CHUNK: u64 = 1 << 20;
+    if range_bytes == 0 {
+        return Err(CorruptionError::Io {
+            path: output_dir.to_path_buf(),
+            detail: "range_bytes must be positive".into(),
+        });
+    }
+    let mut files = read_manifest(output_dir)?.files;
+    let manifest_size = std::fs::metadata(output_dir.join(MANIFEST_FILENAME))
+        .map(|meta| meta.len())
+        .unwrap_or(0);
+    files.push(ManifestEntry {
+        rel_path: MANIFEST_FILENAME.into(),
+        size_bytes: manifest_size,
+    });
+    let ranges: Vec<(usize, u64)> = files
+        .iter()
+        .enumerate()
+        .flat_map(|(index, entry)| {
+            let count = entry.size_bytes.div_ceil(range_bytes).max(1);
+            (0..count).map(move |range| (index, range))
+        })
+        .collect();
+    let crcs = ranges
+        .par_iter()
+        .map(|&(index, range)| {
+            let entry = &files[index];
+            let path = output_dir.join(&entry.rel_path);
+            let io = |e: std::io::Error| CorruptionError::Io {
+                path: path.clone(),
+                detail: e.to_string(),
+            };
+            let file = std::fs::File::open(&path).map_err(io)?;
+            let start = range * range_bytes;
+            let end = entry.size_bytes.min(start + range_bytes);
+            let mut hasher = crc32fast::Hasher::new();
+            let mut buffer = vec![0u8; CHUNK.min(end - start) as usize];
+            let mut offset = start;
+            while offset < end {
+                let length = CHUNK.min(end - offset) as usize;
+                file.read_exact_at(&mut buffer[..length], offset).map_err(io)?;
+                hasher.update(&buffer[..length]);
+                offset += length as u64;
+            }
+            Ok(hasher.finalize())
+        })
+        .collect::<std::result::Result<Vec<u32>, CorruptionError>>()?;
+    let mut out: BTreeMap<String, Vec<u32>> = BTreeMap::new();
+    for (&(index, _), crc) in ranges.iter().zip(crcs) {
+        out.entry(files[index].rel_path.clone()).or_default().push(crc);
+    }
+    Ok(out)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -933,4 +1006,35 @@ mod tests {
         verify_snapshot_integrity(local_root.path())?;
         Ok(())
     }
+
+    #[test]
+    fn range_checksums_cover_every_listed_file_and_the_manifest() {
+        let tmp = tempfile::tempdir().unwrap();
+        let big: Vec<u8> = (0..10u32).flat_map(|i| i.to_le_bytes()).collect();
+        write_marker_files(tmp.path(), &[("a.bin", &big), ("nested/empty", b"")]);
+        write_manifest(
+            tmp.path(),
+            vec![
+                ManifestEntry {
+                    rel_path: "a.bin".into(),
+                    size_bytes: 40,
+                },
+                ManifestEntry {
+                    rel_path: "nested/empty".into(),
+                    size_bytes: 0,
+                },
+            ],
+        );
+        let crcs = snapshot_range_checksums(tmp.path(), 16).unwrap();
+        let manifest = std::fs::read(tmp.path().join(MANIFEST_FILENAME)).unwrap();
+        assert_eq!(crcs["a.bin"], [&big[..16], &big[16..32], &big[32..]].map(crc32fast::hash));
+        assert_eq!(crcs["nested/empty"], vec![0]);
+        assert_eq!(crcs[MANIFEST_FILENAME], manifest.chunks(16).map(crc32fast::hash).collect::<Vec<_>>());
+
+        std::fs::write(tmp.path().join("a.bin"), &big[..20]).unwrap();
+        assert!(matches!(
+            snapshot_range_checksums(tmp.path(), 16).unwrap_err(),
+            CorruptionError::Io { .. }
+        ));
+    }
 }

packages/diskgraph/diskgraph-pybind/python/diskgraph/__init__.py

@@ -40,6 +40,7 @@
     write_success_marker,
     touch_success_marker,
     verify_snapshot_integrity,
+    snapshot_range_checksums,
     has_resumable_ingest,
     validated_index_resume_progress,
     compute_col_stats_for_parquet,
@@ -141,6 +142,7 @@ def __getattr__(name):
     "write_success_marker",
     "touch_success_marker",
     "verify_snapshot_integrity",
+    "snapshot_range_checksums",
     "has_resumable_ingest",
     "validated_index_resume_progress",
     "compute_col_stats_for_parquet",

packages/diskgraph/diskgraph-pybind/python/tests/test_exception_exports.py

@@ -16,3 +16,22 @@ def test_public_corruption_error_catches_snapshot_verifier(tmp_path):
 def test_build_identity_is_exported():
     assert diskgraph.__version__
     assert diskgraph.BUILD_COMMIT
+
+
+def test_snapshot_range_checksums_match_zlib(tmp_path):
+    import json
+    import zlib
+
+    payload = bytes(range(256)) * 5
+    (tmp_path / "a.bin").write_bytes(payload)
+    (tmp_path / "_MANIFEST.json").write_text(json.dumps({
+        "schema_version": 1,
+        "files": [{"rel_path": "a.bin", "size_bytes": len(payload)}],
+    }))
+    crcs = diskgraph.snapshot_range_checksums(str(tmp_path), 512)
+    assert crcs["a.bin"] == [zlib.crc32(payload[i:i + 512])
+                             for i in range(0, len(payload), 512)]
+    assert crcs["_MANIFEST.json"] == [
+        zlib.crc32((tmp_path / "_MANIFEST.json").read_bytes())]
+    with pytest.raises(diskgraph.DiskgraphCorruptionError):
+        diskgraph.snapshot_range_checksums(str(tmp_path / "missing"), 512)

packages/diskgraph/diskgraph-pybind/src/lib.rs

@@ -1553,6 +1553,21 @@ fn verify_snapshot_integrity(py: Python<'_>, output_dir: &str) -> PyResult<()> {
         .map_err(|e| DiskgraphCorruptionError::new_err(e.to_string()))
 }
 
+/// CRC-32 of every `range_bytes` span of each file in the snapshot's
+/// `_MANIFEST.json` and of the manifest itself, keyed by relative path.
+#[pyfunction]
+fn snapshot_range_checksums(
+    py: Python<'_>,
+    output_dir: &str,
+    range_bytes: u64,
+) -> PyResult<std::collections::BTreeMap<String, Vec<u32>>> {
+    let output_dir = std::path::PathBuf::from(output_dir);
+    py.detach(|| {
+        diskgraph_ingest::snapshot_mirror::snapshot_range_checksums(&output_dir, range_bytes)
+    })
+    .map_err(|e| DiskgraphCorruptionError::new_err(e.to_string()))
+}
+
 #[pyfunction]
 fn has_resumable_ingest(output_dir: &str) -> PyResult<bool> {
     diskgraph_ingest::initial_resume_phase(std::path::Path::new(output_dir))
@@ -1608,6 +1623,7 @@ fn diskgraph_pybind(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(success_marker::write_success_marker, m)?)?;
     m.add_function(wrap_pyfunction!(success_marker::touch_success_marker, m)?)?;
     m.add_function(wrap_pyfunction!(verify_snapshot_integrity, m)?)?;
+    m.add_function(wrap_pyfunction!(snapshot_range_checksums, m)?)?;
     m.add_function(wrap_pyfunction!(has_resumable_ingest, m)?)?;
     m.add_function(wrap_pyfunction!(validated_index_resume_progress, m)?)?;
     m.add_function(wrap_pyfunction!(compute_col_stats_for_parquet, m)?)?;

packages/diskgraph/third_party/artifact-manifest.json

@@ -3241,6 +3241,13 @@
           "source_kind": "registry-or-git",
           "version": "0.3.0"
         },
+        {
+          "license": "MIT OR Apache-2.0",
+          "name": "crc32fast",
+          "source": "registry+https://github.com/rust-lang/crates.io-index",
+          "source_kind": "registry-or-git",
+          "version": "1.5.0"
+        },
         {
           "license": "MIT OR Apache-2.0",
           "name": "crossbeam-deque",
@@ -5202,6 +5209,6 @@
       ]
     }
   ],
-  "cargo_lock_sha256": "297851eb08303a13a97f091c68cf3ce5ccac633ad3d7fe09724c1506e486cdaf",
+  "cargo_lock_sha256": "a1c319a5eb83c1a60f056c213932c12e385f0cff468da4db0021947f11b7aa07",
   "schema": 1
 }

packages/kumo-core/src/kumo_core/sources/object_store.py

@@ -491,7 +491,7 @@ def _copy(self, path: str, found: Sequence[Any], dest: str,
         if self.fetch is not None:
             return self._download_objects(path, found, dest, worker_ceiling,
                                           parquet=True)
-        with _copies(found, worker_ceiling) as workers:
+        with copies(found, worker_ceiling) as workers:
             return fetch_all([
                 StagedFile(
                     entry.path, entry.size
@@ -521,7 +521,7 @@ def download(entry: Any) -> str:
             return target
 
         try:
-            with _copies(found, worker_ceiling) as workers, ThreadPoolExecutor(
+            with copies(found, worker_ceiling) as workers, ThreadPoolExecutor(
                     max_workers=workers) as pool:
                 futures = [pool.submit(download, entry) for entry in found]
                 try:
@@ -816,8 +816,8 @@ def _parquet_batches(fragment: Any,
 
 
 @contextmanager
-def _copies(found: Sequence[Any],
-            worker_ceiling: int | None) -> Iterator[int]:
+def copies(found: Sequence[Any],
+           worker_ceiling: int | None = None) -> Iterator[int]:
     """Arrow-backed downloaders copy on Arrow's IO pool."""
     with memory_share(
             COPY_WORKER_BYTES, lambda ceiling: min(

packages/kumo-engine/docs/mlops.md

@@ -46,3 +46,24 @@ Every fitted bundle carries `_provenance.json` (`ModelManifest`):
 An embedded model keeps only the ids of its own model parents, so a long chain of warm-start retrains does not grow each manifest. `fit` returns a new `TrainingResult` per call (`.id`, `.manifest`); `Trainer.result` and `Trainer.manifest` describe the latest fit or the loaded bundle.
 
 With MLOps on, each bundle is stored under `<root>/models/<bundle id>/`. A complete copy that is already there is kept; a partial one is replaced.
+
+### Moving snapshots between machines
+
+To build an index on one machine and train or score on another, export it to a file destination and load it there:
+
+```python
+from kumo.engine import GraphSnapshot, S3Connector, SnapshotOutputConfig
+
+location = graph.snapshot().export(SnapshotOutputConfig(S3Connector(), "s3://bucket/snapshots"))
+# On the training machine:
+snapshot = GraphSnapshot.load_from_connector(S3Connector(), path=location)
+```
+
+- **Layout:** `export` writes `<path>/<graph id>/<data version>/`, the two halves of `snapshot.id`, and returns that location.
+- **Transfer:** files move in parallel with a CRC32 per range. An interrupted export resumes, and a load either completes or leaves nothing.
+- **Reuse:** loading again reuses the local copy, unless that copy fails DiskGraph's integrity check; then it is downloaded again.
+- **Manifest:** the snapshot's `_provenance.json` travels with the export.
+- **Trust:** load only from paths whose writers you trust. Checksums detect corruption, not tampering.
+- **Destinations:** any destination whose sink holds files at the path works: S3, GCS, local files, or a volume or stage once those sinks are installed. A warehouse connector raises.
+
+With MLOps on (`graph.snapshot(mlops=...)` or a `kumo.mlops` block), each snapshot is exported under `<root>/snapshots/`, unless a complete export of it is already there.

packages/kumo-engine/kumo/engine/__init__.py

@@ -57,6 +57,7 @@
     ModelOutputConfig,
     PredictionOutputConfig,
     PredictionResult,
+    SnapshotOutputConfig,
     Trainer,
     TrainingResult,
     TrainingTableOutputConfig,
@@ -103,6 +104,7 @@
     "ModelOutputConfig",
     "PredictionOutputConfig",
     "TrainingTableOutputConfig",
+    "SnapshotOutputConfig",
 ]
 
 ensure_nofile()