uqa_sql/semantics/
text_indexes.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Validation of text-retrieval field references against relation and index metadata.
8
9use super::{multi_field_match_shape, MultiFieldMatchShape};
10use crate::{plan::SourcePlan, SQLError, ScalarExpr};
11use uqa_core::Value;
12
13pub trait TextMatchCatalog {
14    fn has_table(&self, table: &str) -> Result<bool, String>;
15    fn has_column(&self, table: &str, column: &str) -> Result<bool, String>;
16    fn column_names(&self, table: &str) -> Result<Vec<String>, String>;
17    fn indexed_fields(&self, table: &str) -> Result<Vec<String>, SQLError>;
18}
19
20const SINGLE_FIELD_TEXT_MATCH_FUNCTIONS: [&str; 4] = [
21    "text_match",
22    "bayesian_match",
23    "fts_match",
24    "bayesian_match_with_prior",
25];
26
27/// Walk an expression tree and hand every text-match field argument to
28/// `validate`. Used by the select runners to reject silently-empty
29/// searches before the WHERE reaches either the operator-tree access path
30/// or scalar evaluation in the relational filter node.
31fn walk_text_match_fields(
32    expr: &ScalarExpr,
33    validate: &mut dyn FnMut(&ScalarExpr, &str) -> Result<(), SQLError>,
34) -> Result<(), SQLError> {
35    match expr {
36        ScalarExpr::Func {
37            name, args, filter, ..
38        } => {
39            let lower = name.to_ascii_lowercase();
40            if SINGLE_FIELD_TEXT_MATCH_FUNCTIONS.contains(&lower.as_str()) {
41                if let Some(field_arg) = args.first() {
42                    if !(lower == "fts_match" && fts_query_is_jsonpath(args.get(1))) {
43                        validate(field_arg, &lower)?;
44                    }
45                }
46            } else if lower == "multi_field_match" {
47                match multi_field_match_shape(args)? {
48                    MultiFieldMatchShape::FieldsThenQuery { fields, .. }
49                    | MultiFieldMatchShape::Pairs { fields } => {
50                        for field_arg in fields {
51                            validate(field_arg, "multi_field_match")?;
52                        }
53                    }
54                }
55            }
56            for arg in args {
57                walk_text_match_fields(arg, validate)?;
58            }
59            if let Some(filter) = filter {
60                walk_text_match_fields(filter, validate)?;
61            }
62            Ok(())
63        }
64        ScalarExpr::And(items)
65        | ScalarExpr::Or(items)
66        | ScalarExpr::Array(items)
67        | ScalarExpr::Row(items) => {
68            for item in items {
69                walk_text_match_fields(item, validate)?;
70            }
71            Ok(())
72        }
73        ScalarExpr::Not(inner) | ScalarExpr::UnaryMinus(inner) => {
74            walk_text_match_fields(inner, validate)
75        }
76        ScalarExpr::Binary { lhs, rhs, .. } => {
77            walk_text_match_fields(lhs, validate)?;
78            walk_text_match_fields(rhs, validate)
79        }
80        ScalarExpr::IsNull { expr, .. } => walk_text_match_fields(expr, validate),
81        ScalarExpr::Between { expr, low, high } => {
82            walk_text_match_fields(expr, validate)?;
83            walk_text_match_fields(low, validate)?;
84            walk_text_match_fields(high, validate)
85        }
86        ScalarExpr::InList { expr, list, .. } => {
87            walk_text_match_fields(expr, validate)?;
88            for item in list {
89                walk_text_match_fields(item, validate)?;
90            }
91            Ok(())
92        }
93        _ => Ok(()),
94    }
95}
96
97pub fn validate_expr_text_match_fields(
98    catalog: &dyn TextMatchCatalog,
99    table: &str,
100    expr: &ScalarExpr,
101) -> Result<(), SQLError> {
102    walk_text_match_fields(
103        expr,
104        &mut |field_arg, function_name| match text_match_field_name(field_arg) {
105            Some(TextMatchField::All) => {
106                validate_text_match_all_fields(catalog, table, function_name)
107            }
108            Some(TextMatchField::Named(field)) => {
109                validate_text_match_field(catalog, table, field, function_name)
110            }
111            None => Ok(()),
112        },
113    )
114}
115
116enum TextMatchField<'a> {
117    All,
118    Named(&'a str),
119}
120
121/// The `_all` pseudo-field arrives either as a string literal or as a
122/// bare column reference, depending on how the query was written.
123fn text_match_field_name(field_arg: &ScalarExpr) -> Option<TextMatchField<'_>> {
124    match field_arg {
125        ScalarExpr::Column(name) | ScalarExpr::QualifiedColumn { column: name, .. } => {
126            if name.is_empty() || name == "_all" {
127                Some(TextMatchField::All)
128            } else {
129                Some(TextMatchField::Named(name))
130            }
131        }
132        ScalarExpr::Literal(Value::Str(s)) if s.is_empty() || s == "_all" => {
133            Some(TextMatchField::All)
134        }
135        _ => None,
136    }
137}
138
139pub fn validate_joined_expr_text_match_fields(
140    catalog: &dyn TextMatchCatalog,
141    from: &SourcePlan,
142    expr: &ScalarExpr,
143) -> Result<(), SQLError> {
144    let mut tables: Vec<(Option<String>, String, Vec<String>)> = Vec::new();
145    let mut has_opaque_source = false;
146    collect_from_tables(from, &mut tables, &mut has_opaque_source);
147    walk_text_match_fields(expr, &mut |field_arg, function_name| {
148        let (qualifier, column) = match field_arg {
149            ScalarExpr::Column(name) => (None, name.as_str()),
150            ScalarExpr::QualifiedColumn {
151                qualifier, column, ..
152            } => (Some(qualifier.as_str()), column.as_str()),
153            _ => return Ok(()),
154        };
155        if column.is_empty() || column == "_all" {
156            return Ok(());
157        }
158        if let Some(qualifier) = qualifier {
159            let resolved = tables
160                .iter()
161                .find(|(alias, name, _)| alias.as_deref() == Some(qualifier) || name == qualifier);
162            return match resolved {
163                Some((_, table, aliases)) => {
164                    let physical = table_source_physical_column(catalog, table, aliases, column)?
165                        .unwrap_or_else(|| column.to_string());
166                    validate_text_match_field(catalog, table, &physical, function_name)
167                }
168                // Unknown qualifiers can point at subqueries or CTEs the
169                // validator cannot introspect.
170                None => Ok(()),
171            };
172        }
173        let mut containing = Vec::new();
174        for (_, name, aliases) in &tables {
175            if let Some(physical) = table_source_physical_column(catalog, name, aliases, column)? {
176                containing.push((name, physical));
177            }
178        }
179        for (name, physical) in &containing {
180            if catalog
181                .indexed_fields(name)?
182                .iter()
183                .any(|field| field == physical)
184            {
185                return Ok(());
186            }
187        }
188        if let Some((table, physical)) = containing.first() {
189            return validate_text_match_field(catalog, table, physical, function_name);
190        }
191        if has_opaque_source {
192            return Ok(());
193        }
194        Err(SQLError::TypeMismatch(format!(
195            "{function_name}: column `{column}` does not exist on any joined table"
196        )))
197    })
198}
199
200fn table_source_physical_column(
201    catalog: &dyn TextMatchCatalog,
202    table: &str,
203    aliases: &[String],
204    visible: &str,
205) -> Result<Option<String>, SQLError> {
206    let columns = catalog
207        .column_names(table)
208        .map_err(|error| SQLError::Internal(format!("read table schema: {error}")))?;
209    if columns.is_empty() {
210        return Ok(Some(visible.to_string()));
211    }
212    Ok(columns
213        .into_iter()
214        .enumerate()
215        .find_map(|(position, physical)| {
216            aliases
217                .get(position)
218                .map_or_else(
219                    || physical.eq_ignore_ascii_case(visible),
220                    |alias| alias.eq_ignore_ascii_case(visible),
221                )
222                .then_some(physical)
223        }))
224}
225
226pub use super::fts_query_is_jsonpath;
227
228fn collect_from_tables(
229    from: &SourcePlan,
230    out: &mut Vec<(Option<String>, String, Vec<String>)>,
231    has_opaque_source: &mut bool,
232) {
233    match from {
234        SourcePlan::Table {
235            name,
236            qualifier,
237            alias,
238            column_aliases,
239            ..
240        } => out.push((
241            Some(alias.as_ref().unwrap_or(qualifier).clone()),
242            name.clone(),
243            column_aliases.clone(),
244        )),
245        SourcePlan::Join {
246            left, right, alias, ..
247        } => {
248            if alias.is_some() {
249                *has_opaque_source = true;
250            } else {
251                collect_from_tables(left, out, has_opaque_source);
252                collect_from_tables(right, out, has_opaque_source);
253            }
254        }
255        _ => *has_opaque_source = true,
256    }
257}
258
259/// Reject silently-empty text searches up front: a match function whose
260/// field is not a real column, or is a column without a text index,
261/// previously returned zero rows with no diagnostic.
262pub fn validate_text_match_field(
263    catalog: &dyn TextMatchCatalog,
264    table: &str,
265    field: &str,
266    function_name: &str,
267) -> Result<(), SQLError> {
268    if !catalog
269        .has_table(table)
270        .map_err(|err| SQLError::Internal(format!("read table catalog: {err}")))?
271    {
272        return Err(SQLError::TypeMismatch(format!(
273            "{function_name}: unknown table `{table}`"
274        )));
275    }
276    let indexed = catalog
277        .indexed_fields(table)?
278        .iter()
279        .any(|fts| fts == field);
280    if !indexed {
281        if !catalog
282            .has_column(table, field)
283            .map_err(|err| SQLError::Internal(format!("read table schema: {err}")))?
284            && !catalog
285                .column_names(table)
286                .map_err(|err| SQLError::Internal(format!("read table schema: {err}")))?
287                .is_empty()
288        {
289            return Err(SQLError::TypeMismatch(format!(
290                "{function_name}: column `{field}` does not exist on table `{table}`"
291            )));
292        }
293        return Err(SQLError::TypeMismatch(format!(
294            "{function_name}: column `{table}.{field}` has no text index; \
295             create one with CREATE INDEX ... ON {table} USING gin ({field})"
296        )));
297    }
298    Ok(())
299}
300
301pub fn validate_text_match_all_fields(
302    catalog: &dyn TextMatchCatalog,
303    table: &str,
304    function_name: &str,
305) -> Result<(), SQLError> {
306    if !catalog
307        .has_table(table)
308        .map_err(|err| SQLError::Internal(format!("read table catalog: {err}")))?
309    {
310        return Err(SQLError::TypeMismatch(format!(
311            "{function_name}: unknown table `{table}`"
312        )));
313    }
314    if catalog.indexed_fields(table)?.is_empty() {
315        return Err(SQLError::TypeMismatch(format!(
316            "{function_name}: table `{table}` has no text-indexed columns; \
317             create one with CREATE INDEX ... ON {table} USING gin (...)"
318        )));
319    }
320    Ok(())
321}
322
323/// Validate the physical field using one retained catalog generation; indexed fields do not require a column-schema read.
324pub fn require_physical_text_index(
325    table: &str,
326    field: &str,
327    indexed_fields: &[String],
328    columns: impl FnOnce() -> Vec<crate::ast::ColumnDef>,
329) -> Result<(), SQLError> {
330    if indexed_fields.iter().any(|indexed| indexed == field) {
331        return Ok(());
332    }
333    let columns = columns();
334    if !columns.is_empty() && !columns.iter().any(|column| column.name == field) {
335        return Err(SQLError::UnknownColumn(field.to_string()));
336    }
337    Err(SQLError::TypeMismatch(format!(
338        "text search: column `{table}.{field}` has no text index; create one with CREATE INDEX ... ON {table} USING gin ({field})"
339    )))
340}
uqa_sql/semantics/text_indexes.rs

uqa_sql/semantics/
text_indexes.rs
