1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
// src/graph/subgraph.rs
//! Subgraph extraction and selection expansion operations
use crate::graph::schema::{CurrentSelection, DirGraph, EdgeData, SchemaInstall};
use crate::graph::storage::{GraphRead, GraphWrite};
use petgraph::graph::NodeIndex;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
/// Expand the current selection by N hops using BFS.
///
/// This function takes all currently selected nodes and expands the selection
/// to include all nodes within `hops` distance from any selected node.
/// The expansion considers edges in both directions (undirected).
pub fn expand_selection(
graph: &DirGraph,
selection: &mut CurrentSelection,
hops: usize,
) -> Result<(), String> {
let level_idx = selection.get_level_count().saturating_sub(1);
let level = selection
.get_level(level_idx)
.ok_or_else(|| "No active selection level".to_string())?;
// Start with current selection
let mut frontier: HashSet<NodeIndex> = level.iter_node_indices().collect();
let mut visited = frontier.clone();
// BFS expansion for N hops
let g = &graph.graph;
for _ in 0..hops {
let mut next_frontier = HashSet::new();
for &node in &frontier {
// Add all neighbors (both directions)
for neighbor in g.neighbors_undirected(node) {
// Only add if not already visited
if visited.insert(neighbor) {
next_frontier.insert(neighbor);
}
}
}
// If no new nodes were found, stop early
if next_frontier.is_empty() {
break;
}
frontier = next_frontier;
}
// Update selection with expanded nodes
let level_mut = selection
.get_level_mut(level_idx)
.ok_or_else(|| "Failed to get mutable selection level".to_string())?;
level_mut.selections.clear();
level_mut.add_selection(None, visited.into_iter().collect());
Ok(())
}
/// Extract a subgraph containing only the selected nodes and edges between them.
///
/// This creates an independent copy of the graph containing only the nodes
/// in the current selection and all edges that connect those nodes.
pub fn extract_subgraph(
source: &DirGraph,
selection: &CurrentSelection,
) -> Result<DirGraph, String> {
// Arena guard: disk-backed node/edge reads materialize into the query
// arena (protocol in disk/graph.rs); no-op on memory/mapped.
let _arena_guard = source.graph.begin_query();
let level_idx = selection.get_level_count().saturating_sub(1);
let level = selection
.get_level(level_idx)
.ok_or_else(|| "No active selection level".to_string())?;
let nodes = level.get_all_nodes();
let node_set: HashSet<NodeIndex> = nodes.iter().copied().collect();
let mut new_graph = DirGraph::new();
// Copy interner so the subgraph can resolve InternedKeys from compact storage
new_graph.interner = source.interner.clone();
// Copy type schemas so compact property storage works correctly
new_graph.type_schemas = source.type_schemas.clone();
// Carry the source's column stores. A copied node keeps its `row_id`, and
// since D1 Phase 3 that row id means nothing without the store the backend
// owns — before, the node carried its own `Arc` and its properties
// travelled with it. Sharing the `Arc` keeps every row id valid; the rows
// belonging to unselected nodes are orphans, which `enable_columnar`
// already detects and compacts away on the next save.
for (type_key, store) in source
.graph
.column_stores_iter()
.map(|(k, v)| (k, Arc::clone(v)))
.collect::<Vec<_>>()
{
GraphWrite::install_column_store(&mut new_graph.graph, type_key, store);
}
// Map from old node indices to new node indices
let mut index_map: HashMap<NodeIndex, NodeIndex> = HashMap::with_capacity(nodes.len());
// Copy selected nodes
for &old_idx in &nodes {
if let Some(node_data) = source.graph.node_weight(old_idx) {
// Add to new graph (single clone instead of double)
let new_idx = GraphWrite::add_node(&mut new_graph.graph, node_data.clone());
index_map.insert(old_idx, new_idx);
// Update type indices
new_graph
.type_indices
.entry_or_default(node_data.node_type_str(&source.interner).to_string())
.push(new_idx);
}
}
// Copy edges between selected nodes
for &old_source_idx in &nodes {
for edge in source.graph.edges(old_source_idx) {
let old_target_idx = edge.target();
// Only copy edge if target is also in selection
if node_set.contains(&old_target_idx) {
if let (Some(&new_source), Some(&new_target)) = (
index_map.get(&old_source_idx),
index_map.get(&old_target_idx),
) {
// Clone edge data (properties are already interned)
let edge_data = EdgeData::new_interned(
edge.weight().connection_type,
edge.weight().properties.clone(),
);
GraphWrite::add_edge(&mut new_graph.graph, new_source, new_target, edge_data);
}
}
}
}
// Copy schema definition if present. The subgraph's nodes are a subset of a
// graph that already satisfied these constraints, so installing them cannot
// find a duplicate the source did not have — but surface the error rather
// than discard it, so a genuine inconsistency in the source is not laundered
// into a silently unconstrained copy.
if let Some(schema) = source.get_schema() {
new_graph
// The target is a fresh graph, so merge and replace coincide; merge
// states the intent (install this schema) without also asserting
// that everything unnamed should be withdrawn.
.set_schema(schema.clone(), SchemaInstall::Merge)
.map_err(|violation| format!("subgraph schema install failed: {violation}"))?;
}
Ok(new_graph)
}
/// Get summary statistics about the subgraph that would be extracted.
///
/// Returns the number of nodes and edges that would be included.
pub fn get_subgraph_stats(
source: &DirGraph,
selection: &CurrentSelection,
) -> Result<SubgraphStats, String> {
// Arena guard: disk-backed node/edge reads materialize into the query
// arena (protocol in disk/graph.rs); no-op on memory/mapped.
let _arena_guard = source.graph.begin_query();
let level_idx = selection.get_level_count().saturating_sub(1);
let level = selection
.get_level(level_idx)
.ok_or_else(|| "No active selection level".to_string())?;
let nodes = level.get_all_nodes();
let node_set: HashSet<NodeIndex> = nodes.iter().copied().collect();
// Count edges between selected nodes
let mut edge_count = 0;
let mut connection_types: HashMap<String, usize> = HashMap::new();
let mut node_types: HashMap<String, usize> = HashMap::new();
// Count node types
for &node_idx in &nodes {
if let Some(node) = source.graph.node_view(node_idx) {
*node_types
.entry(node.node_type_str(&source.interner).to_string())
.or_insert(0) += 1;
}
}
// Count edges and connection types
for &source_idx in &nodes {
for edge in source.graph.edges(source_idx) {
if node_set.contains(&edge.target()) {
edge_count += 1;
let conn_type = edge.weight().connection_type_str(&source.interner);
*connection_types.entry(conn_type.to_string()).or_insert(0) += 1;
}
}
}
Ok(SubgraphStats {
node_count: nodes.len(),
edge_count,
node_types,
connection_types,
})
}
/// Statistics about a potential subgraph extraction
#[derive(Debug, Clone)]
pub struct SubgraphStats {
pub node_count: usize,
pub edge_count: usize,
pub node_types: HashMap<String, usize>,
pub connection_types: HashMap<String, usize>,
}