upend/db/src/engine.rs

315 lines
14 KiB
Rust

use std::collections::HashMap;
use std::iter::zip;
use super::inner::models::Entry;
use super::inner::schema::data;
use crate::inner::models;
use anyhow::Result;
use diesel::expression::grouped::Grouped;
use diesel::expression::operators::{And, Not, Or};
use diesel::sql_types::Bool;
use diesel::sqlite::Sqlite;
use diesel::IntoSql;
use diesel::RunQueryDsl;
use diesel::{
r2d2::{ConnectionManager, PooledConnection},
SqliteConnection,
};
use diesel::{BoxableExpression, QueryDsl};
use diesel::{ExpressionMethods, TextExpressionMethods};
use upend_base::entry::EntryValue;
use upend_base::lang::{PatternQuery, Query, QueryComponent, QueryPart, QueryQualifier};
#[derive(Debug, Clone)]
pub struct QueryExecutionError(String);
impl std::fmt::Display for QueryExecutionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
impl std::error::Error for QueryExecutionError {}
pub fn execute(
connection: &PooledConnection<ConnectionManager<SqliteConnection>>,
query: Query,
) -> Result<Vec<Entry>, QueryExecutionError> {
use crate::inner::schema::data::dsl::*;
if let Some(predicates) = to_sqlite_predicates(query.clone())? {
let db_query = data.filter(predicates);
db_query
.load::<models::Entry>(connection)
.map_err(|e| QueryExecutionError(e.to_string()))
} else {
match query {
Query::SingleQuery(_) => Err(QueryExecutionError(
"Forced manual evaluation of an atomic query, this should never happen.".into(),
)),
Query::MultiQuery(mq) => match mq.qualifier {
QueryQualifier::Not => Err(QueryExecutionError(
"Stopped manual evaluation at NOT sub-query due to performance limits. Please \
rework your query."
.into(),
)),
_ => {
let subquery_results = mq
.queries
.iter()
.map(|q| execute(connection, *q.clone()))
.collect::<Result<Vec<Vec<Entry>>, QueryExecutionError>>()?;
match mq.qualifier {
QueryQualifier::Not => unreachable!(),
QueryQualifier::And => Ok(subquery_results
.into_iter()
.reduce(|acc, cur| {
acc.into_iter()
.filter(|e| {
cur.iter().map(|e| &e.identity).any(|x| x == &e.identity)
})
.collect()
})
.unwrap()), // TODO
QueryQualifier::Or => Ok(subquery_results.into_iter().flatten().collect()),
QueryQualifier::Join => {
let pattern_queries = mq
.queries
.into_iter()
.map(|q| match *q {
Query::SingleQuery(QueryPart::Matches(pq)) => Some(pq),
_ => None,
})
.collect::<Option<Vec<_>>>();
if let Some(pattern_queries) = pattern_queries {
let entries = zip(pattern_queries, subquery_results).map(
|(query, results)| {
results
.into_iter()
.map(|e| EntryWithVars::new(&query, e))
.collect::<Vec<EntryWithVars>>()
},
);
let joined = entries
.reduce(|acc, cur| {
acc.into_iter()
.filter(|tested_entry| {
tested_entry.vars.iter().any(|(k1, v1)| {
cur.iter().any(|other_entry| {
other_entry
.vars
.iter()
.any(|(k2, v2)| k1 == k2 && v1 == v2)
})
})
})
.collect()
})
.unwrap(); // TODO
Ok(joined.into_iter().map(|ev| ev.entry).collect())
} else {
Err(QueryExecutionError(
"Cannot join on non-atomic queries.".into(),
))
}
}
}
}
},
}
}
}
struct EntryWithVars {
entry: Entry,
vars: HashMap<String, String>,
}
impl EntryWithVars {
pub fn new(query: &PatternQuery, entry: Entry) -> Self {
let mut vars = HashMap::new();
if let QueryComponent::Variable(Some(var_name)) = &query.entity {
vars.insert(
var_name.clone(),
upend_base::hash::b58_encode(&entry.entity),
);
}
if let QueryComponent::Variable(Some(var_name)) = &query.attribute {
vars.insert(var_name.clone(), entry.attribute.clone());
}
if let QueryComponent::Variable(Some(var_name)) = &query.value {
if let Some(value_str) = &entry.value_str {
vars.insert(var_name.clone(), value_str.clone());
}
}
EntryWithVars { entry, vars }
}
}
type SqlPredicate = dyn BoxableExpression<data::table, Sqlite, SqlType = Bool>;
type SqlResult = Option<Box<SqlPredicate>>;
fn to_sqlite_predicates(query: Query) -> Result<SqlResult, QueryExecutionError> {
match query {
Query::SingleQuery(qp) => match qp {
QueryPart::Matches(eq) => {
let mut subqueries: Vec<Box<SqlPredicate>> = vec![];
match &eq.entity {
QueryComponent::Exact(q_entity) => {
subqueries.push(Box::new(data::entity.eq(q_entity.encode().map_err(
|e| QueryExecutionError(format!("failed producing sql: {e}")),
)?)))
}
QueryComponent::In(q_entities) => {
let entities: Result<Vec<_>, _> =
q_entities.iter().map(|t| t.encode()).collect();
subqueries.push(Box::new(data::entity.eq_any(entities.map_err(|e| {
QueryExecutionError(format!("failed producing sql: {e}"))
})?)))
}
QueryComponent::Contains(q_entity) => subqueries.push(Box::new(
data::entity_searchable.like(format!("%{}%", q_entity)),
)),
QueryComponent::Variable(_) => {}
};
match &eq.attribute {
QueryComponent::Exact(q_attribute) => {
subqueries.push(Box::new(data::attribute.eq(q_attribute.0.clone())))
}
QueryComponent::In(q_attributes) => subqueries.push(Box::new(
data::attribute.eq_any(q_attributes.iter().map(|a| &a.0).cloned()),
)),
QueryComponent::Contains(q_attribute) => subqueries
.push(Box::new(data::attribute.like(format!("%{}%", q_attribute)))),
QueryComponent::Variable(_) => {}
};
match &eq.value {
QueryComponent::Exact(q_value) => match q_value {
EntryValue::Number(n) => subqueries.push(Box::new(data::value_num.eq(*n))),
_ => subqueries.push(Box::new(data::value_str.eq(q_value.to_string()))),
},
QueryComponent::In(q_values) => {
let first = q_values.first().ok_or_else(|| {
QueryExecutionError(
"Malformed expression: Inner value cannot be empty.".into(),
)
})?;
match first {
EntryValue::Number(_) => subqueries.push(Box::new(
data::value_num.eq_any(
q_values
.iter()
.map(|v| {
if let EntryValue::Number(n) = v {
Ok(*n)
} else {
Err(QueryExecutionError(format!(
"IN queries must not combine numeric and \
string values! ({v} is not a number)"
)))
}
})
.collect::<Result<Vec<f64>, QueryExecutionError>>()?,
),
)),
_ => subqueries.push(Box::new(
data::value_str.eq_any(
q_values
.iter()
.map(|v| {
if let EntryValue::Number(_) = v {
Err(QueryExecutionError(format!(
"IN queries must not combine numeric and \
string values! (Found {v})"
)))
} else {
Ok(v.to_string())
}
})
.collect::<Result<Vec<String>, QueryExecutionError>>()?,
),
)),
}
}
QueryComponent::Contains(q_value) => {
subqueries.push(Box::new(data::value_str.like(format!("S%{}%", q_value))))
}
QueryComponent::Variable(_) => {}
};
match subqueries.len() {
0 => Ok(Some(Box::new(true.into_sql::<Bool>()))),
1 => Ok(Some(subqueries.remove(0))),
_ => {
let mut result: Box<And<Box<SqlPredicate>, Box<SqlPredicate>>> =
Box::new(And::new(subqueries.remove(0), subqueries.remove(0)));
while !subqueries.is_empty() {
result = Box::new(And::new(result, subqueries.remove(0)));
}
Ok(Some(Box::new(result)))
}
}
}
QueryPart::Type(_) => unimplemented!("Type queries are not yet implemented."),
},
Query::MultiQuery(mq) => {
let mq_result = mq
.queries
.into_iter()
.map(|sq| to_sqlite_predicates(*sq))
.collect::<Result<Vec<SqlResult>, QueryExecutionError>>()?;
let mq_result: Option<Vec<Box<SqlPredicate>>> = mq_result.into_iter().collect();
if let Some(mut subqueries) = mq_result {
match subqueries.len() {
0 => Ok(Some(Box::new(true.into_sql::<Bool>()))),
1 => {
if let QueryQualifier::Not = mq.qualifier {
Ok(Some(Box::new(Not::new(subqueries.remove(0)))))
} else {
Ok(Some(subqueries.remove(0)))
}
}
_ => match mq.qualifier {
QueryQualifier::Join => Ok(None),
QueryQualifier::And => {
let mut result: Box<And<Box<SqlPredicate>, Box<SqlPredicate>>> =
Box::new(And::new(subqueries.remove(0), subqueries.remove(0)));
while !subqueries.is_empty() {
result = Box::new(And::new(result, subqueries.remove(0)));
}
Ok(Some(Box::new(Grouped(result))))
}
QueryQualifier::Or => {
let mut result =
Box::new(Or::new(subqueries.remove(0), subqueries.remove(0)));
while !subqueries.is_empty() {
result = Box::new(Or::new(result, subqueries.remove(0)));
}
Ok(Some(Box::new(Grouped(result))))
}
QueryQualifier::Not => {
Err(QueryExecutionError("NOT only takes one subquery.".into()))
}
},
}
} else {
Ok(None)
}
}
}
}