znippy_plugin_git/
graph.rs

1//! `__gunnar_graph__` — the commit graph as Arrow.
2//!
3//! Columns: `oid`, `parents` (list of oid hex), `tree`, `committer_time`,
4//! `generation`. Reserved, so ordinary `list` / `decompress` / iceberg readers
5//! skip it; and because it is a real Arrow IPC stream, slicing its manifest byte
6//! range hands DuckDB / Polars / DataFusion a queryable commit graph with zero
7//! consumer code.
8//!
9//! **Generation numbers** follow `gix-commitgraph` / git's own convention:
10//! a commit with no parent *inside this archive* has generation 1, and otherwise
11//! `generation = 1 + max(generation of parents present)`. So ancestry and
12//! merge-base tests become integer comparisons instead of object walks: if
13//! `gen(a) <= gen(b)` then `b` cannot be an ancestor of `a`.
14//!
15//! A parent that is absent from the archive contributes nothing to the max. That
16//! is the honest reading for a *cold tier*: a shallow or partially-repacked
17//! archive is a subgraph, and its generation numbers are only ever compared
18//! within it.
19
20use std::collections::HashMap;
21use std::path::Path;
22use std::sync::Arc;
23
24use anyhow::{Result, anyhow};
25use znippy_common::GUNNAR_GRAPH_MODULE;
26use znippy_common::arrow::array::{
27    Array, Int64Array, Int64Builder, ListBuilder, StringArray, StringBuilder, UInt32Array,
28    UInt32Builder,
29};
30use znippy_common::arrow::datatypes::{DataType, Field, Schema};
31use znippy_common::arrow::ipc::reader::StreamReader;
32use znippy_common::arrow::record_batch::RecordBatch;
33use znippy_common::read_reserved_section_bytes;
34
35/// One row of the commit graph.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct CommitNode {
38    pub oid: String,
39    pub parents: Vec<String>,
40    pub tree: Option<String>,
41    /// Committer timestamp, unix seconds. `None` when the commit carried none.
42    pub committer_time: Option<i64>,
43    /// 1 for a root (within this archive); `1 + max(parent generations)` else.
44    pub generation: u32,
45}
46
47pub fn graph_schema() -> Arc<Schema> {
48    Arc::new(Schema::new(vec![
49        Field::new("oid", DataType::Utf8, false),
50        Field::new(
51            "parents",
52            DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
53            false,
54        ),
55        Field::new("tree", DataType::Utf8, true),
56        Field::new("committer_time", DataType::Int64, true),
57        Field::new("generation", DataType::UInt32, false),
58    ]))
59}
60
61/// Assign generation numbers to `nodes` (which need not be in any order) and
62/// return them sorted by `(generation, oid)` — parents before children, which is
63/// also the order the reachability pass wants.
64///
65/// Iterative, never recursive: a linear commit history is a chain thousands deep
66/// and a recursive walk would blow the stack on a real repository.
67pub fn assign_generations(mut nodes: Vec<CommitNode>) -> Vec<CommitNode> {
68    let index: HashMap<&str, usize> = nodes
69        .iter()
70        .enumerate()
71        .map(|(i, n)| (n.oid.as_str(), i))
72        .collect();
73
74    // Kahn's algorithm over parent → child edges.
75    let n = nodes.len();
76    let mut parent_ids: Vec<Vec<usize>> = Vec::with_capacity(n);
77    let mut children: Vec<Vec<usize>> = vec![Vec::new(); n];
78    let mut indeg: Vec<usize> = vec![0; n];
79    for (i, node) in nodes.iter().enumerate() {
80        let mut ps = Vec::new();
81        for p in &node.parents {
82            if let Some(&pi) = index.get(p.as_str()) {
83                if pi != i {
84                    ps.push(pi);
85                }
86            }
87        }
88        ps.sort_unstable();
89        ps.dedup();
90        indeg[i] = ps.len();
91        for &pi in &ps {
92            children[pi].push(i);
93        }
94        parent_ids.push(ps);
95    }
96
97    let mut generation = vec![1u32; n];
98    let mut queue: Vec<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
99    let mut head = 0usize;
100    let mut settled = 0usize;
101    while head < queue.len() {
102        let i = queue[head];
103        head += 1;
104        settled += 1;
105        let g = generation[i];
106        for &c in &children[i] {
107            generation[c] = generation[c].max(g.saturating_add(1));
108            indeg[c] -= 1;
109            if indeg[c] == 0 {
110                queue.push(c);
111            }
112        }
113    }
114    // A cycle cannot occur in a well-formed commit DAG, but a corrupt archive is
115    // not a reason to hang or panic: unsettled nodes keep generation 1 and the
116    // graph is still written.
117    debug_assert!(settled == n || n == 0);
118
119    for (i, node) in nodes.iter_mut().enumerate() {
120        node.generation = generation[i];
121    }
122    nodes.sort_by(|a, b| a.generation.cmp(&b.generation).then_with(|| a.oid.cmp(&b.oid)));
123    nodes
124}
125
126/// Serialize the commit graph to a single Arrow record batch.
127pub fn build_graph_batch(nodes: &[CommitNode]) -> Result<RecordBatch> {
128    let n = nodes.len();
129    let mut oid_b = StringBuilder::with_capacity(n, n * 64);
130    let mut parents_b = ListBuilder::new(StringBuilder::new());
131    let mut tree_b = StringBuilder::with_capacity(n, n * 64);
132    let mut time_b = Int64Builder::with_capacity(n);
133    let mut gen_b = UInt32Builder::with_capacity(n);
134
135    for node in nodes {
136        oid_b.append_value(&node.oid);
137        for p in &node.parents {
138            parents_b.values().append_value(p);
139        }
140        parents_b.append(true);
141        match &node.tree {
142            Some(t) => tree_b.append_value(t),
143            None => tree_b.append_null(),
144        }
145        match node.committer_time {
146            Some(t) => time_b.append_value(t),
147            None => time_b.append_null(),
148        }
149        gen_b.append_value(node.generation);
150    }
151
152    RecordBatch::try_new(
153        graph_schema(),
154        vec![
155            Arc::new(oid_b.finish()),
156            Arc::new(parents_b.finish()),
157            Arc::new(tree_b.finish()),
158            Arc::new(time_b.finish()),
159            Arc::new(gen_b.finish()),
160        ],
161    )
162    .map_err(|e| anyhow!("commit-graph batch: {e}"))
163}
164
165/// Decode a `__gunnar_graph__` Arrow IPC section.
166pub fn decode_graph(bytes: &[u8]) -> Result<Vec<CommitNode>> {
167    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
168        .map_err(|e| anyhow!("commit-graph reader: {e}"))?;
169    let mut out = Vec::new();
170    for batch in reader {
171        let batch = batch.map_err(|e| anyhow!("commit-graph batch read: {e}"))?;
172        let oids = col::<StringArray>(&batch, "oid")?;
173        let parents = batch
174            .column_by_name("parents")
175            .ok_or_else(|| anyhow!("commit-graph: no `parents` column"))?
176            .as_any()
177            .downcast_ref::<znippy_common::arrow::array::ListArray>()
178            .ok_or_else(|| anyhow!("commit-graph: `parents` is not a list"))?;
179        let trees = col::<StringArray>(&batch, "tree")?;
180        let times = col::<Int64Array>(&batch, "committer_time")?;
181        let gens = col::<UInt32Array>(&batch, "generation")?;
182
183        for i in 0..batch.num_rows() {
184            let plist = parents.value(i);
185            let plist = plist
186                .as_any()
187                .downcast_ref::<StringArray>()
188                .ok_or_else(|| anyhow!("commit-graph: parent items are not strings"))?;
189            out.push(CommitNode {
190                oid: oids.value(i).to_string(),
191                parents: (0..plist.len()).map(|j| plist.value(j).to_string()).collect(),
192                tree: (!trees.is_null(i)).then(|| trees.value(i).to_string()),
193                committer_time: (!times.is_null(i)).then(|| times.value(i)),
194                generation: gens.value(i),
195            });
196        }
197    }
198    Ok(out)
199}
200
201/// Read the commit graph out of a sealed archive. `Ok(None)` when absent.
202pub fn read_graph(archive: &Path) -> Result<Option<Vec<CommitNode>>> {
203    match read_reserved_section_bytes(archive, GUNNAR_GRAPH_MODULE)? {
204        Some(b) => Ok(Some(decode_graph(&b)?)),
205        None => Ok(None),
206    }
207}
208
209fn col<'a, T: Array + 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
210    batch
211        .column_by_name(name)
212        .ok_or_else(|| anyhow!("commit-graph: no `{name}` column"))?
213        .as_any()
214        .downcast_ref::<T>()
215        .ok_or_else(|| anyhow!("commit-graph: `{name}` has an unexpected type"))
216}
217
218#[cfg(test)]
219mod tests {
220    use super::*;
221
222    fn node(oid: &str, parents: &[&str]) -> CommitNode {
223        CommitNode {
224            oid: oid.to_string(),
225            parents: parents.iter().map(|s| s.to_string()).collect(),
226            tree: Some(format!("t{}", &oid[1..])),
227            committer_time: Some(1_700_000_000),
228            generation: 0,
229        }
230    }
231
232    fn hexid(c: char) -> String {
233        std::iter::repeat_n(c, 40).collect()
234    }
235
236    #[test]
237    fn generations_are_one_plus_the_max_parent() {
238        let a = hexid('a');
239        let b = hexid('b');
240        let c = hexid('c');
241        let d = hexid('d');
242        // a -> b -> d, a -> c -> d  (d is a merge)
243        let nodes = assign_generations(vec![
244            node(&d, &[&b, &c]),
245            node(&b, &[&a]),
246            node(&c, &[&a]),
247            node(&a, &[]),
248        ]);
249        let g: HashMap<&str, u32> = nodes.iter().map(|n| (n.oid.as_str(), n.generation)).collect();
250        assert_eq!(g[a.as_str()], 1, "root");
251        assert_eq!(g[b.as_str()], 2);
252        assert_eq!(g[c.as_str()], 2);
253        assert_eq!(g[d.as_str()], 3, "merge is 1 + max(2,2)");
254
255        // Sorted parents-before-children.
256        let gens: Vec<u32> = nodes.iter().map(|n| n.generation).collect();
257        assert!(gens.windows(2).all(|w| w[0] <= w[1]), "not topologically ordered: {gens:?}");
258    }
259
260    #[test]
261    fn a_parent_outside_the_archive_does_not_inflate_the_generation() {
262        let a = hexid('a');
263        let missing = hexid('f');
264        let nodes = assign_generations(vec![node(&a, &[&missing])]);
265        assert_eq!(nodes[0].generation, 1, "a commit whose parent is not in the archive is a root here");
266    }
267
268    #[test]
269    fn deep_chain_does_not_blow_the_stack() {
270        // 50_000 deep — a recursive generation walk dies here.
271        let ids: Vec<String> = (0..50_000u32).map(|i| format!("{i:040x}")).collect();
272        let mut nodes = Vec::with_capacity(ids.len());
273        for (i, id) in ids.iter().enumerate() {
274            let parents: Vec<&str> = if i == 0 { vec![] } else { vec![ids[i - 1].as_str()] };
275            nodes.push(node(id, &parents));
276        }
277        let out = assign_generations(nodes);
278        let max = out.iter().map(|n| n.generation).max().unwrap();
279        assert_eq!(max, 50_000);
280    }
281
282    #[test]
283    fn batch_roundtrips_through_arrow_ipc() {
284        let a = hexid('a');
285        let b = hexid('b');
286        let nodes = assign_generations(vec![node(&b, &[&a]), node(&a, &[])]);
287        let batch = build_graph_batch(&nodes).unwrap();
288
289        let mut buf = Vec::new();
290        {
291            let mut w = znippy_common::arrow::ipc::writer::StreamWriter::try_new(
292                &mut buf,
293                &graph_schema(),
294            )
295            .unwrap();
296            w.write(&batch).unwrap();
297            w.finish().unwrap();
298        }
299        let back = decode_graph(&buf).unwrap();
300        assert_eq!(back, nodes);
301        assert_eq!(back[1].parents, vec![a.clone()]);
302        assert_eq!(back[1].generation, 2);
303    }
304}
znippy_plugin_git/graph.rs

znippy_plugin_git/
graph.rs
