Struct Runner 

pub struct Runner {
    pub mode: ExecMode,
    pub default_partitions: usize,
    pub checkpoint_config: Option<CheckpointConfig>,
}

Executes a pipeline produced by the builder API.

Construct a Runner and call Runner::run_collect with a pipeline and terminal node id. See helpers for higher-level collect_* convenience methods that build a Runner for you.

Fields

mode: ExecMode

Selected execution mode.

default_partitions: usize

Default partition count when neither the caller nor the planner suggests one.

checkpoint_config: Option<CheckpointConfig>

Available on crate feature checkpointing only.

Optional checkpoint configuration for fault tolerance.

Implementations

impl Runner

pub fn run_collect<T: 'static + Send + Sync + Clone>( &self, p: &Pipeline, terminal: NodeId, ) -> Result<Vec<T>>

Execute the pipeline ending at terminal, collecting the terminal vector as Vec<T>.

This function:

1. Builds an optimized plan with the planner.
2. Chooses sequential or parallel engine based on self.mode.
3. Honors planner’s suggested partitioning unless overridden.

Errors

An error is returned if the plan is malformed (e.g., a missing source), if a node encounters an unexpected input type, or if the terminal materialized type does not match T.

Panics

If the pipeline is in an inconsistent state, such as during concurrent modifications.

Examples found in repository

examples/checkpoint_recovery_demo.rs (line 50)

fn run_with_time_based_checkpoints() -> Result<()> {
    println!("=== Time-Based Checkpointing ===\n");

    let p = Pipeline::default();
    let data = from_vec(&p, (0..50_000).collect::<Vec<i32>>());

    let result_collection = data
        .map(|x: &i32| x * 2)
        .filter(|x: &i32| x % 5 == 0)
        .key_by(|x: &i32| x % 50)
        .combine_values(Count);

    let checkpoint_config = CheckpointConfig {
        enabled: true,
        directory: "./checkpoints_time_based".into(),
        policy: CheckpointPolicy::TimeInterval(2), // Every 2 seconds
        auto_recover: true,
        max_checkpoints: Some(3),
    };

    println!("Policy: Checkpoint every 2 seconds");
    println!("Max checkpoints: 3 (oldest will be deleted)\n");

    let runner = Runner {
        mode: ExecMode::Sequential,
        checkpoint_config: Some(checkpoint_config),
        ..Default::default()
    };

    let result = runner.run_collect::<(i32, u64)>(&p, result_collection.node_id())?;

    println!("\nCompleted! Results: {} groups", result.len());
    Ok(())
}

#[cfg(feature = "checkpointing")]
fn run_with_node_based_checkpoints() -> Result<()> {
    println!("=== Node-Based Checkpointing ===\n");

    let p = Pipeline::default();
    let data = from_vec(&p, (0..30_000).collect::<Vec<i32>>());

    let result_collection = data
        .map(|x: &i32| x + 1)
        .map(|x: &i32| x * 3)
        .filter(|x: &i32| *x > 100)
        .key_by(|x: &i32| x % 20)
        .map_values(|x: &i32| f64::from(*x))
        .combine_values(AverageF64);

    let checkpoint_config = CheckpointConfig {
        enabled: true,
        directory: "./checkpoints_node_based".into(),
        policy: CheckpointPolicy::EveryNNodes(2), // Every 2 nodes
        auto_recover: true,
        max_checkpoints: Some(5),
    };

    println!("Policy: Checkpoint every 2 nodes");
    println!("This creates more frequent checkpoints for fine-grained recovery\n");

    let runner = Runner {
        mode: ExecMode::Sequential,
        checkpoint_config: Some(checkpoint_config),
        ..Default::default()
    };

    let result = runner.run_collect::<(i32, f64)>(&p, result_collection.node_id())?;

    println!("\nCompleted! Results: {} groups", result.len());
    println!("Sample averages:");
    for (k, v) in result.iter().take(5) {
        println!("  Key {k}: Average = {v:.2}");
    }
    Ok(())
}

