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        | ScalarExpr::CompositeRow { items, .. } => {
69            for item in items {
70                walk_text_match_fields(item, validate)?;
71            }
72            Ok(())
73        }
74        ScalarExpr::Not(inner) | ScalarExpr::UnaryMinus(inner) => {
75            walk_text_match_fields(inner, validate)
76        }
77        ScalarExpr::Binary { lhs, rhs, .. } => {
78            walk_text_match_fields(lhs, validate)?;
79            walk_text_match_fields(rhs, validate)
80        }
81        ScalarExpr::IsNull { expr, .. } => walk_text_match_fields(expr, validate),
82        ScalarExpr::Between { expr, low, high } => {
83            walk_text_match_fields(expr, validate)?;
84            walk_text_match_fields(low, validate)?;
85            walk_text_match_fields(high, validate)
86        }
87        ScalarExpr::InList { expr, list, .. } => {
88            walk_text_match_fields(expr, validate)?;
89            for item in list {
90                walk_text_match_fields(item, validate)?;
91            }
92            Ok(())
93        }
94        _ => Ok(()),
95    }
96}
97
98pub fn validate_expr_text_match_fields(
99    catalog: &dyn TextMatchCatalog,
100    table: &str,
101    expr: &ScalarExpr,
102) -> Result<(), SQLError> {
103    walk_text_match_fields(
104        expr,
105        &mut |field_arg, function_name| match text_match_field_name(field_arg) {
106            Some(TextMatchField::All) => {
107                validate_text_match_all_fields(catalog, table, function_name)
108            }
109            Some(TextMatchField::Named(field)) => {
110                validate_text_match_field(catalog, table, field, function_name)
111            }
112            None => Ok(()),
113        },
114    )
115}
116
117enum TextMatchField<'a> {
118    All,
119    Named(&'a str),
120}
121
122/// The `_all` pseudo-field arrives either as a string literal or as a
123/// bare column reference, depending on how the query was written.
124fn text_match_field_name(field_arg: &ScalarExpr) -> Option<TextMatchField<'_>> {
125    match field_arg {
126        ScalarExpr::Column(name) | ScalarExpr::QualifiedColumn { column: name, .. } => {
127            if name.is_empty() || name == "_all" {
128                Some(TextMatchField::All)
129            } else {
130                Some(TextMatchField::Named(name))
131            }
132        }
133        ScalarExpr::Literal(Value::Str(s)) if s.is_empty() || s == "_all" => {
134            Some(TextMatchField::All)
135        }
136        _ => None,
137    }
138}
139
140pub fn validate_joined_expr_text_match_fields(
141    catalog: &dyn TextMatchCatalog,
142    from: &SourcePlan,
143    expr: &ScalarExpr,
144) -> Result<(), SQLError> {
145    let mut tables: Vec<(Option<String>, String, Vec<String>)> = Vec::new();
146    let mut has_opaque_source = false;
147    collect_from_tables(from, &mut tables, &mut has_opaque_source);
148    walk_text_match_fields(expr, &mut |field_arg, function_name| {
149        let (qualifier, column) = match field_arg {
150            ScalarExpr::Column(name) => (None, name.as_str()),
151            ScalarExpr::QualifiedColumn {
152                qualifier, column, ..
153            } => (Some(qualifier.as_str()), column.as_str()),
154            _ => return Ok(()),
155        };
156        if column.is_empty() || column == "_all" {
157            return Ok(());
158        }
159        if let Some(qualifier) = qualifier {
160            let resolved = tables
161                .iter()
162                .find(|(alias, name, _)| alias.as_deref() == Some(qualifier) || name == qualifier);
163            return match resolved {
164                Some((_, table, aliases)) => {
165                    let physical = table_source_physical_column(catalog, table, aliases, column)?
166                        .unwrap_or_else(|| column.to_string());
167                    validate_text_match_field(catalog, table, &physical, function_name)
168                }
169                // Unknown qualifiers can point at subqueries or CTEs the
170                // validator cannot introspect.
171                None => Ok(()),
172            };
173        }
174        let mut containing = Vec::new();
175        for (_, name, aliases) in &tables {
176            if let Some(physical) = table_source_physical_column(catalog, name, aliases, column)? {
177                containing.push((name, physical));
178            }
179        }
180        for (name, physical) in &containing {
181            if catalog
182                .indexed_fields(name)?
183                .iter()
184                .any(|field| field == physical)
185            {
186                return Ok(());
187            }
188        }
189        if let Some((table, physical)) = containing.first() {
190            return validate_text_match_field(catalog, table, physical, function_name);
191        }
192        if has_opaque_source {
193            return Ok(());
194        }
195        Err(SQLError::TypeMismatch(format!(
196            "{function_name}: column `{column}` does not exist on any joined table"
197        )))
198    })
199}
200
201fn table_source_physical_column(
202    catalog: &dyn TextMatchCatalog,
203    table: &str,
204    aliases: &[String],
205    visible: &str,
206) -> Result<Option<String>, SQLError> {
207    let columns = catalog
208        .column_names(table)
209        .map_err(|error| SQLError::Internal(format!("read table schema: {error}")))?;
210    if columns.is_empty() {
211        return Ok(Some(visible.to_string()));
212    }
213    Ok(columns
214        .into_iter()
215        .enumerate()
216        .find_map(|(position, physical)| {
217            aliases
218                .get(position)
219                .map_or_else(
220                    || physical.eq_ignore_ascii_case(visible),
221                    |alias| alias.eq_ignore_ascii_case(visible),
222                )
223                .then_some(physical)
224        }))
225}
226
227pub use super::fts_query_is_jsonpath;
228
229fn collect_from_tables(
230    from: &SourcePlan,
231    out: &mut Vec<(Option<String>, String, Vec<String>)>,
232    has_opaque_source: &mut bool,
233) {
234    match from {
235        SourcePlan::Table {
236            name,
237            qualifier,
238            alias,
239            column_aliases,
240            ..
241        } => out.push((
242            Some(alias.as_ref().unwrap_or(qualifier).clone()),
243            name.clone(),
244            column_aliases.clone(),
245        )),
246        SourcePlan::Join {
247            left, right, alias, ..
248        } => {
249            if alias.is_some() {
250                *has_opaque_source = true;
251            } else {
252                collect_from_tables(left, out, has_opaque_source);
253                collect_from_tables(right, out, has_opaque_source);
254            }
255        }
256        _ => *has_opaque_source = true,
257    }
258}
259
260/// Reject silently-empty text searches up front: a match function whose
261/// field is not a real column, or is a column without a text index,
262/// previously returned zero rows with no diagnostic.
263pub fn validate_text_match_field(
264    catalog: &dyn TextMatchCatalog,
265    table: &str,
266    field: &str,
267    function_name: &str,
268) -> Result<(), SQLError> {
269    if !catalog
270        .has_table(table)
271        .map_err(|err| SQLError::Internal(format!("read table catalog: {err}")))?
272    {
273        return Err(SQLError::TypeMismatch(format!(
274            "{function_name}: unknown table `{table}`"
275        )));
276    }
277    let indexed = catalog
278        .indexed_fields(table)?
279        .iter()
280        .any(|fts| fts == field);
281    if !indexed {
282        if !catalog
283            .has_column(table, field)
284            .map_err(|err| SQLError::Internal(format!("read table schema: {err}")))?
285            && !catalog
286                .column_names(table)
287                .map_err(|err| SQLError::Internal(format!("read table schema: {err}")))?
288                .is_empty()
289        {
290            return Err(SQLError::TypeMismatch(format!(
291                "{function_name}: column `{field}` does not exist on table `{table}`"
292            )));
293        }
294        return Err(SQLError::TypeMismatch(format!(
295            "{function_name}: column `{table}.{field}` has no text index; \
296             create one with CREATE INDEX ... ON {table} USING gin ({field})"
297        )));
298    }
299    Ok(())
300}
301
302pub fn validate_text_match_all_fields(
303    catalog: &dyn TextMatchCatalog,
304    table: &str,
305    function_name: &str,
306) -> Result<(), SQLError> {
307    if !catalog
308        .has_table(table)
309        .map_err(|err| SQLError::Internal(format!("read table catalog: {err}")))?
310    {
311        return Err(SQLError::TypeMismatch(format!(
312            "{function_name}: unknown table `{table}`"
313        )));
314    }
315    if catalog.indexed_fields(table)?.is_empty() {
316        return Err(SQLError::TypeMismatch(format!(
317            "{function_name}: table `{table}` has no text-indexed columns; \
318             create one with CREATE INDEX ... ON {table} USING gin (...)"
319        )));
320    }
321    Ok(())
322}
323
324/// Validate the physical field using one retained catalog generation; indexed fields do not require a column-schema read.
325pub fn require_physical_text_index(
326    table: &str,
327    field: &str,
328    indexed_fields: &[String],
329    columns: impl FnOnce() -> Vec<crate::ast::ColumnDef>,
330) -> Result<(), SQLError> {
331    if indexed_fields.iter().any(|indexed| indexed == field) {
332        return Ok(());
333    }
334    let columns = columns();
335    if !columns.is_empty() && !columns.iter().any(|column| column.name == field) {
336        return Err(SQLError::UnknownColumn(field.to_string()));
337    }
338    Err(SQLError::TypeMismatch(format!(
339        "text search: column `{table}.{field}` has no text index; create one with CREATE INDEX ... ON {table} USING gin ({field})"
340    )))
341}
uqa_sql/semantics/text_indexes.rs

uqa_sql/semantics/
text_indexes.rs
