rete-cli 0.4.2

//! Build inputs: gzip, Turtle/TriG streaming, and the external (memory-bounded)
//! build's acceptance of them.
//!
//! The public RDF dumps that actually need `--memory-budget-mb` do not ship as
//! plain `.nt` — they ship as `dump.ttl.gz` / `dump.trig.gz`. Expanding one to
//! N-Triples first costs an order of magnitude more disk than the source (146.8
//! bytes/triple measured on SemOpenAlex), which is exactly the resource the
//! external build exists to avoid spending. These tests pin the contract that
//! makes that unnecessary: every accepted syntax streams, compressed or not, and
//! the bytes it produces do not depend on which spelling of the same graph the
//! input used.

mod common;

use std::io::Write as _;

use common::{fixture, rete};
use predicates::prelude::PredicateBooleanExt as _;

const TTL: &str = concat!(
    "@prefix ex: <http://example.test/> .\n",
    "ex:alice ex:knows ex:bob ;\n",
    "         ex:name \"Alice\"@en .\n",
    "ex:bob ex:name \"Bob\"@en .\n",
);

/// The same three statements as `TTL`, but inside a named graph.
const TRIG: &str = concat!(
    "@prefix ex: <http://example.test/> .\n",
    "ex:people {\n",
    "  ex:alice ex:knows ex:bob ;\n",
    "           ex:name \"Alice\"@en .\n",
    "  ex:bob ex:name \"Bob\"@en .\n",
    "}\n",
);

fn gzip(bytes: &[u8]) -> Vec<u8> {
    let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
    enc.write_all(bytes).unwrap();
    enc.finish().unwrap()
}

/// `.ttl.gz` builds, and to the same bytes as the uncompressed `.ttl`.
/// Compression is a transport detail; it must not reach the output.
#[test]
fn gzipped_turtle_builds_identically() {
    let f = fixture();
    let plain = f.write("g.ttl", TTL);
    let gz = f.write("g.ttl.gz", gzip(TTL.as_bytes()));
    let out_plain = f.path("plain.rete");
    let out_gz = f.path("gz.rete");

    for (src, out) in [(&plain, &out_plain), (&gz, &out_gz)] {
        rete()
            .args(["build"])
            .arg(src)
            .arg("-o")
            .arg(out)
            .assert()
            .success();
    }
    assert_eq!(
        std::fs::read(&out_plain).unwrap(),
        std::fs::read(&out_gz).unwrap(),
        "gzip must not change the built file"
    );
}

/// A gzip file may be a CONCATENATION of members — what `cat a.gz b.gz` writes,
/// and how several public dumps are assembled. `GzDecoder` stops silently after
/// the first member and would drop the rest of the graph on the floor; the
/// decoder must be the multi-member one.
#[test]
fn concatenated_gzip_members_all_read() {
    let f = fixture();
    let first = "<http://example.test/a> <http://example.test/p> \"1\" .\n";
    let second = "<http://example.test/b> <http://example.test/p> \"2\" .\n";
    let mut both = gzip(first.as_bytes());
    both.extend_from_slice(&gzip(second.as_bytes()));
    let src = f.write("both.nt.gz", both);
    let out = f.path("both.rete");

    rete()
        .args(["build"])
        .arg(&src)
        .arg("-o")
        .arg(&out)
        .assert()
        .success()
        .stdout(predicates::str::contains("2 triples"));
}

/// The external build accepts Turtle — the point of the exercise. Its output is
/// documented to be byte-identical to a standard `--no-pyramid` build, and that
/// promise must survive the new input path.
#[test]
fn external_build_from_gzipped_turtle_matches_in_ram() {
    let f = fixture();
    let gz = f.write("g.ttl.gz", gzip(TTL.as_bytes()));
    let plain = f.write("g.ttl", TTL);
    let out_ext = f.path("ext.rete");
    let out_ram = f.path("ram.rete");

    rete()
        .args(["build"])
        .arg(&gz)
        .args(["--memory-budget-mb", "64", "-o"])
        .arg(&out_ext)
        .assert()
        .success();
    rete()
        .args(["build"])
        .arg(&plain)
        .args(["--no-pyramid", "-o"])
        .arg(&out_ram)
        .assert()
        .success();

    assert_eq!(
        std::fs::read(&out_ext).unwrap(),
        std::fs::read(&out_ram).unwrap(),
        "external build of .ttl.gz must equal the in-RAM --no-pyramid build"
    );
}

