lix 0.15.0

Embeddable version control for apps and AI agents.
Documentation 
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

//! Declarative rows→Arrow batch building shared by the table specs.
//!
//! A spec describes its output columns once as a [`ColumnTable`] — a static
//! list of `(name, accessor)` pairs — and [`ColumnTable::build`] materializes
//! any projected schema from it, matching fields by name. This replaces the
//! per-provider hand-written `*_record_batch` match ladders while keeping
//! every error message under the caller's control via [`ColumnTableError`].

use std::sync::Arc;

use datafusion::arrow::array::{ArrayRef, BooleanArray, Int64Array, LargeBinaryArray, StringArray};
use datafusion::arrow::datatypes::SchemaRef;
use datafusion::arrow::error::ArrowError;
use datafusion::arrow::record_batch::{RecordBatch, RecordBatchOptions};

use crate::LixError;

/// How one output column is produced from a row of `R`.
///
/// Accessors are capture-free `fn` pointers so tables can live in statics;
/// formatting/serialization stays exactly where the old match arms had it.
pub(super) enum Col<R: 'static> {
    /// Borrowing text column (fast path, no per-cell allocation).
    Utf8(for<'a> fn(&'a R) -> Option<&'a str>),
    /// Owned text column (serialization, `to_string()` ids).
    Utf8Owned(fn(&R) -> Option<String>),
    /// Owned text column whose accessor can fail (e.g. row-pk projection).
    Utf8Fallible(fn(&R) -> Result<Option<String>, LixError>),
    Bool(fn(&R) -> Option<bool>),
    I64(fn(&R) -> Option<i64>),
    Binary(fn(&R) -> Option<Vec<u8>>),
}

/// Build failure, kept structural so each table maps it to its own
/// byte-identical error message.
pub(super) enum ColumnTableError {
    /// The projected schema asked for a column the table does not define.
    UnsupportedColumn(String),
    /// Arrow rejected the assembled batch.
    Arrow(ArrowError),
    /// Arrow rejected the zero-column (row-count-only) batch.
    ArrowZeroColumn(ArrowError),
    /// A fallible cell accessor failed.
    Row(LixError),
}

pub(super) struct ColumnTable<R: 'static> {
    pub(super) columns: &'static [(&'static str, Col<R>)],
}

impl<R> ColumnTable<R> {
    /// Look up a column's accessor by name. Specs that mix table-driven
    /// system columns with bespoke ones (schema surfaces) resolve the shared
    /// accessor here and materialize it with [`build_array`].
    pub(super) fn col(&self, name: &str) -> Option<&Col<R>> {
        self.columns
            .iter()
            .find(|(column_name, _)| *column_name == name)
            .map(|(_, col)| col)
    }

    /// Build a batch matching `schema`'s fields by name. The zero-column
    /// projection (`SELECT count(*)`) is handled here once for every table.
    pub(super) fn build(
        &self,
        schema: SchemaRef,
        rows: &[R],
    ) -> Result<RecordBatch, ColumnTableError> {
        if schema.fields().is_empty() {
            let options = RecordBatchOptions::new().with_row_count(Some(rows.len()));
            return RecordBatch::try_new_with_options(schema, vec![], &options)
                .map_err(ColumnTableError::ArrowZeroColumn);
        }

        let arrays = schema
            .fields()
            .iter()
            .map(|field| {
                let name = field.name().as_str();
                let col = self
                    .col(name)
                    .ok_or_else(|| ColumnTableError::UnsupportedColumn(name.to_string()))?;
                build_array(col, rows)
            })
            .collect::<Result<Vec<_>, ColumnTableError>>()?;

        RecordBatch::try_new(schema, arrays).map_err(ColumnTableError::Arrow)
    }
}

/// Materialize one output column from `rows` via its accessor. The accessor
/// is an `fn` pointer evaluated once per row, exactly like the hand-written
/// match arms this replaces.
#[expect(trivial_casts)]
pub(super) fn build_array<R>(col: &Col<R>, rows: &[R]) -> Result<ArrayRef, ColumnTableError> {
    Ok(match col {
        Col::Utf8(get) => string_array(rows.iter().map(get)),
        Col::Utf8Owned(get) => Arc::new(StringArray::from_iter(rows.iter().map(get))) as ArrayRef,
        Col::Utf8Fallible(get) => Arc::new(StringArray::from(
            rows.iter()
                .map(get)
                .collect::<Result<Vec<_>, LixError>>()
                .map_err(ColumnTableError::Row)?,
        )) as ArrayRef,
        Col::Bool(get) => Arc::new(BooleanArray::from_iter(rows.iter().map(get))) as ArrayRef,
        Col::I64(get) => Arc::new(Int64Array::from_iter(rows.iter().map(get))) as ArrayRef,
        Col::Binary(get) => Arc::new(LargeBinaryArray::from(
            rows.iter()
                .map(get)
                .collect::<Vec<_>>()
                .iter()
                .map(|value| value.as_deref())
                .collect::<Vec<_>>(),
        )) as ArrayRef,
    })
}

/// Nullable Utf8 array from borrowed values; shared by the spec files.
#[expect(trivial_casts)]
pub(super) fn string_array<'a>(values: impl Iterator<Item = Option<&'a str>>) -> ArrayRef {
    Arc::new(StringArray::from_iter(values)) as ArrayRef
}