Graphs

OpenZL compresses by routing typed streams through a graph of codecs. This guide explains how the graphs of this crate work, which standard graphs exist, and how to build your own with mopenzl::graph. Frame-level mechanics live in compression graphs.

Types · Standard graphs · GraphBuilder · Parameters · Selectors, function graphs, segmenters · Chunking · try_graph · Permissive mode · Sessions

Typed inputs and outputs

Every stream has a DataType:

The constructors validate the layout (widths, whole elements, lengths summing to the content size) and return InputError otherwise. Numbers are always little-endian bytes. Input::with_metadata(&[(id, value)]) attaches integer metadata that graphs can read, such as the tag the clustering graph groups by; metadata is never written to the frame.

Compressor::compress_inputs(&[Input]) writes one frame from one or more inputs and Decompressor::decompress_typed returns one TypedOutput per input, in order. Compressor::compress is compress_inputs with a single serial input.

A node is one codec invocation: it takes typed streams and produces typed streams (delta_int: numeric to numeric; float32_deconstruct: numeric to a struct plus a serial; tokenize: any to an alphabet plus numeric indices). A graph is anything that accepts streams and decides what happens to them: a node plus successor graphs, a selector, a function graph, a segmenter, or store. Nodes do no implicit type conversion when you run them yourself; the engine converts implicitly only when a stream reaches store (numeric and struct data are converted to serial bytes first, which emits a conversion codec).

Standard graphs

StandardGraph names the graphs of the reference’s registry with their numeric IDs. CompressorGraph::standard(graph) (or CompressorGraph::from) makes a ready compressor graph, and Compressor::builder(config).with_graph(..) installs it. Without a graph the compressor runs the default: compress_generic.

use mopenzl::graph::{CompressorGraph, StandardGraph};
use mopenzl::input::{Input, Records, Strings};
use mopenzl::{Compressor, DataType, Decompressor, EncoderConfig};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let text = b"alpha beta gamma delta alpha beta gamma delta ".repeat(20);
    let lens: Vec<u32> = vec![46; 20];
    let records: Vec<u8> = (0..4000u32)
        .flat_map(|i| [(i % 7) as u8, 1, 2, 3])
        .collect();
    let inputs = [
        Input::from(&text[..]),
        Input::from(Records::try_from((&records[..], 4))?),
        Input::from(Strings::try_from((&text[..], &lens[..]))?),
    ];

    // `compress_generic` takes any number of inputs and routes each by type.
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(CompressorGraph::standard(StandardGraph::CompressGeneric))
        .build()?;
    let frame = compressor.compress_inputs(&inputs)?;

    let outputs = Decompressor::default().decompress_typed(&frame)?;
    let types: Vec<DataType> = outputs.iter().map(|o| o.data_type()).collect();
    assert_eq!(
        types,
        [DataType::Serial, DataType::Struct, DataType::String]
    );
    assert_eq!(outputs[0].content(), text);
    assert_eq!(outputs[1].width(), 4);
    assert_eq!(outputs[2].string_lens(), lens);
    Ok(())
}

Which graph when

Start with CompressGeneric (the default) and move to a specific graph when you know your data. S, St, N and T below mean serial, struct, numeric and string inputs. Graphs that take “any” accept several inputs.

Public graphs

IDs from 32 on are private in the reference but exist here with stable numbers; they are useful as successors and for reading the registry:

Graphs that need successors

Some standard graphs route to graphs you must supply with GraphBuilder::parameterize_graph(graph, GraphParams::default().with_graphs([..])); run bare they fail (“needs exactly three custom graphs”). Each variant’s rustdoc says what it needs; in short:

FieldLz and Lz need none; their custom graphs are optional overrides picked by index through params::field_lz and params::lz. SegmentNumeric needs none. Every other standard graph is self-contained.

StandardGraph and StandardNode are #[non_exhaustive]; TryFrom<u32> and From<..> for u32 convert to and from the reference’s IDs. Every standard graph needs the codecs it uses: a graph whose codec feature is off fails like a failing codec (see permissive mode and codecs).

Building graphs

GraphBuilder starts with only the standard graphs and nodes and derives new ones. Each call returns a GraphId or NodeId that is meaningful to that builder only; build(start) freezes everything into a cheap-to-clone, thread-shareable CompressorGraph rooted at start.