/// TriG parses, and its named graph survives into the file.
#[test]
fn trig_keeps_its_named_graph() {
    let f = fixture();
    let src = f.write("g.trig", TRIG);
    let out = f.path("g.rete");

    rete()
        .args(["build"])
        .arg(&src)
        .arg("-o")
        .arg(&out)
        .assert()
        .success()
        .stdout(predicates::str::contains("named graph"));
    rete()
        .args(["graphs"])
        .arg(&out)
        .assert()
        .success()
        .stdout(predicates::str::contains("http://example.test/people"));
}

/// `--collapse-graphs` folds those statements into the default graph — which is
/// what makes a named-graph dump answer `?s ?p ?o` at all. The result must equal
/// the same data written as plain Turtle.
#[test]
fn collapse_graphs_matches_the_default_graph_build() {
    let f = fixture();
    let trig = f.write("g.trig", TRIG);
    let ttl = f.write("g.ttl", TTL);
    let out_collapsed = f.path("collapsed.rete");
    let out_ttl = f.path("ttl.rete");

    rete()
        .args(["build"])
        .arg(&trig)
        .args(["--collapse-graphs", "-o"])
        .arg(&out_collapsed)
        .assert()
        .success();
    rete()
        .args(["build"])
        .arg(&ttl)
        .arg("-o")
        .arg(&out_ttl)
        .assert()
        .success();

    assert_eq!(
        std::fs::read(&out_collapsed).unwrap(),
        std::fs::read(&out_ttl).unwrap(),
        "a collapsed TriG must build to the same file as the equivalent Turtle"
    );
}

/// The external build carries named graphs through the same spill the default
/// graph uses, so a TriG input must produce **byte-identical** output to the
/// in-RAM `--no-pyramid` build — the guarantee `--memory-budget-mb` already made
/// for the default graph, extended to quads (#139).
#[test]
fn external_build_keeps_named_graphs_byte_identically() {
    let f = fixture();
    let src = f.write("g.trig", TRIG);
    let out_ext = f.path("ext.rete");
    let out_ram = f.path("ram.rete");

    rete()
        .args(["build"])
        .arg(&src)
        .args(["--memory-budget-mb", "64", "-o"])
        .arg(&out_ext)
        .assert()
        .success()
        .stdout(predicates::str::contains("named graph"));
    rete()
        .args(["build"])
        .arg(&src)
        .args(["--no-pyramid", "-o"])
        .arg(&out_ram)
        .assert()
        .success();

    assert_eq!(
        std::fs::read(&out_ext).unwrap(),
        std::fs::read(&out_ram).unwrap(),
        "external build of a TriG must equal the in-RAM --no-pyramid build"
    );
    rete()
        .args(["graphs"])
        .arg(&out_ext)
        .assert()
        .success()
        .stdout(predicates::str::contains("http://example.test/people"));

    // …and --collapse-graphs still folds them into the default graph.
    let out_flat = f.path("flat.rete");
    rete()
        .args(["build"])
        .arg(&src)
        .args(["--collapse-graphs", "--memory-budget-mb", "64", "-o"])
        .arg(&out_flat)
        .assert()
        .success();
}

