use std::sync::Arc;
use surrealdb_types::{SqlFormat, ToSql, write_sql};
use crate::catalog::Index;
use crate::exec::physical_expr::{EvalContext, PhysicalExpr};
use crate::exec::{AccessMode, BoxFut, ContextLevel};
use crate::expr::FlowResult;
use crate::expr::idiom::Idiom;
use crate::expr::operator::MatchesOperator;
use crate::idx::IndexKeyBase;
use crate::idx::ft::fulltext::{FullTextIndex, QueryTerms};
use crate::kvs::index::filter_online_indexes;
use crate::val::{TableName, Value};
pub struct MatchesOp {
pub(crate) left: Arc<dyn PhysicalExpr>,
pub(crate) right: Arc<dyn PhysicalExpr>,
pub(crate) operator: MatchesOperator,
pub(crate) idiom: Idiom,
pub(crate) query: String,
ft_cache: tokio::sync::OnceCell<Option<(FullTextIndex, QueryTerms)>>,
}
impl MatchesOp {
pub fn new(
left: Arc<dyn PhysicalExpr>,
right: Arc<dyn PhysicalExpr>,
operator: MatchesOperator,
idiom: Idiom,
query: String,
) -> Self {
Self {
left,
right,
operator,
idiom,
query,
ft_cache: tokio::sync::OnceCell::new(),
}
}
async fn ft_resources(
&self,
ctx: &EvalContext<'_>,
) -> Result<&Option<(FullTextIndex, QueryTerms)>, anyhow::Error> {
self.ft_cache
.get_or_try_init(|| async {
use crate::catalog::providers::TableProvider;
let frozen = ctx.exec_ctx.ctx();
let root = ctx.exec_ctx.root();
let opt = root
.options
.as_ref()
.ok_or_else(|| anyhow::anyhow!("MatchesOp requires Options context"))?;
let tx = ctx.txn();
let db_ctx = ctx
.exec_ctx
.database()
.map_err(|e| anyhow::anyhow!("MatchesOp requires database context: {}", e))?;
let ns_id = db_ctx.ns_ctx.ns.namespace_id;
let db_id = db_ctx.db.database_id;
let table_name = if let Some(value) = ctx.current_value {
extract_table_from_value(value)
} else {
None
};
let table_name = match table_name {
Some(t) => t,
None => {
if let Some(mc) = frozen.get_matches_context()
&& let Some(table) = mc.table()
{
table.clone()
} else {
return Ok(None);
}
}
};
let indexes = tx
.all_tb_indexes(ns_id, db_id, &table_name, ctx.exec_ctx.version_stamp())
.await?;
let indexes = if ctx.exec_ctx.version_stamp().is_none() {
filter_online_indexes(tx.as_ref(), ns_id, db_id, indexes).await?
} else {
indexes
};
let index_def = indexes.iter().find(|idx| {
matches!(&idx.index, Index::FullText(_))
&& idx.cols.iter().any(|col| col.0 == self.idiom.0)
});
let index_def = match index_def {
Some(def) => def,
None => return Ok(None),
};
let ft_params = match &index_def.index {
Index::FullText(params) => params,
_ => unreachable!("Already checked for FullText above"),
};
let ikb = IndexKeyBase::new(ns_id, db_id, table_name, index_def.index_id);
let fti = FullTextIndex::new(
frozen.get_index_stores(),
tx.as_ref(),
ikb,
ft_params,
&frozen.config.file_allowlist,
)
.await?;
let query_terms = {
let mut stack = reblessive::TreeStack::new();
stack
.enter(|stk| {
fti.extract_querying_terms(stk, frozen, opt, self.query.clone())
})
.finish()
.await?
};
Ok(Some((fti, query_terms)))
})
.await
}
}
impl PhysicalExpr for MatchesOp {
fn name(&self) -> &'static str {
"MatchesOp"
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn required_context(&self) -> ContextLevel {
let children = self.left.required_context().max(self.right.required_context());
children.max(ContextLevel::Root)
}
fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
Box::pin(async move {
let ft = self.ft_resources(&ctx).await?;
let (fti, qt) = match ft {
None => return Ok(Value::Bool(false)),
Some(resources) => resources,
};
if qt.is_empty() {
return Ok(Value::Bool(false));
}
let rid = extract_record_id(ctx.current_value)?;
let tx = ctx.txn();
let matches = match fti.get_doc_id(&tx, &rid).await? {
Some(doc_id) => qt.contains_doc(doc_id),
None => false,
};
Ok(Value::Bool(matches))
})
}
fn access_mode(&self) -> AccessMode {
self.left.access_mode().combine(self.right.access_mode())
}
}
impl ToSql for MatchesOp {
fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
write_sql!(f, fmt, "{} {} {}", self.left, self.operator, self.right);
}
}
impl Clone for MatchesOp {
fn clone(&self) -> Self {
Self {
left: Arc::clone(&self.left),
right: Arc::clone(&self.right),
operator: self.operator.clone(),
idiom: self.idiom.clone(),
query: self.query.clone(),
ft_cache: tokio::sync::OnceCell::new(),
}
}
}
impl std::fmt::Debug for MatchesOp {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("MatchesOp")
.field("idiom", &self.idiom)
.field("query", &self.query)
.field("operator", &self.operator)
.field("initialized", &self.ft_cache.initialized())
.finish()
}
}
fn extract_record_id(value: Option<&Value>) -> Result<crate::val::RecordId, anyhow::Error> {
let current =
value.ok_or_else(|| anyhow::anyhow!("MATCHES evaluation requires a current document"))?;
match current {
Value::Object(obj) => match obj.get("id") {
Some(Value::RecordId(rid)) => Ok(rid.clone()),
Some(_) => Err(anyhow::anyhow!("Current document 'id' field is not a RecordId")),
None => Err(anyhow::anyhow!("Current document has no 'id' field")),
},
Value::RecordId(rid) => Ok(rid.clone()),
_ => Err(anyhow::anyhow!(
"Expected current document to be an Object for MATCHES evaluation, got: {}",
current.kind_of()
)),
}
}
fn extract_table_from_value(value: &Value) -> Option<TableName> {
match value {
Value::Object(obj) => match obj.get("id") {
Some(Value::RecordId(rid)) => Some(rid.table.clone()),
_ => None,
},
Value::RecordId(rid) => Some(rid.table.clone()),
_ => None,
}
}