use mopenzl::graph::{GraphBuilder, StandardGraph, StandardNode};
use mopenzl::input::{Input, Numbers};
use mopenzl::{Compressor, Decompressor, EncoderConfig};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    // delta -> zigzag -> bit-pack: three nodes in a chain, then store.
    let packed = builder.pipeline(
        &[
            StandardNode::DeltaInt.into(),
            StandardNode::Zigzag.into(),
            StandardNode::BitpackInt.into(),
        ],
        StandardGraph::Store,
    )?;
    let graph = builder.build(packed)?;

    let values: Vec<u8> = (0..2000u32)
        .flat_map(|i| (500 + i % 17).to_le_bytes())
        .collect();
    let numbers = Numbers::try_from((&values[..], 4))?;
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress_inputs(&[Input::from(numbers)])?;
    assert!(frame.len() < values.len() / 2);

    let outputs = Decompressor::default().decompress_typed(&frame)?;
    assert_eq!(outputs[0].content(), values);
    Ok(())
}

Builders report mistakes early: a successor count that does not match a node’s outputs, an unknown or foreign ID, an empty pipeline or an unknown start graph is a GraphError (Invalid, UnknownId). Failures while a graph runs surface from the compressor as EncodeError::Graph(GraphError::Execution(..)), or GraphError::User(..) when your own graph returned the error.

Parameterized nodes and graphs

Nodes and some graphs take local parameters: integer, byte-blob or object parameters keyed by the reference’s IDs. LocalParams::default().with_int(id, v) and .with_bytes(id, bytes) build them; giving a node or graph parameters replaces the ones it had. The global level reaches Zstandard unchanged, and the Zstandard node applies its own integer parameters after it, as raw ZSTD_c_* IDs (params::zstd::COMPRESSION_LEVEL is 100), in list order.

The IDs are the reference’s, and mopenzl::graph::params names them: one module per codec or graph (zstd, lz, lz4, field_lz, entropy, divide_by, bitunpack, partition, sparse_num, sentinel, tokenize, split, dispatch, mux_lengths, segmenters, sddl, sddl2, clustering, …). Each constant cites the C macro or enum it mirrors (ZL_ENTROPY_MIN_GAIN_BYTES_PID, ZL_LzParam_windowLog, …) and its doc says whether the parameter is an integer or a copy (bytes) parameter.

use mopenzl::graph::params::zstd;
use mopenzl::graph::{GraphBuilder, GraphParams, LocalParams, StandardGraph, StandardNode};
use mopenzl::{Compressor, Decompressor, EncoderConfig};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    // The Zstandard node with its compression level pinned to 19.
    let zstd19 = builder.parameterize_node(
        StandardNode::Zstd,
        LocalParams::default().with_int(zstd::COMPRESSION_LEVEL, 19),
    )?;
    let strong = builder.static_graph(zstd19, &[StandardGraph::Store.into()])?;
    // A copy of `brute_force`, with its candidates replaced.
    let race = builder.parameterize_graph(
        StandardGraph::BruteForce,
        GraphParams::default()
            .with_graphs([
                strong,
                StandardGraph::Lz4.into(),
                StandardGraph::Store.into(),
            ])
            .with_name("zstd19-or-lz4"),
    )?;
    let graph = builder.build(race)?;

    let input = b"parameters replace, they do not merge ".repeat(500);
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress(&input)?;
    assert!(frame.len() < input.len() / 10);
    assert_eq!(Decompressor::default().decompress(&frame)?, input);
    Ok(())
}

GraphBuilder::brute_force(&[..]) is the shorthand for the parameterize_graph call above. LocalParams::with_string_lengths supplies the lengths of the set_string_lens node. The key local parameters of every codec are in codecs.

Custom graphs in Rust

Three traits mirror the reference’s callbacks. All are Send + Sync + 'static so a built graph is shareable between threads.

Selector

A selector looks at a stream and returns one of its successors (or any other graph of the compressor). SelectorContext::try_graph compresses the stream with a candidate and returns the resulting frame size, so a selector can measure instead of guess.

use mopenzl::graph::{
    GraphBuilder, GraphId, InputView, Selector, SelectorContext, StandardGraph, TypeSet,
};
use mopenzl::{Compressor, DataType, Decompressor, EncoderConfig};

/// Picks the successor that produces the smallest frame.
struct Smallest;

