use crate::crate_graph::CrateGraph;
use crate::crate_schema::{crate_edge_col, crate_node_col};
use crate::schema::{edge_col, node_col};
use arrow::array::{Array, BooleanArray, RecordBatch, StringArray};
use std::collections::{HashMap, HashSet, VecDeque};
pub fn sort_crates(graph: &CrateGraph) -> Result<Vec<String>, String> {
crate::crate_graph::topo_sort_crates(graph)
}
pub fn sort_crates_parallel(graph: &CrateGraph) -> Result<Vec<Vec<String>>, String> {
let id_col = graph
.crate_nodes
.column(crate_node_col::ID)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CrateNode id column is not StringArray")?;
let wm_col = graph
.crate_nodes
.column(crate_node_col::WORKSPACE_MEMBER)
.as_any()
.downcast_ref::<BooleanArray>()
.ok_or("CrateNode workspace_member column is not BooleanArray")?;
let workspace_members: HashSet<String> = (0..id_col.len())
.filter(|&i| wm_col.value(i))
.map(|i| id_col.value(i).to_string())
.collect();
build_parallel_layers(&workspace_members, |layers| {
wire_crate_edges(graph, &workspace_members, layers)
})
}
pub fn sort_functions_in_crate(
crate_name: &str,
nodes: &RecordBatch,
edges: &RecordBatch,
) -> Result<Vec<String>, String> {
let (node_ids, adjacency, in_degree) = build_function_graph(crate_name, nodes, edges)?;
kahn_sort(node_ids, adjacency, in_degree)
}
pub fn sort_functions_parallel(
crate_name: &str,
nodes: &RecordBatch,
edges: &RecordBatch,
) -> Result<Vec<Vec<String>>, String> {
let (node_ids, adjacency, in_degree) = build_function_graph(crate_name, nodes, edges)?;
kahn_layers(node_ids, adjacency, in_degree)
}
#[derive(Debug)]
pub struct ParallelismStats {
pub layer_count: usize,
pub max_layer_width: usize,
pub critical_path: Vec<String>,
}
pub fn crate_parallelism_stats(graph: &CrateGraph) -> Result<ParallelismStats, String> {
let layers = sort_crates_parallel(graph)?;
let max_layer_width = layers.iter().map(|l| l.len()).max().unwrap_or(0);
let layer_count = layers.len();
let id_col = graph
.crate_nodes
.column(crate_node_col::ID)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CrateNode id column is not StringArray")?;
let wm_col = graph
.crate_nodes
.column(crate_node_col::WORKSPACE_MEMBER)
.as_any()
.downcast_ref::<BooleanArray>()
.ok_or("CrateNode workspace_member column is not BooleanArray")?;
let workspace_members: HashSet<String> = (0..id_col.len())
.filter(|&i| wm_col.value(i))
.map(|i| id_col.value(i).to_string())
.collect();
let mut fwd: HashMap<String, Vec<String>> = HashMap::new();
for name in &workspace_members {
fwd.entry(name.clone()).or_default();
}
wire_crate_edges(graph, &workspace_members, &mut fwd);
let flat = sort_crates(graph)?;
let mut dist: HashMap<String, usize> = HashMap::new();
let mut prev: HashMap<String, String> = HashMap::new();
for name in &flat {
dist.insert(name.clone(), 0);
}
for name in &flat {
let d = dist[name];
for nbr in fwd.get(name).into_iter().flatten() {
if workspace_members.contains(nbr) {
let entry = dist.entry(nbr.clone()).or_insert(0);
if d + 1 > *entry {
*entry = d + 1;
prev.insert(nbr.clone(), name.clone());
}
}
}
}
let sink = dist
.iter()
.max_by_key(|(_, v)| *v)
.map(|(k, _)| k.clone())
.unwrap_or_default();
let mut path = Vec::new();
let mut cur = sink.clone();
loop {
path.push(cur.clone());
match prev.get(&cur) {
Some(p) => cur = p.clone(),
None => break,
}
}
path.reverse();
Ok(ParallelismStats {
layer_count,
max_layer_width,
critical_path: path,
})
}
fn wire_crate_edges(
graph: &CrateGraph,
workspace_members: &HashSet<String>,
adjacency: &mut HashMap<String, Vec<String>>,
) {
let source_col = graph
.crate_edges
.column(crate_edge_col::SOURCE)
.as_any()
.downcast_ref::<StringArray>();
let target_col = graph
.crate_edges
.column(crate_edge_col::TARGET)
.as_any()
.downcast_ref::<StringArray>();
let sk_col = graph
.crate_edges
.column(crate_edge_col::SOURCE_KIND)
.as_any()
.downcast_ref::<StringArray>();
let dev_col = graph
.crate_edges
.column(crate_edge_col::DEV_DEP)
.as_any()
.downcast_ref::<BooleanArray>();
let (Some(src), Some(tgt), Some(sk), Some(dev)) = (source_col, target_col, sk_col, dev_col)
else {
return;
};
for i in 0..src.len() {
let kind = sk.value(i);
if kind != "path" && kind != "workspace" {
continue;
}
if dev.value(i) {
continue;
}
let s = src.value(i).to_string();
let t = tgt.value(i).to_string();
if workspace_members.contains(&s) && workspace_members.contains(&t) {
adjacency.entry(t).or_default().push(s);
}
}
}
fn kahn_sort(
nodes: HashSet<String>,
adjacency: HashMap<String, Vec<String>>,
in_degree: HashMap<String, usize>,
) -> Result<Vec<String>, String> {
let layers = kahn_layers(nodes, adjacency, in_degree)?;
Ok(layers.into_iter().flatten().collect())
}
fn kahn_layers(
nodes: HashSet<String>,
adjacency: HashMap<String, Vec<String>>,
in_degree: HashMap<String, usize>,
) -> Result<Vec<Vec<String>>, String> {
let mut in_deg = in_degree;
let mut adj = adjacency;
let mut queue: VecDeque<String> = nodes
.iter()
.filter(|n| in_deg.get(*n).copied().unwrap_or(0) == 0)
.cloned()
.collect();
let mut queue_vec: Vec<String> = queue.drain(..).collect();
queue_vec.sort();
let mut queue: VecDeque<String> = queue_vec.into();
let mut layers: Vec<Vec<String>> = Vec::new();
let mut visited = 0usize;
while !queue.is_empty() {
let mut current_layer: Vec<String> = queue.drain(..).collect();
current_layer.sort();
visited += current_layer.len();
let mut next_layer_candidates: Vec<String> = Vec::new();
for node in ¤t_layer {
for nbr in adj.remove(node).unwrap_or_default() {
let deg = in_deg.entry(nbr.clone()).or_insert(0);
if *deg > 0 {
*deg -= 1;
}
if *deg == 0 {
next_layer_candidates.push(nbr);
}
}
}
next_layer_candidates.sort();
next_layer_candidates.dedup();
layers.push(current_layer);
queue.extend(next_layer_candidates);
}
if visited < nodes.len() {
return Err(format!(
"cyclic dependency detected: {} nodes in cycle(s) — {} sorted of {}",
nodes.len() - visited,
visited,
nodes.len()
));
}
Ok(layers)
}
type FunctionGraph = (
HashSet<String>,
HashMap<String, Vec<String>>,
HashMap<String, usize>,
);
fn build_function_graph(
crate_name: &str,
nodes: &RecordBatch,
edges: &RecordBatch,
) -> Result<FunctionGraph, String> {
let id_col = nodes
.column(node_col::ID)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CodeNode id column is not StringArray")?;
let file_col = nodes
.column(node_col::FILE_PATH)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CodeNode file_path column is not StringArray")?;
let crate_prefix = format!("crates/{}/", crate_name);
let mut node_ids: HashSet<String> = HashSet::new();
let mut id_set: HashSet<String> = HashSet::new();
for i in 0..id_col.len() {
let fp = file_col.value(i);
if fp.starts_with(&crate_prefix) {
let id = id_col.value(i).to_string();
node_ids.insert(id.clone());
id_set.insert(id);
}
}
let src_col = edges
.column(edge_col::SOURCE_ID)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CodeEdge source_id column is not StringArray")?;
let tgt_col = edges
.column(edge_col::TARGET_ID)
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CodeEdge target_id column is not StringArray")?;
let pred_array = arrow::compute::cast(
edges.column(edge_col::PREDICATE),
&arrow::datatypes::DataType::Utf8,
)
.map_err(|e| format!("Failed to cast predicate column to Utf8: {e}"))?;
let pred_col = pred_array
.as_any()
.downcast_ref::<StringArray>()
.ok_or("CodeEdge predicate column is not StringArray after cast")?;
let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
let mut in_degree: HashMap<String, usize> = HashMap::new();
for id in &node_ids {
adjacency.entry(id.clone()).or_default();
in_degree.entry(id.clone()).or_insert(0);
}
for i in 0..src_col.len() {
let pred = pred_col.value(i);
if pred != "calls" && pred != "uses" {
continue;
}
let src = src_col.value(i).to_string();
let tgt = tgt_col.value(i).to_string();
if id_set.contains(&src) && id_set.contains(&tgt) && src != tgt {
adjacency.entry(tgt).or_default().push(src.clone());
*in_degree.entry(src).or_insert(0) += 1;
}
}
Ok((node_ids, adjacency, in_degree))
}
fn build_parallel_layers(
workspace_members: &HashSet<String>,
wire: impl Fn(&mut HashMap<String, Vec<String>>),
) -> Result<Vec<Vec<String>>, String> {
let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
let mut in_degree: HashMap<String, usize> = HashMap::new();
for name in workspace_members {
adjacency.entry(name.clone()).or_default();
in_degree.entry(name.clone()).or_insert(0);
}
wire(&mut adjacency);
let mut in_deg: HashMap<String, usize> = HashMap::new();
for name in workspace_members {
in_deg.entry(name.clone()).or_insert(0);
}
for targets in adjacency.values() {
for t in targets {
if workspace_members.contains(t) {
*in_deg.entry(t.clone()).or_insert(0) += 1;
}
}
}
kahn_layers(workspace_members.clone(), adjacency, in_deg)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::crate_graph::build_crate_graph;
use std::path::PathBuf;
fn workspace_root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.parent()
.unwrap()
.to_path_buf()
}
#[test]
fn test_sort_crates_wraps_topo_sort() {
let graph = build_crate_graph(&workspace_root()).expect("build_crate_graph should succeed");
let order = sort_crates(&graph).expect("no cycles expected");
assert!(!order.is_empty(), "should return at least one crate");
let core_pos = order.iter().position(|n| n == "nusy-arrow-core");
let being_pos = order.iter().position(|n| n == "nusy-being");
if let (Some(c), Some(b)) = (core_pos, being_pos) {
assert!(c < b, "nusy-arrow-core must come before nusy-being");
}
}
#[test]
fn test_sort_crates_parallel_layers_are_independent() {
let graph = build_crate_graph(&workspace_root()).expect("build_crate_graph should succeed");
let layers = sort_crates_parallel(&graph).expect("no cycles expected");
assert!(!layers.is_empty(), "should have at least one layer");
let total: usize = layers.iter().map(|l| l.len()).sum();
assert!(total > 0, "at least some workspace crates returned");
let flat: HashSet<String> = layers.iter().flatten().cloned().collect();
assert_eq!(
flat.len(),
total,
"no duplicates across layers: {} unique, {} total",
flat.len(),
total
);
let layer_index: HashMap<String, usize> = layers
.iter()
.enumerate()
.flat_map(|(i, layer)| layer.iter().map(move |name| (name.clone(), i)))
.collect();
use crate::crate_schema::crate_edge_col;
use arrow::array::StringArray;
let src_col = graph
.crate_edges
.column(crate_edge_col::SOURCE)
.as_any()
.downcast_ref::<StringArray>()
.expect("source column is StringArray");
let tgt_col = graph
.crate_edges
.column(crate_edge_col::TARGET)
.as_any()
.downcast_ref::<StringArray>()
.expect("target column is StringArray");
let dev_col = graph
.crate_edges
.column(crate_edge_col::DEV_DEP)
.as_any()
.downcast_ref::<BooleanArray>()
.expect("dev_dep column is BooleanArray");
for i in 0..src_col.len() {
if dev_col.value(i) {
continue; }
let dependent = src_col.value(i); let dep = tgt_col.value(i); let (Some(&dep_layer), Some(&dependent_layer)) =
(layer_index.get(dep), layer_index.get(dependent))
else {
continue; };
assert!(
dep_layer < dependent_layer,
"layer independence violated: '{dependent}' depends on '{dep}' but '{dep}' is in layer {dep_layer} and '{dependent}' is in layer {dependent_layer}; dependency must be in earlier layer"
);
}
}
#[test]
fn test_sort_crates_parallel_nusy_arrow_core_in_first_layers() {
let graph = build_crate_graph(&workspace_root()).expect("build_crate_graph should succeed");
let layers = sort_crates_parallel(&graph).expect("no cycles");
let core_layer = layers
.iter()
.position(|l| l.contains(&"nusy-arrow-core".to_string()));
assert!(
core_layer.is_some(),
"nusy-arrow-core must appear in parallel layers"
);
assert!(
core_layer.unwrap() <= 1,
"nusy-arrow-core is a root crate, should be in layer 0 or 1 (got {})",
core_layer.unwrap()
);
}
#[test]
fn test_sort_crates_parallel_count_matches_flat_sort() {
let graph = build_crate_graph(&workspace_root()).expect("build_crate_graph should succeed");
let flat = sort_crates(&graph).expect("no cycles");
let layers = sort_crates_parallel(&graph).expect("no cycles");
let parallel_total: usize = layers.iter().map(|l| l.len()).sum();
assert_eq!(
flat.len(),
parallel_total,
"flat sort and parallel sort must cover the same crates"
);
}
#[test]
fn test_detects_cyclic_dependency() {
use crate::crate_schema::{crate_edge_schema, crate_node_schema};
use arrow::array::{BooleanArray, RecordBatch, StringArray};
use std::sync::Arc;
let nodes = RecordBatch::try_new(
crate_node_schema(),
vec![
Arc::new(StringArray::from(vec!["crate-a", "crate-b"])),
Arc::new(StringArray::from(vec!["0.1.0", "0.1.0"])),
Arc::new(BooleanArray::from(vec![true, true])),
Arc::new(StringArray::from(vec![None::<&str>, None])),
Arc::new(StringArray::from(vec!["2021", "2021"])),
],
)
.expect("build test nodes");
let edges = RecordBatch::try_new(
crate_edge_schema(),
vec![
Arc::new(StringArray::from(vec!["crate-a", "crate-b"])),
Arc::new(StringArray::from(vec!["crate-b", "crate-a"])),
Arc::new(StringArray::from(vec!["*", "*"])),
Arc::new(BooleanArray::from(vec![false, false])),
Arc::new(BooleanArray::from(vec![false, false])),
Arc::new(BooleanArray::from(vec![false, false])),
Arc::new(StringArray::from(vec!["path", "path"])),
],
)
.expect("build test edges");
let graph = CrateGraph {
crate_nodes: nodes,
crate_edges: edges,
};
let result = sort_crates(&graph);
assert!(result.is_err(), "cycle should be detected");
let err = result.unwrap_err();
assert!(
err.contains("cycle") || err.contains("cyclic"),
"error message should mention cycle: {err}"
);
}
#[test]
fn test_parallelism_stats() {
let graph = build_crate_graph(&workspace_root()).expect("build_crate_graph should succeed");
let stats = crate_parallelism_stats(&graph).expect("stats should succeed");
assert!(
stats.layer_count >= 3,
"expected >= 3 layers, got {}",
stats.layer_count
);
assert!(
stats.max_layer_width >= 2,
"expected max_layer_width >= 2, got {}",
stats.max_layer_width
);
assert!(
!stats.critical_path.is_empty(),
"critical path should not be empty"
);
assert!(
stats.critical_path.len() >= 2,
"critical path should span at least 2 crates"
);
}
#[test]
fn test_sort_functions_callee_before_caller() {
use crate::schema::{CodeEdge, CodeEdgePredicate, CodeNode, CodeNodeKind};
use crate::schema::{build_code_edges_batch, build_code_nodes_batch};
let nodes = build_code_nodes_batch(&[
CodeNode {
id: "fn_a".to_string(),
kind: CodeNodeKind::Function,
name: "a".to_string(),
file_path: Some("crates/tc/src/lib.rs".to_string()),
..Default::default()
},
CodeNode {
id: "fn_b".to_string(),
kind: CodeNodeKind::Function,
name: "b".to_string(),
file_path: Some("crates/tc/src/lib.rs".to_string()),
..Default::default()
},
])
.expect("build nodes");
let edges = build_code_edges_batch(&[CodeEdge {
source_id: "fn_a".to_string(),
target_id: "fn_b".to_string(),
predicate: CodeEdgePredicate::Calls,
weight: None,
commit_id: None,
}])
.expect("build edges");
let order = sort_functions_in_crate("tc", &nodes, &edges).expect("no cycle");
assert!(
order.contains(&"fn_a".to_string()),
"fn_a must be in sort output"
);
assert!(
order.contains(&"fn_b".to_string()),
"fn_b must be in sort output"
);
let pos_a = order.iter().position(|id| id == "fn_a").unwrap();
let pos_b = order.iter().position(|id| id == "fn_b").unwrap();
assert!(
pos_b < pos_a,
"callee fn_b (pos {pos_b}) must come before caller fn_a (pos {pos_a})"
);
}
#[test]
fn test_sort_functions_parallel() {
use crate::schema::build_code_nodes_batch;
use crate::schema::{CodeNode, CodeNodeKind, code_edges_schema};
use arrow::array::RecordBatch;
use std::sync::Arc;
let nodes = build_code_nodes_batch(&[
CodeNode {
id: "fn_x".to_string(),
kind: CodeNodeKind::Function,
name: "x".to_string(),
file_path: Some("crates/ind/src/lib.rs".to_string()),
..Default::default()
},
CodeNode {
id: "fn_y".to_string(),
kind: CodeNodeKind::Function,
name: "y".to_string(),
file_path: Some("crates/ind/src/lib.rs".to_string()),
..Default::default()
},
])
.expect("build nodes");
let edges = RecordBatch::new_empty(Arc::new(code_edges_schema()));
let layers = sort_functions_parallel("ind", &nodes, &edges).expect("no cycle");
let all_ids: Vec<String> = layers.iter().flatten().cloned().collect();
assert!(
all_ids.contains(&"fn_x".to_string()),
"fn_x must appear in parallel output"
);
assert!(
all_ids.contains(&"fn_y".to_string()),
"fn_y must appear in parallel output"
);
assert_eq!(
layers.len(),
1,
"two independent functions should be in one layer"
);
assert_eq!(layers[0].len(), 2, "both functions in layer 0");
}
#[test]
fn test_sort_functions_parallel_callee_in_earlier_layer() {
use crate::schema::{CodeEdge, CodeEdgePredicate, CodeNode, CodeNodeKind};
use crate::schema::{build_code_edges_batch, build_code_nodes_batch};
let nodes = build_code_nodes_batch(&[
CodeNode {
id: "fn_root".to_string(),
kind: CodeNodeKind::Function,
name: "root".to_string(),
file_path: Some("crates/tc2/src/lib.rs".to_string()),
..Default::default()
},
CodeNode {
id: "fn_helper".to_string(),
kind: CodeNodeKind::Function,
name: "helper".to_string(),
file_path: Some("crates/tc2/src/lib.rs".to_string()),
..Default::default()
},
CodeNode {
id: "fn_leaf".to_string(),
kind: CodeNodeKind::Function,
name: "leaf".to_string(),
file_path: Some("crates/tc2/src/lib.rs".to_string()),
..Default::default()
},
])
.expect("build nodes");
let edges = build_code_edges_batch(&[
CodeEdge {
source_id: "fn_root".to_string(),
target_id: "fn_helper".to_string(),
predicate: CodeEdgePredicate::Calls,
weight: None,
commit_id: None,
},
CodeEdge {
source_id: "fn_helper".to_string(),
target_id: "fn_leaf".to_string(),
predicate: CodeEdgePredicate::Calls,
weight: None,
commit_id: None,
},
])
.expect("build edges");
let layers = sort_functions_parallel("tc2", &nodes, &edges).expect("no cycle");
let layer_index: HashMap<String, usize> = layers
.iter()
.enumerate()
.flat_map(|(i, layer)| layer.iter().map(move |id| (id.clone(), i)))
.collect();
let root_layer = *layer_index.get("fn_root").expect("fn_root in layers");
let helper_layer = *layer_index.get("fn_helper").expect("fn_helper in layers");
let leaf_layer = *layer_index.get("fn_leaf").expect("fn_leaf in layers");
assert!(
leaf_layer < helper_layer,
"leaf (layer {leaf_layer}) must be in earlier layer than helper (layer {helper_layer})"
);
assert!(
helper_layer < root_layer,
"helper (layer {helper_layer}) must be in earlier layer than root (layer {root_layer})"
);
}
}