/// `rete merge` fed the external builder, so it refused named-graph shards up
/// front. With #139 it folds them: two shards carrying *different* named graphs
/// merge into one file that has both, and the union of their statements.
#[test]
fn merge_folds_named_graph_shards() {
    let f = fixture();
    let a = f.write(
        "a.nq",
        concat!(
            "<http://ex/s1> <http://ex/p> <http://ex/o1> <http://ex/gA> .\n",
            "<http://ex/d> <http://ex/p> <http://ex/o> .\n",
        ),
    );
    let b = f.write(
        "b.nq",
        concat!(
            "<http://ex/s2> <http://ex/p> <http://ex/o2> <http://ex/gB> .\n",
            "<http://ex/s1> <http://ex/p> <http://ex/o1> <http://ex/gA> .\n",
        ),
    );
    let (ra, rb, out) = (f.path("a.rete"), f.path("b.rete"), f.path("all.rete"));
    for (src, dst) in [(&a, &ra), (&b, &rb)] {
        rete()
            .args(["build"])
            .arg(src)
            .arg("-o")
            .arg(dst)
            .assert()
            .success();
    }
    rete()
        .args(["merge"])
        .arg(&ra)
        .arg(&rb)
        .arg("-o")
        .arg(&out)
        .assert()
        .success();

    let graphs = rete().args(["graphs"]).arg(&out).assert().success();
    let listed = String::from_utf8(graphs.get_output().stdout.clone()).unwrap();
    assert!(listed.contains("http://ex/gA"), "gA missing: {listed}");
    assert!(listed.contains("http://ex/gB"), "gB missing: {listed}");
    // the duplicate quad in gA is written once: 2 named + 1 default
    rete()
        .args(["info"])
        .arg(&out)
        .assert()
        .success()
        .stdout(predicates::str::contains("quad_count: 3"));
}

/// G22: `rete merge` reads its inputs inside its own `--memory-budget-mb`.
/// Up to 0.4.1 a small shard was read whole (its decoded dictionary and every
/// permutation resident) and a large one opened with uncapped caches, none of
/// it charged to the budget: the merge ran about 1 GiB over it and did not fit
/// a 4 GiB container at 3,000 MiB. Now every shard, however small, is opened
/// by range with its caches capped to a quarter of the budget, and the
/// builder's chunk phase takes the rest. The merged file does not change.
#[test]
fn merge_reads_its_inputs_inside_the_budget() {
    let f = fixture();
    let (mut a, mut b) = (String::new(), String::new());
    for i in 0..300 {
        let shard = if i % 2 == 0 { &mut a } else { &mut b };
        shard.push_str(&format!(
            "<http://ex/s{i}> <http://ex/p{}> \"value {i}\" <http://ex/g{}> .\n\
             <http://ex/s{i}> <http://ex/next> <http://ex/s{}> .\n",
            i % 5,
            i % 7,
            (i + 1) % 300
        ));
    }
    let union = f.write("all.nq", format!("{a}{b}"));
    let (a, b) = (f.write("a.nq", &a), f.write("b.nq", &b));
    let (ra, rb) = (f.path("a.rete"), f.path("b.rete"));
    for (src, dst) in [(&a, &ra), (&b, &rb)] {
        rete()
            .args(["build"])
            .arg(src)
            .arg("-o")
            .arg(dst)
            .assert()
            .success();
    }
    let merge = |budget: &str, out: &std::path::Path| {
        let run = rete()
            .args(["merge"])
            .arg(&ra)
            .arg(&rb)
            .arg("-o")
            .arg(out)
            .args(["--memory-budget-mb", budget])
            .env("RETE_OPEN_DEBUG", "1")
            .assert()
            .success();
        String::from_utf8(run.get_output().stderr.clone()).unwrap()
    };
    let (m512, m64) = (f.path("m512.rete"), f.path("m64.rete"));
    let log = merge("512", &m512);
    // Both shards by range, never read whole, though each is a few KB.
    assert!(!log.contains("EAGER"), "{log}");
    assert!(
        log.matches("[open_local_ranged]").count() >= 2,
        "an input was not opened by range: {log}"
    );
    // Their caches inside a quarter of the budget, and the chunks in the rest.
    assert!(log.contains("[memory-budget] total=128MiB"), "{log}");
    assert!(
        log.contains("extbuild: budget 512 MiB, 128 MiB of it for reading the input"),
        "{log}"
    );
    merge("64", &m64);
    assert_eq!(
        std::fs::read(&m512).unwrap(),
        std::fs::read(&m64).unwrap(),
        "the budget changed the merged file"
    );
    // ...which is the memory-bounded build of the union, byte for byte.
    let built = f.path("built.rete");
    rete()
        .args(["build"])
        .arg(&union)
        .args(["--memory-budget-mb", "64", "-o"])
        .arg(&built)
        .assert()
        .success();
    assert_eq!(
        std::fs::read(&m512).unwrap(),
        std::fs::read(&built).unwrap(),
        "the merge is not the build of its inputs' union"
    );
}