#[cfg(feature = "checkpointing")]
fn run_with_hybrid_checkpoints() -> Result<()> {
    println!("=== Hybrid Checkpointing ===\n");

    let p = Pipeline::default();
    let data = from_vec(&p, (0..40_000).collect::<Vec<i32>>());

    let result_collection = data
        .map(|x: &i32| x % 1000)
        .key_by(|x: &i32| *x)
        .map_values(|_: &i32| 1u64)
        .combine_values(Sum::<u64>::default())
        .map(|(k, v): &(i32, u64)| (k / 10, *v))
        .key_by(|(k, _): &(i32, u64)| *k)
        .map_values(|(_, v): &(i32, u64)| *v)
        .combine_values(Sum::<u64>::default());

    let checkpoint_config = CheckpointConfig {
        enabled: true,
        directory: "./checkpoints_hybrid".into(),
        policy: CheckpointPolicy::Hybrid {
            barriers: true,
            interval_secs: 3,
        },
        auto_recover: true,
        max_checkpoints: Some(10),
    };

    println!("Policy: Checkpoint after barriers OR every 3 seconds");
    println!("This provides the most aggressive checkpointing\n");

    let runner = Runner {
        mode: ExecMode::Sequential,
        checkpoint_config: Some(checkpoint_config),
        ..Default::default()
    };

    let result = runner.run_collect::<(i32, u64)>(&p, result_collection.node_id())?;

    println!("\nCompleted! Results: {} groups", result.len());
    Ok(())
}

examples/checkpointing_demo.rs (line 63)

fn main() -> Result<()> {
    println!("=== Ironbeam Checkpointing Demo ===\n");

    // Create a pipeline with a large dataset to simulate long-running jobs
    let p = Pipeline::default();

    println!("Building pipeline with 100,000 records...");
    let data = from_vec(&p, (0..100_000).collect::<Vec<i32>>());

    // Perform some transformations that would take time in a real scenario
    let transformed = data
        .map(|x: &i32| x * 2)
        .filter(|x: &i32| x % 3 == 0)
        .key_by(|x: &i32| x % 100) // Group by modulo 100
        .map_values(|x: &i32| x / 2)
        .combine_values(Sum::<i32>::default()); // Sum values per key

    // Configure checkpointing
    let checkpoint_config = CheckpointConfig {
        enabled: true,
        directory: "./Ironbeam_checkpoints".into(),
        policy: CheckpointPolicy::AfterEveryBarrier,
        auto_recover: true,
        max_checkpoints: Some(5),
    };

    println!("\nCheckpoint Configuration:");
    println!("  Directory: {:?}", checkpoint_config.directory.display());
    println!("  Policy: AfterEveryBarrier");
    println!("  Auto-recovery: enabled");
    println!("  Max checkpoints: 5");

    // Create a runner with checkpointing
    let runner = Runner {
        mode: ExecMode::Sequential,
        checkpoint_config: Some(checkpoint_config),
        ..Default::default()
    };

    println!("\nExecuting pipeline with checkpointing...");
    println!("(Watch for checkpoint messages during execution)\n");

    // Execute the pipeline
    let result = runner.run_collect::<(i32, i32)>(&p, transformed.node_id())?;

    println!("\n=== Execution Complete ===");
    println!("Total results: {}", result.len());
    println!("Sample results (first 10):");
    for (k, v) in result.iter().take(10) {
        println!("  Key {k}: Sum = {v}");
    }

    println!("\n=== Checkpointing Summary ===");
    println!("Checkpoints were automatically created after each barrier node.");
    println!("If this pipeline had failed, it could resume from the last checkpoint.");
    println!("Since execution completed successfully, all checkpoints were cleaned up.");

    Ok(())
}

Trait Implementations

impl Default for Runner

fn default() -> Self

Returns the “default value” for a type.