diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3129940..63914aa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,3 +39,29 @@ jobs: - name: Run tests run: cargo test + + - name: Build CLI binary + run: cargo build --bin graphframes + + - name: Smoke-test the CLI + run: | + set -euo pipefail + BIN=./target/debug/graphframes + V=./testing/data/ldbc/example-directed-v.parquet + E=./testing/data/ldbc/example-directed-e.parquet + COMMON="--vertices $V --edges $E --src-col-name source --dst-col-name target --checkpoint-dir /tmp/gf_ci_work" + + $BIN page-rank $COMMON --tol 0.01 --output file:///tmp/ci_pr/ + $BIN wcc $COMMON --seed 42 --output file:///tmp/ci_wcc/ + $BIN mis $COMMON --output file:///tmp/ci_mis/ + $BIN kcore $COMMON --output file:///tmp/ci_kcore/ + $BIN hyperanf $COMMON --n-hops 3 --output file:///tmp/ci_hyperanf/ + $BIN shortest-path $COMMON --landmarks 1 --output file:///tmp/ci_sp/ + + # Each command must have produced parquet output. + for d in pr wcc mis kcore hyperanf sp; do + if ! ls -A "/tmp/ci_$d" | grep -q .; then + echo "::error::no output written in /tmp/ci_$d" + exit 1 + fi + done diff --git a/.gitignore b/.gitignore index 9286fbf..d47656c 100644 --- a/.gitignore +++ b/.gitignore @@ -11,8 +11,10 @@ release/ *~ \#* *# -*.parquet gf_df_tmp* gf_checkpoints* test-test* .#* +graph500* +twitter_mpi* + diff --git a/Cargo.lock b/Cargo.lock index 269e33b..6099d6b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,6 +67,21 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" +[[package]] +name = "anstream" +version = "0.6.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a" +dependencies = [ + "anstyle", + "anstyle-parse 0.2.7", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + [[package]] name = "anstream" version = "1.0.0" @@ -74,7 +89,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" dependencies = [ "anstyle", - "anstyle-parse", + "anstyle-parse 1.0.0", "anstyle-query", "anstyle-wincon", "colorchoice", @@ -88,6 +103,15 @@ version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" +[[package]] +name = "anstyle-parse" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2" +dependencies = [ + "utf8parse", +] + [[package]] name = "anstyle-parse" version = "1.0.0" @@ -595,6 +619,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be92d32e80243a54711e5d7ce823c35c41c9d929dc4ab58e1276f625841aadf9" dependencies = [ "clap_builder", + "clap_derive", ] [[package]] @@ -603,8 +628,22 @@ version = "4.5.41" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "707eab41e9622f9139419d573eca0900137718000c517d47da73045f54331c3d" dependencies = [ + "anstream 0.6.21", "anstyle", "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.5.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef4f52386a59ca4c860f7393bcf8abd8dfd91ecccc0f774635ff68e92eeef491" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1584,7 +1623,7 @@ version = "0.11.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de671bd27a75a797dc9ae289ba1e77276e75e2026408aab65185384e2d5cd3f6" dependencies = [ - "anstream", + "anstream 1.0.0", "anstyle", "env_filter", "jiff", @@ -1819,6 +1858,7 @@ name = "graphframes-rs" version = "0.0.1" dependencies = [ "async-trait", + "clap", "criterion", "datafusion", "datasketches", @@ -3029,6 +3069,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" diff --git a/Cargo.toml b/Cargo.toml index 33b487e..26d712a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,8 @@ exclude = [ ] [dependencies] + +# Core dependencies datafusion = "54.1.0" async-trait = "0.1" tokio = {version = "1", features = ["full"] } @@ -16,11 +18,18 @@ url = "2" futures = "0.3" uuid = {version = "1", features = ["v4"] } log = "0.4" -env_logger = "0.11" -snmalloc-rs = "0.7" rand = { version = "0.9", features = ["std_rng"] } datasketches = { version = "0.3.0", default-features = false, features = ["hll"] } +# CLI dependencies +clap = { version = "4.5", features = ["derive", "env"], optional = true} +env_logger = { version = "0.11", optional = true } +snmalloc-rs = { version = "0.7", optional = true } + +[features] +default = ["cli"] +cli = ["dep:clap", "dep:env_logger", "dep:snmalloc-rs"] + [dev-dependencies] criterion = { version = "0.7", features = ["html_reports", "async_tokio"] } tokio = { version = "1", features = ["full"] } @@ -38,5 +47,6 @@ name = "sp_benchmark" harness = false [[bin]] -name = "run-algorithm" +name = "graphframes" path = "src/main.rs" +required-features = ["cli"] diff --git a/README.md b/README.md index d6b52ce..0a5f209 100644 --- a/README.md +++ b/README.md @@ -2,11 +2,72 @@ An experimental single node out-of core graph algorithms. -**!NOTE!** +## Usage -_At the moment it is just a core. The existing `main.rs` should be fine for benchmarking / experimenting, but it is NOT A STABLE PUBLIC CONTRACT! I'm (the author) have an experience with distributed (out-of-core) graph algorithms but I have zero experience with building publuc surface APIs for DataFusion projects. So, there is not public API behind the existing low-level builder methods of the `GraphFrame` struct. I do not know yet what is the best way to make a DataFusion plugin, how should it work (Python? FFI? Arrow Flight?) as well how such an API should looks like. Should it even be a programmatic API with full access to methods or just a CLI/Server that supports DSL? If you are interesting in the project and have an idea (or experience with DataFusion-based public surface APIs) how the public surface should look like please, open an issue and share. I will appreciate. I'm (the author) interested in learning Rust, DataFusion and out-of-core graph analytics, so for me the public surface is not the top priority!_ +### CLI -**Supported algorithms** +#### Installation + +```sh +cargo install --git https://github.com/SemyonSinchenko/graphframes-rs +``` + +This builds and installs the `graphframes` binary. The CLI is part of the default feature set, so no extra flags are required. + +#### CLI + +The binary takes an algorithm as a subcommand, reads a graph from a vertices file and an edges file, and writes the result as parquet to an output directory. + +```sh +graphframes page-rank \ + --vertices vertices.parquet \ + --edges edges.parquet \ + --output file:///tmp/pagerank/ \ + --tol 0.01 +``` + +Vertices must contain an Int64 `id` column; edges must contain Int64 `src` and `dst` columns. If your input uses different names, map them with `--id-col-name`, `--src-col-name`, and `--dst-col-name`. The input format defaults to parquet; `--format csv` and `--format json` are also available. + +Available algorithms: + +- `page-rank`: PageRank centrality; +- `wcc`: Weakly Connected Components; +- `mis`: Maximal Independent Set; +- `kcore`: K-Core decomposition; +- `hyperanf`: Approximate Neighbor Function; +- `shortest-path`: Multi-Source Shortest Path; + +Each has its own arguments — run `graphframes --help` for the full list. For what each algorithm computes, see [References](#references). + +#### Tuning + +This project is designed to be out-of-core and rely on DataFusion spilling during joins and manually offloading to disk intermediate stages. Three knobs trade off memory, parallelism, and disk usage. Each can be set on the command line **or** via an environment variable; the flag takes precedence over the env var, which takes precedence over the default. + +**a) Memory pool** — `--max-memory` (env `GRAPHFRAMES_MAX_MEMORY`, default `4G`). + +Size of the DataFusion's spill pool ([`FairSpillPool`](https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.FairSpillPool.html)). Operators that exceed their share of the pool spill intermediate data to disk instead of failing with an out-of-memory error. A larger value reduces spilling at the cost of RAM. + +**b) Parallelism** — `--num-workers` (env `GRAPHFRAMES_NUM_WORKERS`, default `2`). + +Maps to DataFusion's `target_partitions`: the number of partitions the engine processes concurrently. + +**c) The trade-off.** + +The spill pool is shared fairly among the operators running at the same time. Raising `--num-workers` runs more partitions concurrently, which tends to increase the number of concurrent memory consumers and so shrinks the fair share each one gets. More workers means more parallelism but more spilling (and slower work per partition); fewer workers means more memory per consumer but less parallelism. There is no universally correct value — measure on your own graph and hardware. Defaults are ~fine for something less than 500M of edges. + +**d) Input / output.** + +Input format is parquet by default; `--format csv` and `--format json` are supported and follow DataFusion's default schema inference and parsing. Output is always parquet, written to a `file://` directory; the directory is created if it does not exist. + +**e) Checkpoint / spill directory** — `--checkpoint-dir` (env `GRAPHFRAMES_WORKDIR`, default `gf_workdir`). + +The work directory holds two kinds of data: the algorithms' parquet checkpoints (written and re-read between iterations) and DataFusion's spill files. Both are high-throughput and latency-sensitive, and on large graphs the algorithms can stream large amounts of data through them. On my experiments during processing billion-scale graphs with limited memory (~5-6GB limit) the process read from disk more than 200GBs. Point this at the fastest local storage available, preferably a dedicated NVMe or SSD. Network filesystems and spinning disks will make spill and checkpoint latency dominate runtime. The directory is created if missing; relative paths resolve against the current directory. `--max-temp-file` (env `GRAPHFRAMES_MAX_TEMP_FILE`, default `200G`) bounds the total size of the spill subdirectory. For graphs of the size of few billions edges and vertices the default may be not enough. + +### Libraray + +TBD + +## References 1. **Pregel**: _Malewicz, Grzegorz, et al. "Pregel: a system for large-scale graph processing." Proceedings of the 2010 ACM SIGMOD International Conference on Management of data. 2010._ - **PageRank**: _Zadeh, R., et al. "Cme 323: Distributed algorithms and optimization, spring 2015." University Lecture (2015)._ diff --git a/src/lib.rs b/src/lib.rs index 90611a1..244d85d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,8 @@ mod expressions; mod memory; mod utils; +pub use utils::GraphFramesConfig; + use datafusion::arrow::datatypes::DataType; use datafusion::common::plan_err; use datafusion::error::Result; diff --git a/src/main.rs b/src/main.rs index 17a2c79..8374092 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,4 @@ +#[cfg(feature = "cli")] #[global_allocator] static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc; @@ -5,12 +6,196 @@ use datafusion::error::{DataFusionError, Result}; use datafusion::execution::memory_pool::FairSpillPool; use datafusion::execution::runtime_env::RuntimeEnvBuilder; use datafusion::execution::session_state::SessionStateBuilder; -use datafusion::object_store::path::Path; +use datafusion::object_store::path::Path as ObjectPath; use datafusion::prelude::*; -use graphframes_rs::{EDGE_DST, EDGE_SRC, GraphFrame}; -use std::env; +use graphframes_rs::GraphFramesConfig; +use graphframes_rs::{EDGE_DST, EDGE_SRC, GraphFrame, VERTEX_ID}; +use std::path::{Path, PathBuf}; use std::sync::Arc; +use clap::{Args, Parser, Subcommand, ValueEnum}; + +#[derive(Debug, Clone, ValueEnum)] +enum Format { + Parquet, + Csv, + Json, +} + +#[derive(Args, Debug)] +struct CommonArgs { + /// Path (or URI) to the vertices file or directory. + /// + /// Must contain an Int64 vertex-id column (`id` by default; override with + /// `--id-col-name`). Extra attribute columns are allowed but ignored. + #[arg(long)] + vertices: String, + + /// Path (or URI) to the edges file or directory. + /// + /// Must contain Int64 source/destination columns (`src`/`dst` by default; + /// override with `--src-col-name` / `--dst-col-name`). + #[arg(long)] + edges: String, + + /// Output directory as a `file://` URI, e.g. `file:///tmp/out/`. + /// + /// The result is always written as parquet regardless of `--format`; + /// parent directories are created automatically. + #[arg(long)] + output: String, + + /// Input file format for `--vertices` and `--edges`. + /// + /// Parquet by default. `csv` / `json` use DataFusion's default schema + /// inference and parsing options. + #[arg(long, value_enum, default_value_t = Format::Parquet)] + format: Format, + + /// Name of the vertex-id column in the input; renamed to the library's `id`. + #[arg(long, default_value = "id")] + id_col_name: String, + + /// Name of the edge source column in the input; renamed to the library's `src`. + #[arg(long, default_value = "src")] + src_col_name: String, + + /// Name of the edge destination column in the input; renamed to `dst`. + #[arg(long, default_value = "dst")] + dst_col_name: String, + + /// DataFusion spill-pool memory limit, e.g. `4G` or `512M`. + /// + /// A command-line value takes precedence over the environment variable. + #[arg(long, env = "GRAPHFRAMES_MAX_MEMORY", default_value = "4G")] + max_memory: String, + + /// Parallelism (= DataFusion `target_partitions`). + /// + /// A command-line value takes precedence over the environment variable. + #[arg(long, env = "GRAPHFRAMES_NUM_WORKERS", default_value_t = 2)] + num_workers: usize, + + /// Base working directory for checkpoints and DataFusion spill files. + /// + /// Relative paths resolve against the current directory; the directory and + /// its `checkpoints/` and `df-spill/` subdirectories are created if missing. + /// A command-line value takes precedence over the environment variable. + #[arg(long, env = "GRAPHFRAMES_WORKDIR", default_value = "gf_workdir")] + checkpoint_dir: String, + + /// Upper bound on the total size of DataFusion's temporary spill directory, + /// e.g. `200G`. + /// + /// A command-line value takes precedence over the environment variable. + #[arg(long, env = "GRAPHFRAMES_MAX_TEMP_FILE", default_value = "200G")] + max_temp_file: String, +} + +#[derive(Subcommand, Debug)] +enum Command { + /// PageRank via Pregel. + PageRank { + #[command(flatten)] + common: CommonArgs, + + #[arg(long)] + tol: f64, + + #[arg(long, default_value_t = 0.15)] + reset_prob: f64, + + #[arg(long, default_value_t = 0)] + max_iter: usize, + }, + + /// Weakly connected components via randomized contraction. + Wcc { + #[command(flatten)] + common: CommonArgs, + + /// Random seed for the per-iteration affine hashes. + #[arg(long)] + seed: u64, + }, + + /// Maximal independent set (Ghaffari's algorithm). + Mis { + #[command(flatten)] + common: CommonArgs, + }, + + /// k-core (coreness) decomposition. + Kcore { + #[command(flatten)] + common: CommonArgs, + + /// 0 (default) = run until convergence. + #[arg(long, default_value_t = 0)] + max_iter: usize, + }, + + /// HyperANF approximate neighbourhood function. + Hyperanf { + #[command(flatten)] + common: CommonArgs, + + /// Number of hops (also the iteration budget). + #[arg(long)] + n_hops: usize, + + /// HLL log2 of k. Must be in 4..=21. + #[arg(long, default_value_t = 12, value_parser = clap::value_parser!(u8).range(4..=21))] + lg_k: u8, + + /// Directed propagation (default). Pass `false` for the symmetric neighbourhood. + #[arg(long, default_value_t = true)] + directed: bool, + }, + + /// Multi-source shortest paths. + ShortestPath { + #[command(flatten)] + common: CommonArgs, + + /// Landmark vertex ids, comma-separated (e.g. --landmarks 1,4,7). + #[arg(long, value_delimiter = ',')] + landmarks: Vec, + + /// Iteration cap; default = effectively unbounded. + #[arg(long)] + max_iterations: Option, + + /// Compute distance *to* the landmarks (reverse edges) instead of from them. + #[arg(long)] + to_landmarks: bool, + }, +} + +#[derive(Parser, Debug)] +#[command( + name = "graphframes", + version, + about = "Out-of-core graph algorithms over DataFusion" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +struct ResolvedDir { + fs: PathBuf, + object: ObjectPath, +} + +fn ensure_dir(p: impl AsRef) -> Result { + let p = p.as_ref(); + std::fs::create_dir_all(p)?; + let fs = std::fs::canonicalize(p)?; + let object = ObjectPath::from_absolute_path(&fs)?; + Ok(ResolvedDir { fs, object }) +} + fn parse_mem(s: &str) -> Result { let s = s.trim(); @@ -35,93 +220,178 @@ fn parse_mem(s: &str) -> Result { Ok(num * mult) } -#[tokio::main] -async fn main() -> Result<()> { - env_logger::init(); - let args: Vec = env::args().collect(); - - let vertices = &args[1]; - let edges = &args[2]; - let algorithm = &args[3]; - let params = &args[4]; - let out = &args[5]; - let max_memory = &args[6]; - let num_partition: usize = args[7] - .parse() - .map_err(|e| DataFusionError::Execution(format!("invalid number of partitions: {e}")))?; - let max_temp_size = if args.len() == 9 { - let temp_limit = &args[8]; - parse_mem(temp_limit)? - } else { - parse_mem("100G")? - }; +fn build_context( + work: &ResolvedDir, + max_pool_mem: &str, + num_workers: usize, + max_temp_file_size: &str, +) -> Result { + let max_pool_mem = parse_mem(max_pool_mem)?; + let max_temp_file_size = parse_mem(max_temp_file_size)? as u64; + let spill = ensure_dir(work.fs.join("df-spill"))?; - let max_pool_mem = parse_mem(max_memory)?; - let env = RuntimeEnvBuilder::new() + let config = SessionConfig::from_env()? + .with_target_partitions(num_workers) + .with_option_extension(GraphFramesConfig::default()); + + let runtime_env = RuntimeEnvBuilder::new() .with_memory_pool(Arc::new(FairSpillPool::new(max_pool_mem))) - .with_temp_file_path(std::env::current_dir()?.join("gf_df_tmp")) - .with_max_temp_directory_size(max_temp_size as u64) + .with_temp_file_path(spill.fs) + .with_max_temp_directory_size(max_temp_file_size) .build_arc()?; + let session_state = SessionStateBuilder::new() - .with_config(SessionConfig::new().with_target_partitions(num_partition)) - .with_runtime_env(env) + .with_config(config) + .with_runtime_env(runtime_env) .with_default_features() .build(); - let ctx = SessionContext::from(session_state); - let vertices = ctx - .read_parquet(vertices, ParquetReadOptions::new()) - .await?; - - let edges = ctx - .read_parquet(edges, ParquetReadOptions::new()) - .await? - .select(vec![ - col("source").alias(EDGE_SRC), - col("target").alias(EDGE_DST), - ])?; - - let graph = GraphFrame::try_new(vertices, edges)?; - - if algorithm == "pagerank" { - // PageRank via Pregel. Algorithm-specifi parameter - // is convergence tolerance. - let tol = params.parse::().unwrap(); - let pr = graph - .pagerank() - .reset_prob(0.15) - .max_iter(0) - .tol(tol) - .set_checkpoint_dir(Path::from( - std::env::current_dir()? - .join("gf_checkpoints") - .to_string_lossy() - .as_ref(), - )) - .run(&ctx, out, false) - .await?; - - println!("num-iterations: {}", pr); - } else if algorithm == "wcc" { - // Weakly connected components via randomized contraction. The only - // algorithm-specific parameter is the random seed (used to seed the - // per-iteration affine hashes). - let seed: u64 = params - .parse() - .map_err(|e| DataFusionError::Execution(format!("invalid random seed: {e}")))?; - let cc = graph - .connected_components() - .set_checkpoint_dir(Path::from( - std::env::current_dir()? - .join("gf_checkpoints") - .to_string_lossy() - .as_ref(), - )) - .set_seed(seed) - .run(&ctx, out, false) - .await?; - - println!("num-iterations: {}", cc); + + Ok(SessionContext::from(session_state)) +} + +async fn read_data(ctx: &SessionContext, path: &str, format: Format) -> Result { + let r = match format { + Format::Parquet => ctx.read_parquet(path, ParquetReadOptions::new()).await?, + Format::Csv => ctx.read_csv(path, CsvReadOptions::new()).await?, + Format::Json => ctx.read_json(path, JsonReadOptions::default()).await?, }; + Ok(r) +} + +async fn build_graph( + ctx: &SessionContext, + vertices: &str, + edges: &str, + id_col: &str, + src_col: &str, + dst_col: &str, + format: Format, +) -> Result { + let r_vertices = read_data(ctx, vertices, format.clone()).await?; + let r_edges = read_data(ctx, edges, format.clone()).await?; + + let vertices = r_vertices.select(vec![col(id_col).alias(VERTEX_ID)])?; + + let edges = r_edges.select(vec![ + col(src_col).alias(EDGE_SRC), + col(dst_col).alias(EDGE_DST), + ])?; + + let g = GraphFrame::try_new(vertices, edges)?; + Ok(g) +} + +async fn setup(common: &CommonArgs) -> Result<(SessionContext, GraphFrame, ObjectPath)> { + let work = ensure_dir(&common.checkpoint_dir)?; + let checkpoints = ensure_dir(work.fs.join("checkpoints"))?; + + let ctx = build_context( + &work, + &common.max_memory, + common.num_workers, + &common.max_temp_file, + )?; + + let g = build_graph( + &ctx, + &common.vertices, + &common.edges, + &common.id_col_name, + &common.src_col_name, + &common.dst_col_name, + common.format.clone(), + ) + .await?; + + Ok((ctx, g, checkpoints.object)) +} + +#[tokio::main] +async fn main() -> Result<()> { + env_logger::init(); + + let cli = Cli::parse(); + + match cli.command { + Command::PageRank { + common, + tol, + reset_prob, + max_iter, + } => { + let (ctx, g, ckpt) = setup(&common).await?; + let _ = g + .pagerank() + .reset_prob(reset_prob) + .tol(tol) + .max_iter(max_iter) + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output, false) + .await?; + } + Command::Wcc { common, seed } => { + let (ctx, g, ckpt) = setup(&common).await?; + let _ = g + .connected_components() + .set_seed(seed) + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output, false) + .await?; + } + Command::Mis { common } => { + let (ctx, g, ckpt) = setup(&common).await?; + let _ = g + .maximal_independent_set() + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output) + .await?; + } + Command::Kcore { common, max_iter } => { + let (ctx, g, ckpt) = setup(&common).await?; + let _ = g + .k_core() + .max_iter(max_iter) + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output, false) + .await?; + } + Command::Hyperanf { + common, + n_hops, + lg_k, + directed, + } => { + let (ctx, g, ckpt) = setup(&common).await?; + let _ = g + .hyperanf() + .n_hops(n_hops) + .lg_k(lg_k) + .directed(directed) + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output, false) + .await?; + } + Command::ShortestPath { + common, + landmarks, + max_iterations, + to_landmarks, + } => { + let (ctx, g, ckpt) = setup(&common).await?; + let mut b = g.shortest_paths(landmarks); + if to_landmarks { + b = b.to_landmarks(); + } + if let Some(mi) = max_iterations { + b = b.max_iterations(mi); + } + let _ = b + .set_checkpoint_dir(ckpt) + .run(&ctx, &common.output, false) + .await?; + } + } + Ok(()) } diff --git a/testing/data/ldbc/example-directed-e.parquet b/testing/data/ldbc/example-directed-e.parquet new file mode 100644 index 0000000..7540655 Binary files /dev/null and b/testing/data/ldbc/example-directed-e.parquet differ diff --git a/testing/data/ldbc/example-directed-v.parquet b/testing/data/ldbc/example-directed-v.parquet new file mode 100644 index 0000000..69d2dd5 Binary files /dev/null and b/testing/data/ldbc/example-directed-v.parquet differ