impl Selector for Smallest {
    fn select(
        &self,
        cx: &mut SelectorContext<'_>,
        input: InputView<'_>,
        successors: &[GraphId],
    ) -> GraphId {
        let mut best = successors[0];
        let mut best_size = usize::MAX;
        for &graph in successors {
            let size = cx.try_graph(&input, graph).unwrap_or(usize::MAX);
            if size < best_size {
                (best, best_size) = (graph, size);
            }
        }
        best
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    let selector = builder.selector(
        "smallest",
        Smallest,
        TypeSet::from(DataType::Serial),
        &[
            StandardGraph::Zstd.into(),
            StandardGraph::Lz4.into(),
            StandardGraph::Store.into(),
        ],
    )?;
    let graph = builder.build(selector)?;

    let input = b"selectors measure instead of guessing ".repeat(300);
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress(&input)?;
    assert_eq!(Decompressor::default().decompress(&frame)?, input);
    Ok(())
}

A selector cannot fail: return a graph that always works (such as Store) as the fallback. SelectorContext also exposes level(), format_version(), depth(), is_node_supported(node), local_int(id) and set_successor_params(..).

Function graph

A function graph gets its inputs as edges, runs nodes on them and assigns every resulting edge a destination graph. Nodes run at the moment you ask; successors run after the function returns, in the order of the edge handles (the inputs first, then node outputs in creation order), not in the order you call set_destination. An edge left without a destination is an error (or triggers the permissive fallback). GraphContext::data(edge) reads a stream, custom_graphs() and custom_nodes() return the lists you registered, and run_node_with and set_destination_with override parameters at run time. run_split_node, run_dispatch_node and run_dispatch_string_node run the split and dispatch nodes with their runtime instructions.

use mopenzl::graph::{
    Edge, FunctionGraph, GraphBuilder, GraphContext, GraphError, StandardGraph, StandardNode,
};
use mopenzl::input::{Input, Strings};
use mopenzl::{Compressor, DataType, Decompressor, EncoderConfig};

/// Separates strings into lengths and content, and compresses each its own way.
struct SplitStrings;

impl FunctionGraph for SplitStrings {
    fn compress(&self, cx: &mut GraphContext<'_>, inputs: &[Edge]) -> Result<(), GraphError> {
        let parts = cx.run_node(StandardNode::SeparateStringComponents, inputs)?;
        let graphs = cx.custom_graphs();
        cx.set_destination(&parts[0..1], graphs[0])?; // content
        cx.set_destination(&parts[1..2], graphs[1])?; // lengths
        Ok(())
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    let function = builder.function_graph(
        "split_strings",
        SplitStrings,
        &[DataType::String.into()],
        false,
        &[
            StandardGraph::Zstd.into(),
            StandardGraph::NumericCompress.into(),
        ],
        &[],
    )?;
    let graph = builder.build(function)?;

    let text = b"red green blue red green blue ".repeat(100);
    let lens: Vec<u32> = [3, 5, 4, 3, 5, 4]
        .iter()
        .cycle()
        .take(600)
        .copied()
        .collect();
    let content_len: u32 = lens.iter().sum();
    let strings = Strings::try_from((&text[..content_len as usize], &lens[..]))?;

    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress_inputs(&[Input::from(strings)])?;
    let outputs = Decompressor::default().decompress_typed(&frame)?;
    assert_eq!(outputs[0].string_lens(), lens);
    assert_eq!(outputs[0].content(), &text[..content_len as usize]);
    Ok(())
}

Segmenter

A segmenter cuts the whole input into chunks, each compressed independently through a successor, and writes one frame. See chunking.

use mopenzl::graph::{GraphBuilder, GraphError, Segmenter, SegmenterContext, StandardGraph};
use mopenzl::{Compressor, DataType, Decompressor, EncoderConfig};

/// Cuts the input in thirds.
struct Thirds;

impl Segmenter for Thirds {
    fn segment(&self, cx: &mut SegmenterContext<'_>) -> Result<(), GraphError> {
        let total = cx.remaining()[0];
        let successor = cx.custom_graphs()[0];
        let third = total.div_ceil(3).max(1);
        let mut left = total;
        while left > 0 {
            let take = third.min(left);
            cx.process_chunk(&[take], successor)?;
            left -= take;
        }
        if total == 0 {
            cx.process_chunk(&[0], successor)?;
        }
        Ok(())
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    let thirds = builder.segmenter(
        "thirds",
        Thirds,
        &[DataType::Serial.into()],
        &[StandardGraph::CompressGeneric.into()],
    )?;
    let graph = builder.build(thirds)?;

    let input = b"three chunks, three independent graphs ".repeat(400);
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress(&input)?;
    assert_eq!(Decompressor::default().decompress(&frame)?, input);
    Ok(())
}

Chunking and segmenters

A frame is a sequence of chunks, each with its own codec layout. The default graph writes the whole input as one chunk (Chunking::Whole, which still splits every 2 GiB). A segmenter cuts several. The standard ones:

• EncoderConfig::with_chunking(Chunking::Fixed(ChunkSize)) is the serial segmenter of the reference CLI: chunks of ChunkSize input bytes (32 KiB to i32::MAX, default 16 MiB), the last one shorter. It wraps whatever graph you installed.
• GraphBuilder::serial_segmenter(chunk, successor) is the same as a start graph of your own, with a chosen successor per chunk. A start graph that is already a segmenter cannot be combined with Chunking::Fixed (an EncodeError::Graph).
• GraphBuilder::num_from_serial_segmenter(width, chunk, successor) cuts serial input aligned to width-byte integers and compresses each chunk as numeric data (the default successor, InterpretNum8Compress to InterpretNum64Compress, reads each serial chunk as numbers and compresses them; a successor you pass receives the chunk as serial data, so it must accept serial input and convert it itself, for example a static graph that starts with ConvertSerialToNumLe32).
• SegmentNumeric cuts numeric input every 16 MiB; the SDDL2 graph is a segmenter that cuts along the segments of its program.

Chunks cap memory: with a declared InputSize::Exact both the encoder session and the decoder work one chunk at a time. Chunking needs format version 21 or newer, which every supported version has.

use mopenzl::graph::GraphBuilder;
use mopenzl::{ChunkSize, Compressor, Decompressor, EncoderConfig};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut builder = GraphBuilder::new();
    // 8-byte integers, 64 KiB chunks; each chunk is read as numbers and compressed.
    let segmenter = builder.num_from_serial_segmenter(8, ChunkSize::try_from(64 << 10)?, None)?;
    let graph = builder.build(segmenter)?;