/// `--text-index` on the memory-bounded path: it builds, the file is the same
/// bytes as the in-RAM `--no-pyramid --text-index` build (cardless, so the
/// whole file, not just the section), and `rete search --contains` answers —
/// for the default scope and for a scoped one.
#[test]
fn external_build_writes_the_in_ram_text_index() {
    let f = fixture();
    let src = f.write(
        "g.nq",
        "<http://ex/s> <http://ex/p> \"hello world\" .\n\
         <http://ex/s> <http://ex/desc> \"has been modified\" .\n\
         <http://ex/t> <http://ex/p> \"hello études\"@fr .\n\
         <http://ex/s/prov> <http://ex/desc> \"has been created\" <http://ex/s/g> .\n",
    );
    let scoped = [
        "--text-index-graphs",
        "all",
        "--text-index-exclude",
        "http://ex/desc",
    ];
    for extra in [&[][..], &scoped[..]] {
        let ram = f.path("ram.rete");
        let ext = f.path("ext.rete");
        rete()
            .args(["build"])
            .arg(&src)
            .args(["--text-index", "--no-pyramid"])
            .args(extra)
            .arg("-o")
            .arg(&ram)
            .assert()
            .success();
        rete()
            .args(["build"])
            .arg(&src)
            .args(["--text-index", "--memory-budget-mb", "64"])
            .args(extra)
            .arg("-o")
            .arg(&ext)
            .assert()
            .success()
            .stderr(predicates::str::contains("extbuild: text index"));
        assert_eq!(
            std::fs::read(&ext).unwrap(),
            std::fs::read(&ram).unwrap(),
            "{extra:?}: memory-bounded and in-RAM builds differ"
        );
        rete()
            .args(["search"])
            .arg(&ext)
            .args(["--contains", "hello"])
            .assert()
            .success()
            .stdout(predicates::str::contains("<http://ex/s>"))
            .stdout(predicates::str::contains("<http://ex/t>"));
    }
    // The scoped file leaves `ex:desc` out, so `modified` finds nothing.
    rete()
        .args(["search"])
        .arg(f.path("ext.rete"))
        .args(["--contains", "modified"])
        .assert()
        .success()
        .stdout(predicates::str::contains("<http://ex/s>").not());
}

/// The scope flags need `--text-index`, take an absolute IRI, and do not mix
/// an include list with an exclude list.
#[test]
fn text_index_scope_flags_are_validated() {
    let f = fixture();
    let src = f.write("g.nt", "<http://ex/s> <http://ex/p> \"hello world\" .\n");
    let out = f.path("out.rete");
    for (args, why) in [
        (
            &["--text-index-exclude", "http://ex/p"][..],
            "required arguments were not provided",
        ),
        (
            &["--text-index-graphs", "all"][..],
            "required arguments were not provided",
        ),
        (
            &["--text-index", "--text-index-exclude", "description"][..],
            "absolute",
        ),
        (
            &[
                "--text-index",
                "--text-index-include",
                "http://ex/a",
                "--text-index-exclude",
                "http://ex/b",
            ][..],
            "cannot be used with",
        ),
    ] {
        rete()
            .args(["build"])
            .arg(&src)
            .args(args)
            .arg("-o")
            .arg(&out)
            .assert()
            .failure()
            .stderr(predicates::str::contains(why));
    }
}

/// RDF/XML is the one syntax the external build still refuses; the error has to
/// say so rather than fall over inside the parser.
#[test]
fn external_build_rejects_rdfxml_with_a_clear_error() {
    let f = fixture();
    let src = f.write("o.rdf", "<rdf:RDF/>");
    let out = f.path("ext.rete");

    rete()
        .args(["build"])
        .arg(&src)
        .args(["--memory-budget-mb", "64", "-o"])
        .arg(&out)
        .assert()
        .failure()
        .stderr(predicates::str::contains("rdfxml"));
}

