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

znippy_plugin_git/
graph.rs
