use std::collections::BTreeMap;
use std::sync::Arc;
use datafusion::arrow::datatypes::SchemaRef;
use datafusion::common::ScalarValue;
use datafusion::logical_expr::expr::InList;
use datafusion::logical_expr::{Expr, Operator};
use tokio::sync::Mutex;
use crate::LixError;
use crate::changelog::CommitId;
use crate::commit_graph::{CommitGraphChangeHistoryRequest, CommitGraphReader};
use crate::row_pk::RowPk;
use super::SqlHistoryQuerySource;
use crate::sql2::change_materialization::{
MaterializedChange, materialize_located_history_change,
};
use crate::storage_adapter::StorageAdapterRead;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct HistoryRoute {
pub(crate) as_of_commit_ids: Vec<String>,
pub(crate) row_pks: Vec<String>,
pub(crate) resolved_row_pks: Vec<RowPk>,
pub(crate) schema_keys: Vec<String>,
pub(crate) file_ids: Vec<String>,
pub(crate) min_depth: Option<i64>,
pub(crate) max_depth: Option<i64>,
pub(crate) invalid_as_of_commit_filter: bool,
pub(crate) contradictory: bool,
}
impl HistoryRoute {
pub(crate) fn from_filters(filters: &[Expr]) -> Self {
let mut route = Self::default();
for filter in filters {
route.invalid_as_of_commit_filter |= !history_anchor_filter_is_exact(filter);
apply_history_filter(filter, &mut route);
}
route
}
pub(crate) fn default_to_as_of_commit_id(&mut self, commit_id: &str) {
if self.as_of_commit_ids.is_empty() && !self.invalid_as_of_commit_filter {
self.as_of_commit_ids.push(commit_id.to_string());
}
}
pub(crate) fn traversal_only(&self) -> Self {
Self {
as_of_commit_ids: self.as_of_commit_ids.clone(),
min_depth: self.min_depth,
max_depth: self.max_depth,
invalid_as_of_commit_filter: self.invalid_as_of_commit_filter,
contradictory: self.contradictory,
..Self::default()
}
}
pub(crate) fn anchors_only(&self) -> Self {
Self {
as_of_commit_ids: self.as_of_commit_ids.clone(),
invalid_as_of_commit_filter: self.invalid_as_of_commit_filter,
contradictory: self.contradictory,
..Self::default()
}
}
pub(crate) fn is_contradictory(&self) -> bool {
self.contradictory
|| self
.min_depth
.zip(self.max_depth)
.is_some_and(|(min, max)| min > max)
|| self.min_depth.is_some_and(|depth| depth < 0)
|| self.max_depth.is_some_and(|depth| depth < 0)
}
pub(crate) fn constrain_row_pks(&mut self, row_pks: Vec<String>) {
self.contradictory |= apply_conjunctive_values_filter(&mut self.row_pks, row_pks);
}
pub(crate) fn set_resolved_row_pks(&mut self, row_pks: Vec<RowPk>) {
self.resolved_row_pks = row_pks;
}
pub(crate) fn matches_surface_row(
&self,
schema_key: &str,
row_pk: &str,
file_id: Option<&str>,
depth: u32,
) -> bool {
if self.is_contradictory() {
return false;
}
if !self.schema_keys.is_empty()
&& !self
.schema_keys
.iter()
.any(|candidate| candidate == schema_key)
{
return false;
}
if !self.row_pks.is_empty() && !self.row_pks.iter().any(|candidate| candidate == row_pk) {
return false;
}
if !self.file_ids.is_empty() {
let Some(file_id) = file_id else {
return false;
};
if !self.file_ids.iter().any(|candidate| candidate == file_id) {
return false;
}
}
if self
.min_depth
.is_some_and(|min_depth| i64::from(depth) < min_depth)
{
return false;
}
if self
.max_depth
.is_some_and(|max_depth| i64::from(depth) > max_depth)
{
return false;
}
true
}
}
#[derive(Debug, Clone)]
pub(crate) struct HistoryEntry {
pub(crate) change: MaterializedChange,
pub(crate) observed_commit_id: String,
pub(crate) commit_created_at: Option<String>,
pub(crate) as_of_commit_id: String,
pub(crate) depth: u32,
}
pub(crate) const HISTORY_COL_ROW_PK: &str = "lixcol_row_ref";
pub(crate) const HISTORY_COL_SCHEMA_KEY: &str = "lixcol_schema_key";
pub(crate) const HISTORY_COL_FILE_ID: &str = "lixcol_file_id";
pub(crate) const HISTORY_COL_METADATA: &str = "lixcol_metadata";
pub(crate) const HISTORY_COL_CHANGE_ID: &str = "lixcol_change_id";
pub(crate) const HISTORY_COL_CHANGE_CREATED_AT: &str = "lixcol_change_created_at";
pub(crate) const HISTORY_COL_SOURCE_CHANGES: &str = "lixcol_source_changes";
pub(crate) const HISTORY_COL_ORIGIN_KEY: &str = "lixcol_origin_key";
pub(crate) const HISTORY_COL_OBSERVED_COMMIT_ID: &str = "lixcol_observed_commit_id";
pub(crate) const HISTORY_COL_COMMIT_CREATED_AT: &str = "lixcol_commit_created_at";
pub(crate) const HISTORY_COL_AS_OF_COMMIT_ID: &str = "lixcol_as_of_commit_id";
pub(crate) const HISTORY_COL_DEPTH: &str = "lixcol_depth";
pub(crate) const HISTORY_COL_IS_DELETED: &str = "lixcol_is_deleted";
pub(crate) fn serialize_history_source_changes(
changes: &[MaterializedChange],
surface_name: &str,
) -> Result<String, LixError> {
let mut ordered_changes = changes.iter().collect::<Vec<_>>();
ordered_changes.sort_by(|left, right| left.id.cmp(&right.id));
let source_changes = ordered_changes
.into_iter()
.map(|change| {
let row_pk = change.row_pk.as_json_array_value()?;
let snapshot_content = parse_optional_source_json(
change.snapshot_content.as_deref(),
surface_name,
"snapshot_content",
)?;
let metadata =
parse_optional_source_json(change.metadata.as_deref(), surface_name, "metadata")?;
Ok(serde_json::json!({
"id": change.id,
"row_pk": row_pk,
"schema_key": change.schema_key,
"file_id": change.file_id,
"snapshot_content": snapshot_content,
"metadata": metadata,
"created_at": change.created_at,
"origin_key": change.origin_key,
}))
})
.collect::<Result<Vec<_>, LixError>>()?;
serde_json::to_string(&source_changes).map_err(|error| {
LixError::new(
LixError::CODE_INTERNAL_ERROR,
format!("failed to serialize {surface_name} source changes: {error}"),
)
})
}
fn parse_optional_source_json(
value: Option<&str>,
surface_name: &str,
field: &str,
) -> Result<Option<serde_json::Value>, LixError> {
value
.map(|value| {
serde_json::from_str(value).map_err(|error| {
LixError::new(
LixError::CODE_INTERNAL_ERROR,
format!("{surface_name} source {field} is invalid JSON: {error}"),
)
})
})
.transpose()
}
pub(crate) struct HistoryViewDescriptor<'a> {
pub(crate) view_name: &'a str,
pub(crate) as_of_commit_column: &'a str,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct HistoryMetadataProjection {
commit_created_at: bool,
}
impl HistoryMetadataProjection {
pub(crate) fn from_scan(projected_schema: &SchemaRef, filters: &[Expr]) -> Self {
let column_name = HISTORY_COL_COMMIT_CREATED_AT;
let commit_created_at = projected_schema.field_with_name(column_name).is_ok()
|| filters.iter().any(|filter| {
filter
.column_refs()
.iter()
.any(|column| column.name == column_name)
});
Self { commit_created_at }
}
#[cfg(test)]
fn commit_created_at(self) -> bool {
self.commit_created_at
}
}
pub(crate) fn parse_history_filter(expr: &Expr) -> Option<()> {
parse_history_filter_terms(expr).map(|_| ())
}
pub(crate) fn history_filter_references_row_ref(expr: &Expr) -> bool {
expr.column_refs().iter().any(|column| column.name == HISTORY_COL_ROW_PK)
}
pub(crate) fn validate_history_anchor_filter(expr: &Expr) -> Result<(), LixError> {
if history_anchor_filter_is_exact(expr) {
return Ok(());
}
Err(invalid_history_anchor_error(
HISTORY_COL_AS_OF_COMMIT_ID,
None,
))
}
pub(crate) fn commit_graph_history_request(
route: &HistoryRoute,
schema_keys: Vec<String>,
limit: Option<usize>,
) -> Option<CommitGraphChangeHistoryRequest> {
let schema_keys = effective_schema_keys(route, schema_keys)?;
Some(CommitGraphChangeHistoryRequest {
limit,
row_pks: if route.resolved_row_pks.is_empty() {
route
.row_pks
.iter()
.filter_map(|row_pk| RowPk::from_json_array_text(row_pk).ok())
.collect()
} else {
route.resolved_row_pks.clone()
},
schema_keys,
file_ids: route.file_ids.clone(),
min_depth: route.min_depth.and_then(nonnegative_u32),
max_depth: route.max_depth.and_then(nonnegative_u32),
include_tombstones: true,
})
}
pub(crate) async fn load_history_entries<S>(
descriptor: HistoryViewDescriptor<'_>,
commit_graph: Arc<Mutex<Box<dyn CommitGraphReader>>>,
query_source: SqlHistoryQuerySource<S>,
route: &HistoryRoute,
schema_keys: Vec<String>,
metadata_projection: HistoryMetadataProjection,
limit: Option<usize>,
) -> Result<Vec<HistoryEntry>, LixError>
where
S: StorageAdapterRead + Clone + Send + Sync + 'static,
{
if route.invalid_as_of_commit_filter {
return Err(invalid_history_anchor_error(
descriptor.as_of_commit_column,
Some(descriptor.view_name),
));
}
if route.is_contradictory() {
return Ok(Vec::new());
}
let Some(request) = commit_graph_history_request(route, schema_keys, limit) else {
return Ok(Vec::new());
};
let as_of_commit_ids = if route.as_of_commit_ids.is_empty() {
std::slice::from_ref(&query_source.default_as_of_commit_id)
} else {
route.as_of_commit_ids.as_slice()
};
let mut rows = Vec::new();
for as_of_commit_id in as_of_commit_ids {
if limit.is_some_and(|limit| rows.len() >= limit) {
break;
}
let as_of_commit_id =
CommitId::parse_lix(as_of_commit_id, "history lixcol_as_of_commit_id")?;
let (entries, reachable_nodes) = {
let mut guard = commit_graph.lock().await;
let history = guard
.change_history_from_commit(&as_of_commit_id, &request)
.await?;
let reachable_nodes = if metadata_projection.commit_created_at {
history.reachable_nodes
} else {
Arc::from([])
};
(history.entries, reachable_nodes)
};
let reachable_by_id = reachable_nodes
.iter()
.map(|reachable| {
(
reachable.commit.commit_id,
(reachable.depth, reachable.commit.created_at.to_string()),
)
})
.collect::<BTreeMap<_, _>>();
for entry in entries {
let change = materialize_located_history_change(entry.change)?;
let commit_created_at = if metadata_projection.commit_created_at {
Some(
reachable_by_id
.get(&entry.observed_commit_id)
.map(|(_, created_at)| created_at)
.cloned()
.ok_or_else(|| {
LixError::new(
LixError::CODE_INTERNAL_ERROR,
format!(
"history commit '{}' is missing its commit timestamp",
entry.observed_commit_id
),
)
})?,
)
} else {
None
};
rows.push(HistoryEntry {
commit_created_at,
change,
observed_commit_id: entry.observed_commit_id.to_string(),
as_of_commit_id: entry.start_commit_id.to_string(),
depth: entry.depth,
});
}
}
Ok(rows)
}
pub(crate) fn invalid_history_anchor_error(
as_of_commit_column: &str,
view_name: Option<&str>,
) -> LixError {
let surface = view_name.map_or_else(String::new, |view_name| format!("{view_name}: "));
LixError::new(
LixError::CODE_UNSUPPORTED_SQL,
format!(
"{surface}history anchor '{as_of_commit_column}' only supports exact equality or non-empty IN predicates that resolve directly to a history scan"
),
)
.with_hint(format!(
"Omit {as_of_commit_column} to use the pinned active branch head, or use WHERE {as_of_commit_column} = $1 (or {as_of_commit_column} IN ($1, $2)) for time travel."
))
}
fn effective_schema_keys(
route: &HistoryRoute,
surface_schema_keys: Vec<String>,
) -> Option<Vec<String>> {
if surface_schema_keys.is_empty() {
return Some(route.schema_keys.clone());
}
if route.schema_keys.is_empty() {
return Some(surface_schema_keys);
}
let mut effective = Vec::new();
for schema_key in surface_schema_keys {
if route.schema_keys.contains(&schema_key) && !effective.contains(&schema_key) {
effective.push(schema_key);
}
}
if effective.is_empty() {
None
} else {
Some(effective)
}
}
fn parse_history_filter_terms(expr: &Expr) -> Option<Vec<HistoryFilterTerm>> {
match expr {
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::And => {
let mut terms = parse_history_filter_terms(&binary_expr.left)?;
terms.extend(parse_history_filter_terms(&binary_expr.right)?);
Some(terms)
}
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::Or => {
parse_history_disjunction(binary_expr)
}
Expr::BinaryExpr(binary_expr) => {
parse_history_binary_filter(binary_expr).map(|term| vec![term])
}
Expr::InList(in_list) => parse_history_in_list_filter(in_list).map(|term| vec![term]),
_ => None,
}
}
fn collect_history_route_terms(expr: &Expr) -> Vec<HistoryFilterTerm> {
match expr {
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::And => {
let mut terms = collect_history_route_terms(&binary_expr.left);
terms.extend(collect_history_route_terms(&binary_expr.right));
terms
}
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::Or => {
parse_history_disjunction(binary_expr).unwrap_or_default()
}
Expr::BinaryExpr(binary_expr) => parse_history_binary_filter(binary_expr)
.map(|term| vec![term])
.unwrap_or_default(),
Expr::InList(in_list) => parse_history_in_list_filter(in_list)
.map(|term| vec![term])
.unwrap_or_default(),
_ => Vec::new(),
}
}
fn parse_history_disjunction(
binary_expr: &datafusion::logical_expr::BinaryExpr,
) -> Option<Vec<HistoryFilterTerm>> {
let left = parse_history_filter_terms(&binary_expr.left)?;
let right = parse_history_filter_terms(&binary_expr.right)?;
let [left] = left.as_slice() else {
return None;
};
let [right] = right.as_slice() else {
return None;
};
merge_history_disjunction_terms(left.clone(), right.clone()).map(|term| vec![term])
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum HistoryFilterTerm {
AsOfCommitIds(Vec<String>),
RowRefs(Vec<RoutedRowRef>),
SchemaKeys(Vec<String>),
FileIds(Vec<String>),
MinDepth(i64),
MaxDepth(i64),
ExactDepth(i64),
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct RoutedRowRef {
row_pk_json: String,
row_pk: RowPk,
}
fn merge_history_disjunction_terms(
left: HistoryFilterTerm,
right: HistoryFilterTerm,
) -> Option<HistoryFilterTerm> {
match (left, right) {
(HistoryFilterTerm::AsOfCommitIds(mut left), HistoryFilterTerm::AsOfCommitIds(right)) => {
extend_unique(&mut left, right);
Some(HistoryFilterTerm::AsOfCommitIds(left))
}
(HistoryFilterTerm::RowRefs(mut left), HistoryFilterTerm::RowRefs(right)) => {
extend_unique(&mut left, right);
Some(HistoryFilterTerm::RowRefs(left))
}
(HistoryFilterTerm::FileIds(mut left), HistoryFilterTerm::FileIds(right)) => {
extend_unique(&mut left, right);
Some(HistoryFilterTerm::FileIds(left))
}
(HistoryFilterTerm::SchemaKeys(mut left), HistoryFilterTerm::SchemaKeys(right)) => {
extend_unique(&mut left, right);
Some(HistoryFilterTerm::SchemaKeys(left))
}
_ => None,
}
}
fn parse_history_binary_filter(
binary_expr: &datafusion::logical_expr::BinaryExpr,
) -> Option<HistoryFilterTerm> {
let (column, right) = match (&*binary_expr.left, &binary_expr.op, &*binary_expr.right) {
(Expr::Column(column), _, right) => (column, right),
(left, Operator::Eq, Expr::Column(column)) => (column, left),
_ => return None,
};
let column_name = canonical_history_column_name(column.name.as_str())?;
match (column_name, &binary_expr.op, right) {
(
"as_of_commit_id" | "schema_key" | "file_id",
Operator::Eq,
Expr::Literal(ScalarValue::Utf8(Some(value)), _),
) => Some(match column_name {
"as_of_commit_id" => HistoryFilterTerm::AsOfCommitIds(vec![value.clone()]),
"schema_key" => HistoryFilterTerm::SchemaKeys(vec![value.clone()]),
"file_id" => HistoryFilterTerm::FileIds(vec![value.clone()]),
_ => unreachable!(),
}),
("row_ref", Operator::Eq, Expr::Literal(ScalarValue::Utf8(Some(value)), _)) => {
routed_row_ref(value).map(|value| HistoryFilterTerm::RowRefs(vec![value]))
}
("depth", Operator::Eq, depth_expr) => {
scalar_i64_literal(depth_expr).map(HistoryFilterTerm::ExactDepth)
}
("depth", Operator::Gt, depth_expr) => {
scalar_i64_literal(depth_expr).map(|value| HistoryFilterTerm::MinDepth(value + 1))
}
("depth", Operator::GtEq, depth_expr) => {
scalar_i64_literal(depth_expr).map(HistoryFilterTerm::MinDepth)
}
("depth", Operator::Lt, depth_expr) => {
scalar_i64_literal(depth_expr).map(|value| HistoryFilterTerm::MaxDepth(value - 1))
}
("depth", Operator::LtEq, depth_expr) => {
scalar_i64_literal(depth_expr).map(HistoryFilterTerm::MaxDepth)
}
_ => None,
}
}
fn history_anchor_filter_is_exact(expr: &Expr) -> bool {
if !history_filter_references_anchor(expr) {
return true;
}
match expr {
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::And => {
history_anchor_filter_is_exact(&binary_expr.left)
&& history_anchor_filter_is_exact(&binary_expr.right)
}
Expr::BinaryExpr(binary_expr) if binary_expr.op == Operator::Or => matches!(
parse_history_disjunction(binary_expr).as_deref(),
Some([HistoryFilterTerm::AsOfCommitIds(_)])
),
Expr::BinaryExpr(binary_expr) => matches!(
parse_history_binary_filter(binary_expr),
Some(HistoryFilterTerm::AsOfCommitIds(_))
),
Expr::InList(in_list) => matches!(
parse_history_in_list_filter(in_list),
Some(HistoryFilterTerm::AsOfCommitIds(_))
),
_ => false,
}
}
fn history_filter_references_anchor(expr: &Expr) -> bool {
expr.column_refs().iter().any(|column| {
canonical_history_column_name(column.name.as_str()) == Some("as_of_commit_id")
})
}
fn parse_history_in_list_filter(in_list: &InList) -> Option<HistoryFilterTerm> {
if in_list.negated {
return None;
}
let Expr::Column(column) = in_list.expr.as_ref() else {
return None;
};
let column_name = canonical_history_column_name(column.name.as_str())?;
let values = in_list
.list
.iter()
.map(string_literal)
.collect::<Option<Vec<_>>>()?;
if values.is_empty() {
return None;
}
match column_name {
"as_of_commit_id" => Some(HistoryFilterTerm::AsOfCommitIds(values)),
"row_ref" => routed_row_refs(values).map(HistoryFilterTerm::RowRefs),
"schema_key" => Some(HistoryFilterTerm::SchemaKeys(values)),
"file_id" => Some(HistoryFilterTerm::FileIds(values)),
_ => None,
}
}
fn apply_history_filter(expr: &Expr, route: &mut HistoryRoute) {
for term in collect_history_route_terms(expr) {
match term {
HistoryFilterTerm::AsOfCommitIds(values) => {
route.contradictory |=
apply_conjunctive_values_filter(&mut route.as_of_commit_ids, values);
}
HistoryFilterTerm::RowRefs(values) => {
let row_pk_json = values
.iter()
.map(|value| value.row_pk_json.clone())
.collect();
route.contradictory |=
apply_conjunctive_values_filter(&mut route.row_pks, row_pk_json);
let typed_row_pks = values
.into_iter()
.filter(|value| route.row_pks.contains(&value.row_pk_json))
.map(|value| value.row_pk)
.collect::<Vec<_>>();
if route.resolved_row_pks.is_empty() {
route.resolved_row_pks = typed_row_pks;
} else {
route
.resolved_row_pks
.retain(|row_pk| typed_row_pks.contains(row_pk));
}
}
HistoryFilterTerm::SchemaKeys(values) => {
route.contradictory |=
apply_conjunctive_values_filter(&mut route.schema_keys, values);
}
HistoryFilterTerm::FileIds(values) => {
route.contradictory |= apply_conjunctive_values_filter(&mut route.file_ids, values);
}
HistoryFilterTerm::ExactDepth(value) => {
route.min_depth = Some(value);
route.max_depth = Some(value);
}
HistoryFilterTerm::MinDepth(value) => {
route.min_depth = Some(route.min_depth.map_or(value, |current| current.max(value)));
}
HistoryFilterTerm::MaxDepth(value) => {
route.max_depth = Some(route.max_depth.map_or(value, |current| current.min(value)));
}
}
}
}
fn apply_conjunctive_values_filter(bucket: &mut Vec<String>, incoming_values: Vec<String>) -> bool {
let mut values = Vec::new();
extend_unique(&mut values, incoming_values);
if values.is_empty() {
return true;
}
if bucket.is_empty() {
extend_unique(bucket, values);
return false;
}
bucket.retain(|existing| values.contains(existing));
bucket.is_empty()
}
fn routed_row_refs(values: Vec<String>) -> Option<Vec<RoutedRowRef>> {
values
.into_iter()
.map(|value| routed_row_ref(&value))
.collect()
}
fn routed_row_ref(value: &str) -> Option<RoutedRowRef> {
let row_pk = crate::row_ref::decode_str(value).ok()?.row_pk;
Some(RoutedRowRef {
row_pk_json: row_pk.as_json_array_text().ok()?,
row_pk,
})
}
fn canonical_history_column_name(name: &str) -> Option<&str> {
match name {
HISTORY_COL_AS_OF_COMMIT_ID => Some("as_of_commit_id"),
HISTORY_COL_SCHEMA_KEY => Some("schema_key"),
HISTORY_COL_FILE_ID => Some("file_id"),
HISTORY_COL_ROW_PK => Some("row_ref"),
HISTORY_COL_DEPTH => Some("depth"),
_ => None,
}
}
fn nonnegative_u32(value: i64) -> Option<u32> {
u32::try_from(value).ok()
}
fn extend_unique<T: PartialEq>(bucket: &mut Vec<T>, values: Vec<T>) {
for value in values {
if !bucket.contains(&value) {
bucket.push(value);
}
}
}
fn string_literal(expr: &Expr) -> Option<String> {
match expr {
Expr::Literal(ScalarValue::Utf8(Some(value)), _) => Some(value.clone()),
_ => None,
}
}
fn scalar_i64_literal(expr: &Expr) -> Option<i64> {
match expr {
Expr::Literal(ScalarValue::Int8(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::Int16(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::Int32(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::Int64(Some(value)), _) => Some(*value),
Expr::Literal(ScalarValue::UInt8(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::UInt16(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::UInt32(Some(value)), _) => Some(i64::from(*value)),
Expr::Literal(ScalarValue::UInt64(Some(value)), _) => i64::try_from(*value).ok(),
_ => None,
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::common::{Column, ScalarValue};
use datafusion::logical_expr::{BinaryExpr, Expr, Like, Operator};
use tokio::sync::Mutex;
use crate::LixError;
use crate::changelog::{ChangeId, CommitId};
use crate::commit_graph::{
CommitGraphChange, CommitGraphChangeHistoryEntry, CommitGraphChangeHistoryRequest,
CommitGraphNode, CommitGraphReader, ReachableCommitGraphNode,
};
use crate::row_pk::RowPk;
use crate::sql2::change_materialization::MaterializedChange;
use crate::sql2::HistoryQuerySource;
use crate::storage_adapter::{
Memory, MemoryRead, SharedStorageAdapterRead, StorageAdapter, StorageReadOptions,
};
use super::{
HISTORY_COL_AS_OF_COMMIT_ID, HISTORY_COL_COMMIT_CREATED_AT, HISTORY_COL_DEPTH,
HISTORY_COL_ROW_PK, HistoryMetadataProjection, HistoryRoute, HistoryViewDescriptor,
commit_graph_history_request, load_history_entries, parse_history_filter,
serialize_history_source_changes,
};
#[test]
fn row_ref_route_preserves_uuid_primary_key_type() {
let row_pk = RowPk::uuid_from_canonical("018f6f7e-7cb2-7d45-8e1f-0a2b3c4d5e6f")
.expect("test UUID should be canonical");
let row_ref = crate::row_ref::encode("lix_file", &row_pk)
.expect("row reference should encode");
let route = HistoryRoute::from_filters(&[eq(
col(HISTORY_COL_ROW_PK),
str_lit(row_ref.as_str()),
)]);
assert_eq!(route.resolved_row_pks, vec![row_pk.clone()]);
let request = commit_graph_history_request(&route, vec![], None)
.expect("history request should route");
assert_eq!(request.row_pks, vec![row_pk]);
}
#[test]
fn source_change_identity_matches_underlying_schema() {
let file_id = "018f6f7e-7cb2-7d45-8e1f-0a2b3c4d5e6f";
let row_pk = RowPk::uuid_from_canonical(file_id).expect("test UUID should be canonical");
let change = MaterializedChange {
id: "change-1".to_owned(),
account_id: crate::ANONYMOUS_ACCOUNT_ID.to_owned(),
row_pk: row_pk.clone(),
schema_key: "lix_file_descriptor".to_owned(),
file_id: Some(file_id.to_owned()),
snapshot_content: None,
metadata: None,
decoded_snapshot: None,
created_at: "2026-08-27T00:00:00Z".to_owned(),
origin_key: None,
};
let json = serialize_history_source_changes(&[change], "test history")
.expect("source changes should serialize");
let value: serde_json::Value =
serde_json::from_str(&json).expect("source changes should be JSON");
assert_eq!(value[0]["schema_key"], "lix_file_descriptor");
assert_eq!(value[0]["row_pk"], row_pk.as_json_array_value().unwrap());
assert_eq!(value[0]["file_id"], file_id);
assert!(value[0].get("row_ref").is_none());
}
#[test]
fn route_extraction_keeps_supported_terms_from_mixed_and_filter() {
let filter = and(
eq(col(HISTORY_COL_AS_OF_COMMIT_ID), str_lit("commit-1")),
Expr::Like(Like::new(
false,
Box::new(col("path")),
Box::new(str_lit("/docs/%")),
None,
false,
)),
);
assert!(
parse_history_filter(&filter).is_none(),
"mixed filters must not be advertised as exact pushdown"
);
let route = HistoryRoute::from_filters(&[filter]);
assert_eq!(route.as_of_commit_ids, vec!["commit-1".to_string()]);
}
#[test]
fn route_extraction_does_not_partially_route_mixed_or_filter() {
let filter = or(
eq(col(HISTORY_COL_AS_OF_COMMIT_ID), str_lit("commit-1")),
Expr::Like(Like::new(
false,
Box::new(col("path")),
Box::new(str_lit("/docs/%")),
None,
false,
)),
);
let route = HistoryRoute::from_filters(&[filter]);
assert!(
route.as_of_commit_ids.is_empty(),
"partial OR pushdown would change SQL semantics"
);
}
#[test]
fn routing_rejects_retired_history_column_names() {
for retired in [
"start_commit_id",
"lixcol_start_commit_id",
"row_pk",
"depth",
] {
let filter = eq(col(retired), str_lit("value"));
assert!(
parse_history_filter(&filter).is_none(),
"retired column '{retired}' must not route"
);
assert!(
HistoryRoute::from_filters(&[filter])
.as_of_commit_ids
.is_empty()
);
}
}
#[test]
fn commit_metadata_projection_tracks_projection_and_filters() {
let unrelated_schema = Arc::new(Schema::new(vec![Field::new(
HISTORY_COL_DEPTH,
DataType::Int64,
false,
)]));
assert!(!HistoryMetadataProjection::from_scan(&unrelated_schema, &[]).commit_created_at());
let projected_schema = Arc::new(Schema::new(vec![Field::new(
HISTORY_COL_COMMIT_CREATED_AT,
DataType::Utf8,
false,
)]));
assert!(HistoryMetadataProjection::from_scan(&projected_schema, &[]).commit_created_at());
let residual_filter = eq(
col(HISTORY_COL_COMMIT_CREATED_AT),
str_lit("2026-07-12T00:00:00Z"),
);
assert!(
HistoryMetadataProjection::from_scan(&unrelated_schema, &[residual_filter])
.commit_created_at()
);
}
#[tokio::test]
async fn history_loader_defaults_to_pinned_head_without_metadata_walk() {
let reachable_calls = Arc::new(AtomicUsize::new(0));
let start_commit_id = CommitId::for_test_label("start");
let rows = load_history_entries(
HistoryViewDescriptor {
view_name: "test_history",
as_of_commit_column: HISTORY_COL_AS_OF_COMMIT_ID,
},
test_commit_graph(Arc::clone(&reachable_calls), start_commit_id),
empty_history_query_source(start_commit_id).await,
&HistoryRoute::default(),
vec!["message".to_string()],
HistoryMetadataProjection::default(),
None,
)
.await
.expect("history load should succeed without commit metadata");
assert_eq!(reachable_calls.load(Ordering::SeqCst), 0);
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].commit_created_at, None);
}
#[tokio::test]
async fn history_loader_reuses_history_topology_for_projected_commit_timestamp() {
let reachable_calls = Arc::new(AtomicUsize::new(0));
let start_commit_id = CommitId::for_test_label("start");
let metadata_schema = Arc::new(Schema::new(vec![Field::new(
HISTORY_COL_COMMIT_CREATED_AT,
DataType::Utf8,
false,
)]));
let rows = load_history_entries(
HistoryViewDescriptor {
view_name: "test_history",
as_of_commit_column: HISTORY_COL_AS_OF_COMMIT_ID,
},
test_commit_graph(Arc::clone(&reachable_calls), start_commit_id),
empty_history_query_source(start_commit_id).await,
&HistoryRoute {
as_of_commit_ids: vec![start_commit_id.to_string()],
..HistoryRoute::default()
},
vec!["message".to_string()],
HistoryMetadataProjection::from_scan(&metadata_schema, &[]),
None,
)
.await
.expect("history load should enrich commit metadata");
assert_eq!(
reachable_calls.load(Ordering::SeqCst),
0,
"commit metadata must not trigger a second topology walk",
);
assert_eq!(rows.len(), 1);
assert_eq!(
rows[0].commit_created_at,
Some(commit_timestamp().to_string())
);
assert_eq!(rows[0].change.created_at, event_timestamp().to_string());
}
#[tokio::test]
async fn history_loader_does_not_substitute_change_time_for_missing_commit_time() {
let reachable_calls = Arc::new(AtomicUsize::new(0));
let as_of_commit_id = CommitId::for_test_label("start");
let metadata_schema = Arc::new(Schema::new(vec![Field::new(
HISTORY_COL_COMMIT_CREATED_AT,
DataType::Utf8,
false,
)]));
let error = load_history_entries(
HistoryViewDescriptor {
view_name: "test_history",
as_of_commit_column: HISTORY_COL_AS_OF_COMMIT_ID,
},
Arc::new(Mutex::new(Box::new(CountingCommitGraphReader {
reachable_calls,
start_commit_id: as_of_commit_id,
include_reachable_commit: false,
}))),
empty_history_query_source(as_of_commit_id).await,
&HistoryRoute {
as_of_commit_ids: vec![as_of_commit_id.to_string()],
..HistoryRoute::default()
},
vec!["message".to_string()],
HistoryMetadataProjection::from_scan(&metadata_schema, &[]),
None,
)
.await
.expect_err("missing commit metadata must be an explicit error");
assert_eq!(error.code, LixError::CODE_INTERNAL_ERROR);
assert!(error.message.contains("missing its commit timestamp"));
}
struct CountingCommitGraphReader {
reachable_calls: Arc<AtomicUsize>,
start_commit_id: CommitId,
include_reachable_commit: bool,
}
#[async_trait::async_trait]
impl CommitGraphReader for CountingCommitGraphReader {
async fn load_node(
&mut self,
_commit_id: &CommitId,
) -> Result<Option<CommitGraphNode>, LixError> {
Ok(None)
}
async fn reachable_nodes(
&mut self,
_head_commit_id: &CommitId,
) -> Result<Arc<[ReachableCommitGraphNode]>, LixError> {
self.reachable_calls.fetch_add(1, Ordering::SeqCst);
if !self.include_reachable_commit {
return Ok(Arc::from([]));
}
Ok(Arc::from([ReachableCommitGraphNode {
commit: CommitGraphNode {
touched_scope_digest: crate::changelog::CommitTouchedScopeDigest::absent(),
commit_id: self.start_commit_id,
change_id: ChangeId::for_test_label("commit-change"),
account_id: crate::ANONYMOUS_ACCOUNT_ID.to_string(),
generation: 0,
parent_commit_ids: Vec::new(),
base_commit_id: None,
first_parent_jump_commit_id: self.start_commit_id,
first_parent_jump_span: 0,
created_at: commit_timestamp(),
},
depth: 0,
}]))
}
async fn change_history_from_commit(
&mut self,
_start_commit_id: &CommitId,
_request: &CommitGraphChangeHistoryRequest,
) -> Result<crate::commit_graph::CommitGraphHistory, LixError> {
let reachable_nodes = self
.include_reachable_commit
.then(|| ReachableCommitGraphNode {
commit: CommitGraphNode {
touched_scope_digest: crate::changelog::CommitTouchedScopeDigest::absent(),
commit_id: self.start_commit_id,
change_id: ChangeId::for_test_label("commit-change"),
account_id: crate::ANONYMOUS_ACCOUNT_ID.to_string(),
generation: 0,
parent_commit_ids: Vec::new(),
base_commit_id: None,
first_parent_jump_commit_id: self.start_commit_id,
first_parent_jump_span: 0,
created_at: commit_timestamp(),
},
depth: 0,
});
Ok(crate::commit_graph::CommitGraphHistory {
entries: vec![CommitGraphChangeHistoryEntry {
change: test_change("row-change", event_timestamp()),
observed_commit_id: self.start_commit_id,
start_commit_id: self.start_commit_id,
depth: 0,
}],
reachable_nodes: reachable_nodes.into_iter().collect::<Vec<_>>().into(),
})
}
}
fn test_commit_graph(
reachable_calls: Arc<AtomicUsize>,
start_commit_id: CommitId,
) -> Arc<Mutex<Box<dyn CommitGraphReader>>> {
Arc::new(Mutex::new(Box::new(CountingCommitGraphReader {
reachable_calls,
start_commit_id,
include_reachable_commit: true,
})))
}
fn test_change(label: &str, created_at: crate::common::LixTimestamp) -> CommitGraphChange {
CommitGraphChange {
id: ChangeId::for_test_label(label),
account_id: crate::ANONYMOUS_ACCOUNT_ID.to_string(),
row_pk: RowPk::single("row-1"),
schema_key: "message".to_string(),
file_id: None,
metadata: None,
snapshot: None,
created_at,
origin_key: None,
}
}
fn event_timestamp() -> crate::common::LixTimestamp {
crate::common::LixTimestamp::expect_parse("event timestamp", "2026-07-11T00:00:00Z")
}
fn commit_timestamp() -> crate::common::LixTimestamp {
crate::common::LixTimestamp::expect_parse("commit timestamp", "2026-07-12T00:00:00Z")
}
async fn empty_history_query_source(
default_as_of_commit_id: CommitId,
) -> HistoryQuerySource<SharedStorageAdapterRead<MemoryRead>> {
let storage = StorageAdapter::new(Memory::new());
let read_scope = storage
.begin_read(StorageReadOptions::default())
.await
.expect("read should open");
let read_scope = SharedStorageAdapterRead::new(read_scope);
HistoryQuerySource {
store: read_scope,
default_as_of_commit_id: default_as_of_commit_id.to_string(),
}
}
fn and(left: Expr, right: Expr) -> Expr {
binary(left, Operator::And, right)
}
fn or(left: Expr, right: Expr) -> Expr {
binary(left, Operator::Or, right)
}
fn eq(left: Expr, right: Expr) -> Expr {
binary(left, Operator::Eq, right)
}
fn binary(left: Expr, op: Operator, right: Expr) -> Expr {
Expr::BinaryExpr(BinaryExpr::new(Box::new(left), op, Box::new(right)))
}
fn col(name: &str) -> Expr {
Expr::Column(Column::from_name(name))
}
fn str_lit(value: &str) -> Expr {
Expr::Literal(ScalarValue::Utf8(Some(value.to_string())), None)
}
}