/// Turtle with **anonymous blank nodes** — `[ … ]` and collections — must build.
///
/// This is the trap the streaming work has to stay clear of: oxttl labels an
/// anonymous blank node with a fresh random id on every parse, so any path that
/// parses the input more than once presents the encoder terms the observing pass
/// never saw. The two-pass assembler does exactly that, which is why `.ttl` files
/// keep taking the whole-text path; the external build reads once and is safe.
/// Both must produce the graph, and the same number of statements.
#[test]
fn turtle_with_anonymous_blank_nodes_builds_on_both_paths() {
    let f = fixture();
    let ttl = concat!(
        "@prefix ex: <http://ex/> .\n",
        "ex:a ex:p [ ex:q \"1\" ] .\n",
        "ex:b ex:list ( \"x\" \"y\" ) .\n",
    );
    let src = f.write("bn.ttl", ttl);
    let out_plain = f.path("bn.rete");
    let out_ext = f.path("bn-ext.rete");

    rete()
        .args(["build"])
        .arg(&src)
        .args(["--no-pyramid", "-o"])
        .arg(&out_plain)
        .assert()
        .success()
        .stdout(predicates::str::contains("7 triples"));
    rete()
        .args(["build"])
        .arg(&src)
        .args(["--memory-budget-mb", "64", "-o"])
        .arg(&out_ext)
        .assert()
        .success()
        .stdout(predicates::str::contains("7 triples"));
}

/// `estimate` reads the same inputs the build does — otherwise the "will this
/// fit?" question cannot be asked about the dumps that need asking.
#[test]
fn estimate_reads_gzipped_turtle() {
    let f = fixture();
    let gz = f.write("g.ttl.gz", gzip(TTL.as_bytes()));

    rete()
        .args(["estimate"])
        .arg(&gz)
        .assert()
        .success()
        .stdout(predicates::str::contains("statements"));
}

/// `--label-index` and the default SCHEMA section are part of the external
/// build's byte-identity promise: the memory-bounded build computes both with
/// external sorts and must write exactly the in-RAM build's bytes.
#[test]
fn external_build_with_label_index_matches_in_ram() {
    let f = fixture();
    let nt = "<http://ex/w1> <http://purl.org/dc/terms/title> \"Acta  Universitatis\" .\n\
              <http://ex/w1> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://ex/Journal> .\n\
              <http://ex/w1> <http://ex/editor> <http://ex/p1> .\n\
              <http://ex/p1> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://ex/Person> .\n\
              <http://ex/p1> <http://xmlns.com/foaf/0.1/familyName> \"Svobodová\" .\n\
              <http://ex/p1> <http://xmlns.com/foaf/0.1/givenName> \"Jana\" .\n\
              <http://ex/Journal> <http://www.w3.org/2000/01/rdf-schema#subClassOf> <http://ex/Venue> .\n";
    let src = f.write("l.nt", nt);
    let (out_ext, out_ram) = (f.path("ext.rete"), f.path("ram.rete"));
    rete()
        .args(["build"])
        .arg(&src)
        .args(["--memory-budget-mb", "64", "--label-index", "-o"])
        .arg(&out_ext)
        .assert()
        .success();
    rete()
        .args(["build"])
        .arg(&src)
        .args(["--no-pyramid", "--label-index", "-o"])
        .arg(&out_ram)
        .assert()
        .success();
    let ext = std::fs::read(&out_ext).unwrap();
    assert_eq!(ext, std::fs::read(&out_ram).unwrap());
    let h = rete_core::Header::from_bytes(&ext).unwrap();
    assert!(h.label_index_section().is_some() && h.schema_section().is_some());
    rete()
        .args(["search"])
        .arg(&out_ext)
        .arg("svob")
        .assert()
        .success()
        .stdout(predicates::str::contains("Svobodová, Jana"));
}