#[cfg(test)]
mod tests {
use crate::session::CreateBranchOptions;
use crate::sql2::test_support::generators::{
DifferentialExpectation, DifferentialParam, DifferentialProbe, DifferentialSqlCase,
ExpectedExecution, deterministic_repro_cases, generated_dml_cases,
};
use crate::sql2::{WriteExecutorMode, WriteExecutorPath};
use crate::storage_adapter::Memory;
use crate::{ExecuteResult, LixError, Value, engine::Engine};
#[derive(Debug, Clone)]
struct DifferentialOutcome {
execution: ExecutionSignature,
executor_path: Option<WriteExecutorPath>,
staged_rows: Vec<ProbeSnapshot>,
final_rows: Vec<ProbeSnapshot>,
}
impl PartialEq for DifferentialOutcome {
fn eq(&self, other: &Self) -> bool {
self.execution == other.execution
&& self.staged_rows == other.staged_rows
&& self.final_rows == other.final_rows
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum ExecutionSignature {
Ok { rows_affected: u64 },
Err { code: String, message: String },
}
#[derive(Debug, Clone, PartialEq)]
struct ProbeSnapshot {
name: String,
rows: Vec<Vec<Value>>,
}
struct ProbeQuery {
name: String,
sql: String,
params: Vec<Value>,
}
#[tokio::test]
async fn deterministic_known_repros_match_reference_writer() {
for case in deterministic_repro_cases() {
Box::pin(assert_case_matches_reference(&case)).await;
}
}
#[tokio::test]
async fn generated_dml_cases_match_reference_writer() {
for case in generated_dml_cases() {
Box::pin(assert_case_matches_reference(&case)).await;
}
}
async fn assert_case_matches_reference(case: &DifferentialSqlCase) {
let reference = Box::pin(run_case(case, WriteExecutorMode::ForceDataFusion)).await;
let candidate_mode = match case.expectation {
DifferentialExpectation::SemanticParityMayFallback => WriteExecutorMode::Auto,
DifferentialExpectation::FastRequiredParity => WriteExecutorMode::ForceFast,
};
let candidate = Box::pin(run_case(case, candidate_mode)).await;
assert_expected_execution(case, &reference.execution);
assert_eq!(
candidate, reference,
"differential SQL seed '{}' diverged under {:?}\nSQL: {}",
case.seed, candidate_mode, case.sql
);
if matches!(case.expected_execution, ExpectedExecution::Err { .. }) {
Box::pin(assert_independent_no_mutation(case, &reference)).await;
} else if case.expectation == DifferentialExpectation::FastRequiredParity {
if matches!(
reference.execution,
ExecutionSignature::Ok { rows_affected: 0 }
) {
Box::pin(assert_independent_no_mutation(case, &reference)).await;
}
assert_eq!(
candidate.executor_path,
Some(WriteExecutorPath::Fast),
"differential SQL seed '{}' did not execute through the fast writer\nSQL: {}",
case.seed,
case.sql
);
} else if matches!(
reference.execution,
ExecutionSignature::Ok { rows_affected: 0 }
) {
Box::pin(assert_independent_no_mutation(case, &reference)).await;
}
}
async fn assert_independent_no_mutation(
case: &DifferentialSqlCase,
reference: &DifferentialOutcome,
) {
let baseline = Box::pin(run_baseline(case)).await;
assert_eq!(
reference.staged_rows, baseline.staged_rows,
"differential SQL seed '{}' changed staged rows in the independent no-mutation check\nSQL: {}",
case.seed, case.sql
);
assert_eq!(
reference.final_rows, baseline.final_rows,
"differential SQL seed '{}' changed final rows in the independent no-mutation check\nSQL: {}",
case.seed, case.sql
);
}
async fn run_case(case: &DifferentialSqlCase, mode: WriteExecutorMode) -> DifferentialOutcome {
let engine = open_initialized_engine().await;
let session = engine.open_session().await.expect("session should open");
create_probe_branches(&session).await;
let active_branch_id = session
.active_branch_id()
.await
.expect("differential session should have an active branch");
for setup_sql in case.setup_sql {
session
.execute_with_write_executor_mode(
setup_sql,
&[],
WriteExecutorMode::ForceDataFusion,
)
.await
.unwrap_or_else(|error| {
panic!(
"differential SQL seed '{}' setup failed\nSQL: {}\nerror: {:?}",
case.seed, setup_sql, error
)
});
}
let params = differential_params(case.params);
let mut transaction = session
.begin_transaction()
.await
.expect("differential transaction should open");
for setup_sql in case.transaction_setup_sql {
Box::pin(transaction.execute_with_write_executor_mode(
setup_sql,
&[],
WriteExecutorMode::ForceDataFusion,
))
.await
.unwrap_or_else(|error| {
panic!(
"differential SQL seed '{}' transaction setup failed\nSQL: {}\nerror: {:?}",
case.seed, setup_sql, error
)
});
}
let execution_result = Box::pin(transaction.execute_with_write_executor_mode_and_trace(
case.sql.as_ref(),
¶ms,
mode,
))
.await;
let execution = execution_signature(&execution_result);
let executor_path = execution_result
.as_ref()
.ok()
.and_then(|(_result, path)| *path);
let staged_rows = Box::pin(probe_transaction_state(
&mut transaction,
case.probes,
&active_branch_id,
))
.await;
match execution_result {
Ok(_) => transaction
.commit()
.await
.map(|_| ())
.expect("successful differential case should commit"),
Err(_) => transaction
.rollback()
.await
.expect("failed differential case should rollback"),
}
let final_rows = probe_session_state(&session, case.probes, &active_branch_id).await;
DifferentialOutcome {
execution,
executor_path,
staged_rows,
final_rows,
}
}
async fn run_baseline(case: &DifferentialSqlCase) -> DifferentialOutcome {
let engine = open_initialized_engine().await;
let session = engine.open_session().await.expect("session should open");
create_probe_branches(&session).await;
let active_branch_id = session
.active_branch_id()
.await
.expect("differential session should have an active branch");
for setup_sql in case.setup_sql {
session
.execute_with_write_executor_mode(
setup_sql,
&[],
WriteExecutorMode::ForceDataFusion,
)
.await
.unwrap_or_else(|error| {
panic!(
"differential SQL seed '{}' baseline setup failed\nSQL: {}\nerror: {:?}",
case.seed, setup_sql, error
)
});
}
let mut transaction = session
.begin_transaction()
.await
.expect("differential baseline transaction should open");
for setup_sql in case.transaction_setup_sql {
Box::pin(
transaction.execute_with_write_executor_mode(
setup_sql,
&[],
WriteExecutorMode::ForceDataFusion,
),
)
.await
.unwrap_or_else(|error| {
panic!(
"differential SQL seed '{}' baseline transaction setup failed\nSQL: {}\nerror: {:?}",
case.seed, setup_sql, error
)
});
}
let staged_rows = Box::pin(probe_transaction_state(
&mut transaction,
case.probes,
&active_branch_id,
))
.await;
transaction
.commit()
.await
.expect("baseline differential case should commit setup");
let final_rows = probe_session_state(&session, case.probes, &active_branch_id).await;
DifferentialOutcome {
execution: ExecutionSignature::Ok { rows_affected: 0 },
executor_path: None,
staged_rows,
final_rows,
}
}
async fn open_initialized_engine() -> Engine {
let storage = Memory::new();
Engine::initialize(storage.clone())
.await
.expect("unit storage should initialize");
Engine::new(storage)
.await
.expect("engine should open over initialized unit storage")
}
async fn create_probe_branches(session: &crate::session::SessionContext) {
for id in [
"01920000-0000-7000-8000-0000000000a1",
"01920000-0000-7000-8000-0000000000b1",
] {
session
.create_branch(CreateBranchOptions {
id: Some(id.to_string()),
name: id.to_string(),
from_commit_id: None,
})
.await
.unwrap_or_else(|error| panic!("failed to create probe branch {id}: {error:?}"));
}
}
fn execution_signature(
result: &Result<(ExecuteResult, Option<WriteExecutorPath>), LixError>,
) -> ExecutionSignature {
match result {
Ok((result, _path)) => ExecutionSignature::Ok {
rows_affected: result.rows_affected(),
},
Err(error) => ExecutionSignature::Err {
code: error.code.clone(),
message: error.message.clone(),
},
}
}
fn assert_expected_execution(case: &DifferentialSqlCase, execution: &ExecutionSignature) {
match (case.expected_execution, execution) {
(ExpectedExecution::Ok, ExecutionSignature::Ok { .. })
| (ExpectedExecution::Err { .. }, ExecutionSignature::Err { .. }) => {}
(ExpectedExecution::Ok, ExecutionSignature::Err { code, message }) => {
panic!(
"differential SQL seed '{}' should succeed but failed with {code}: {message}\nSQL: {}",
case.seed, case.sql
);
}
(ExpectedExecution::Err { code }, ExecutionSignature::Ok { rows_affected }) => {
panic!(
"differential SQL seed '{}' should fail with {code} but succeeded with {rows_affected} rows affected\nSQL: {}",
case.seed, case.sql
);
}
}
if let (
ExpectedExecution::Err {
code: expected_code,
},
ExecutionSignature::Err { code, message },
) = (case.expected_execution, execution)
{
assert_eq!(
code, expected_code,
"differential SQL seed '{}' failed with the wrong error code: {code}: {message}\nSQL: {}",
case.seed, case.sql
);
}
}
fn differential_params(params: &[DifferentialParam]) -> Vec<Value> {
params
.iter()
.map(|param| match param {
DifferentialParam::Jsonb(value) => {
let value =
serde_json::from_str(value).expect("differential JSON param should parse");
Value::Jsonb(value)
}
DifferentialParam::Text(value) => Value::Text((*value).to_string()),
DifferentialParam::Blob(value) => Value::Blob((*value).to_vec().into()),
})
.collect()
}
async fn probe_session_state(
session: &crate::session::SessionContext,
probes: &[DifferentialProbe],
_active_branch_id: &str,
) -> Vec<ProbeSnapshot> {
let mut snapshots = Vec::with_capacity(probes.len());
for probe in probes {
let query = probe_query(probe);
snapshots.push(ProbeSnapshot {
name: query.name,
rows: session
.execute(&query.sql, &query.params)
.await
.unwrap_or_else(|error| {
panic!(
"final differential probe failed\nSQL: {}\nerror: {error:?}",
query.sql
)
})
.rows()
.iter()
.map(|row| row.values().to_vec())
.collect(),
});
}
snapshots
}
async fn probe_transaction_state(
transaction: &mut crate::session::SessionTransaction,
probes: &[DifferentialProbe],
_active_branch_id: &str,
) -> Vec<ProbeSnapshot> {
let mut snapshots = Vec::with_capacity(probes.len());
for probe in probes {
let query = probe_query(probe);
snapshots.push(ProbeSnapshot {
name: query.name,
rows: transaction
.execute(&query.sql, &query.params)
.await
.unwrap_or_else(|error| {
panic!(
"staged differential probe failed\nSQL: {}\nerror: {error:?}",
query.sql
)
})
.rows()
.iter()
.map(|row| row.values().to_vec())
.collect(),
});
}
snapshots
}
fn probe_query(probe: &DifferentialProbe) -> ProbeQuery {
match probe {
DifferentialProbe::RegisteredSchemaActive => ProbeQuery {
name: "lix_registered_schema".to_string(),
sql: "SELECT schema_key, value, lixcol_metadata, lixcol_global, lixcol_untracked \
FROM lix_registered_schema \
ORDER BY schema_key"
.to_string(),
params: Vec::new(),
},
DifferentialProbe::LixFileActive { paths } => {
let mut params = Vec::with_capacity(paths.len());
let placeholders = paths
.iter()
.enumerate()
.map(|(index, path)| {
params.push(Value::Text((*path).to_string()));
format!("${}", index + 1)
})
.collect::<Vec<_>>()
.join(", ");
ProbeQuery {
name: format!("lix_file:{paths:?}"),
sql: format!(
"SELECT path, content \
FROM lix_file \
WHERE path IN ({placeholders}) \
ORDER BY path"
),
params,
}
}
}
}
}