kglite 0.17.8

Pure-Rust embedded Cypher knowledge graph engine with in-memory, mmap, and disk storage, and agent-facing schema introspection
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

//! A compute step publishes its complete CSV only after evaluation succeeds.
use std::path::{Path, PathBuf};

pub(super) struct StagedCsv {
    writer: csv::Writer<tempfile::NamedTempFile>,
    destination: PathBuf,
    operation: &'static str,
}

impl StagedCsv {
    pub(super) fn new(destination: &Path, operation: &'static str) -> Result<Self, String> {
        let parent = destination.parent().ok_or_else(|| {
            format!(
                "{operation}: output {} has no parent",
                destination.display()
            )
        })?;
        std::fs::create_dir_all(parent)
            .map_err(|e| format!("{operation}: create {}: {e}", parent.display()))?;
        let file = tempfile::NamedTempFile::new_in(parent)
            .map_err(|e| format!("{operation}: stage {}: {e}", destination.display()))?;
        Ok(Self {
            writer: csv::WriterBuilder::new()
                .quote_style(csv::QuoteStyle::Necessary)
                .from_writer(file),
            destination: destination.to_path_buf(),
            operation,
        })
    }

    pub(super) fn writer(&mut self) -> &mut csv::Writer<tempfile::NamedTempFile> {
        &mut self.writer
    }

    /// Call after closing the input reader (required for Windows replacement)
    /// and before the infallible blueprint update. This is per-step publication,
    /// not crash durability or whole-pipeline rollback. Drop removes an abandoned stage.
    pub(super) fn publish(self) -> Result<(), String> {
        let Self {
            writer,
            destination,
            operation,
        } = self;
        let file = writer
            .into_inner()
            .map_err(|e| format!("{operation}: flush {}: {e}", destination.display()))?;
        file.persist(&destination)
            .map_err(|e| format!("{operation}: publish {}: {e}", destination.display()))?;
        Ok(())
    }
}