use super::*;
use uqa_operators::TextScoringMode;
fn populate_calibration_fixture(engine: &Engine) {
engine
.sql("CREATE TABLE docs (id INTEGER PRIMARY KEY, body TEXT)", &[])
.unwrap();
engine
.sql("CREATE INDEX docs_fts ON docs USING gin (body)", &[])
.unwrap();
engine
.sql(
"INSERT INTO docs (id, body) VALUES \
(1, 'rust search engine'), \
(2, 'rust database query'), \
(3, 'search ranking calibration')",
&[],
)
.unwrap();
}
fn calibration_then_failure() -> OperatorTree {
OperatorTree::Composed(vec![
OperatorTree::Term {
query: "rust".into(),
field: Some("body".into()),
scoring: Some(TextScoringMode::BayesianBM25),
top_k: None,
},
OperatorTree::KNN {
query_vector: vec![1.0],
k: 1,
field: "missing_embedding".into(),
},
])
}
fn assert_failed_tree_rolls_back_calibration(engine: &Engine) {
use crate::operator_tree_bridge::OperatorTreeDriver as _;
for physical in [false, true] {
assert!(engine.load_scoring_params("docs.body").unwrap().is_none());
let result = if physical {
EngineDriver::new(engine, "docs", &[]).execute_node(&calibration_then_failure())
} else {
execute_operator_tree(engine, "docs", "docs", &[], &calibration_then_failure())
};
result.expect_err("the malformed downstream vector leaf must fail");
assert!(
engine.load_scoring_params("docs.body").unwrap().is_none(),
"failed operator execution leaked auto-calibration state"
);
assert_eq!(engine.transaction_depth(), 0);
}
}
#[test]
fn failed_calibrating_tree_rolls_back_memory_state() {
let engine = Engine::new();
populate_calibration_fixture(&engine);
assert_failed_tree_rolls_back_calibration(&engine);
}
#[test]
fn failed_calibrating_tree_rolls_back_catalog_and_reopen_state() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("calibration.sqlite");
let engine = Engine::open(&path).unwrap();
populate_calibration_fixture(&engine);
assert_failed_tree_rolls_back_calibration(&engine);
drop(engine);
let reopened = Engine::open(&path).unwrap();
assert!(reopened.load_scoring_params("docs.body").unwrap().is_none());
}
#[test]
fn graph_catalog_workers_retain_the_callers_statement_scope() {
use uqa_execution::query::graph_lifecycle::GraphLifecycle;
use uqa_sql::semantics::graph_functions::GraphNameCatalog;
let directory = tempfile::tempdir().unwrap();
for engine in [
Engine::new(),
Engine::open(&directory.path().join("graph.sqlite")).unwrap(),
] {
engine.create_graph("g").unwrap();
engine
.sql("BEGIN ISOLATION LEVEL SERIALIZABLE", &[])
.unwrap();
engine
.with_direct_read_snapshot(|engine| {
std::thread::scope(|scope| {
scope
.spawn(|| {
assert_eq!(GraphNameCatalog::list_graphs(engine).unwrap(), vec!["g"]);
assert!(GraphLifecycle::has_graph(engine, "g").unwrap());
assert!(GraphLifecycle::list_graph_labels(engine, "g")
.unwrap()
.is_some());
})
.join()
.unwrap();
});
Ok(())
})
.unwrap();
assert_eq!(engine.transaction_depth(), 1);
engine.commit().unwrap();
}
}