    let input: Vec<u8> = (0..40_000u64)
        .flat_map(|i| (i * 1000).to_le_bytes())
        .collect();
    let mut compressor = Compressor::builder(EncoderConfig::default())
        .with_graph(graph)
        .build()?;
    let frame = compressor.compress(&input)?;
    assert_eq!(Decompressor::default().decompress(&frame)?, input);
    Ok(())
}

Trial compression: try_graph

SelectorContext::try_graph(&input, graph) and GraphContext::try_graph(edge, graph) compress the stream with graph in a throwaway context (checksums off, no comment) and return the size of the complete frame it would produce, or an error when the graph fails. brute_force is built on it. Trials cost a full compression each; use them on the successors that matter. The selector example shows the idiom.

Permissive mode

Normally a failing node, graph or successor fails the compression. With EncoderConfig::with_permissive(true), a failing successor outside a segmenter is rolled back and its input recompressed with compress_generic, as in the reference. This is the net under graphs that cannot always apply: Constant on data that is not constant, an MlSelector whose model does not match its successors, a codec whose Cargo feature is off.

use mopenzl::graph::{CompressorGraph, StandardGraph};
use mopenzl::{Compressor, Decompressor, EncoderConfig};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let not_constant = b"this is certainly not one repeated byte".repeat(10);
    let graph = CompressorGraph::standard(StandardGraph::Constant);

    let mut strict = Compressor::builder(EncoderConfig::default())
        .with_graph(graph.clone())
        .build()?;
    assert!(strict.compress(&not_constant).is_err());

    let config = EncoderConfig::default().with_permissive(true);
    let mut permissive = Compressor::builder(config).with_graph(graph).build()?;
    let frame = permissive.compress(&not_constant)?;
    assert_eq!(Decompressor::default().decompress(&frame)?, not_constant);
    Ok(())
}

Two more nets always apply: a lone stream below min_stream_size (10) is stored, and, with store_on_expansion (the default), a chunk that would not shrink is rebuilt as store.

Sessions and graphs

Compressor::start and the std::io writer work with any graph, with these rules:

• The default graph and a serial segmenter (including Chunking::Fixed) compress incrementally, chunk by chunk, and match the one-shot bytes.
• Every other graph is buffered: the session holds its input until Operation::End, runs the graph once and then drains the output. A declared InputSize::Exact is checked at the end.
• Typed and multiple inputs go through compress_inputs; sessions, readers and writers take serial bytes.

See streaming.