lix 0.16.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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

use std::sync::Arc;

use async_trait::async_trait;
use datafusion::arrow::array::{Array, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::common::{DataFusionError, Result};
use datafusion::datasource::TableType;
use datafusion::execution::context::ExecutionProps;
use datafusion::logical_expr::Expr;
use datafusion::physical_plan::ExecutionPlan;

use crate::LixError;
use crate::sql2::error::lix_error_to_datafusion_error;
use crate::sql2::result_metadata::row_ref_field;
use crate::sql2::{DiffCommand, DiffCommandSelection, SqlWriteContext, WriteAccess};

use super::spec::{InsertApply, PlannedScan, TableSpec, register_spec_table, scan_row_source};

pub(super) async fn register_diff_command_provider(
    session: &datafusion::prelude::SessionContext,
    surface_name: &str,
    command: DiffCommand,
    write_ctx: SqlWriteContext,
) -> Result<(), LixError> {
    register_spec_table(
        session,
        surface_name,
        Arc::new(DiffCommandSpec {
            table_name: surface_name.to_string(),
            command,
        }),
        WriteAccess::write(write_ctx),
    )
}

struct DiffCommandSpec {
    table_name: String,
    command: DiffCommand,
}

#[async_trait]
impl TableSpec for DiffCommandSpec {
    fn table_name(&self) -> &str {
        &self.table_name
    }

    fn schema(&self) -> SchemaRef {
        command_schema()
    }

    fn table_type(&self) -> TableType {
        // A command sink is closest to an insertable view in the standard
        // information schema. Lix's exact classification is exposed through
        // information_schema.lix_surfaces.
        TableType::View
    }

    async fn plan_scan(
        &self,
        _projection: Option<&Vec<usize>>,
        _filters: &[Expr],
        _limit: Option<usize>,
        _props: &ExecutionProps,
    ) -> Result<PlannedScan> {
        let table = self.table_name.clone();
        Ok(PlannedScan {
            schema: command_schema(),
            ordering: None,
            source: scan_row_source(command_schema(), table, |table| async move {
                Err(DataFusionError::Execution(format!(
                    "{table} is a command sink and cannot be read"
                )))
            }),
        })
    }

    async fn plan_insert(
        &self,
        write_ctx: SqlWriteContext,
        _input: &Arc<dyn ExecutionPlan>,
    ) -> Result<Option<InsertApply>> {
        let command = self.command;
        Ok(Some(Arc::new(move |batches| {
            let write_ctx = write_ctx.clone();
            Box::pin(async move {
                let selections = selections_from_batches(&batches)?;
                if selections.is_empty() {
                    return Ok(0);
                }
                write_ctx
                    .execute_diff_command(command, selections)
                    .await
                    .map(|outcome| outcome.rows_affected)
                    .map_err(lix_error_to_datafusion_error)
            })
        })))
    }
}

fn command_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        row_ref_field("row_ref", false),
        Field::new("commit_id", DataType::Utf8, false),
    ]))
}

fn selections_from_batches(batches: &[RecordBatch]) -> Result<Vec<DiffCommandSelection>> {
    let mut selections = Vec::new();
    for batch in batches {
        let row_refs = batch
            .column_by_name("row_ref")
            .ok_or_else(|| DataFusionError::Execution("row_ref column is required".to_string()))?
            .as_any()
            .downcast_ref::<StringArray>()
            .ok_or_else(|| DataFusionError::Execution("row_ref must be lix_row_ref".to_string()))?;
        for index in 0..row_refs.len() {
            if row_refs.is_null(index) {
                return Err(DataFusionError::Execution(
                    "row_ref cannot be NULL".to_string(),
                ));
            }
            let resolved = crate::row_ref::decode_str(row_refs.value(index))
                .map_err(lix_error_to_datafusion_error)?;
            selections.push(DiffCommandSelection {
                relation: resolved.relation,
                row_pk: resolved.row_pk,
                source_commits: None,
            });
        }
    }
    selections.sort_by(|left, right| {
        (&left.relation, &left.row_pk).cmp(&(&right.relation, &right.row_pk))
    });
    if selections
        .windows(2)
        .any(|pair| pair[0].relation == pair[1].relation && pair[0].row_pk == pair[1].row_pk)
    {
        return Err(DataFusionError::Execution(
            "diff command selection contains duplicate row_ref values".to_string(),
        ));
    }
    Ok(selections)
}