alopex_sql/distributed_read/
catalog_v0_8.rs

1//! Versioned, closed classifier for the v0.8 distributed-read SQL surface.
2//!
3//! Classification is intentionally conservative.  It records only the public
4//! shape that is eligible for a future normalized wire descriptor; it never
5//! serializes a [`LogicalPlan`](crate::planner::LogicalPlan) or uses a
6//! permissive "is query" predicate.
7
8use std::collections::BTreeSet;
9
10use serde::{Deserialize, Serialize};
11
12use crate::planner::{
13    AggregateFunction, LogicalPlan, Projection, TableReference, TableReferenceAccess, TypedExpr,
14    TypedExprKind,
15};
16
17/// Version identifier embedded in all accepted remote-read descriptors.
18pub const REMOTE_READ_CATALOG_VERSION: &str = "v0.8";
19
20/// Scalar function identities explicitly admitted to the v0.8 remote catalog.
21/// Adding a scalar signature elsewhere does not extend remote support until it
22/// is deliberately added here and appears in the public coverage matrix.
23pub const REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS: &[&str] = &[
24    "abs",
25    "sign",
26    "round",
27    "floor",
28    "ceil",
29    "ceiling",
30    "trunc",
31    "mod",
32    "power",
33    "pow",
34    "sqrt",
35    "exp",
36    "ln",
37    "log",
38    "log10",
39    "sin",
40    "cos",
41    "tan",
42    "asin",
43    "acos",
44    "atan",
45    "atan2",
46    "degrees",
47    "radians",
48    "pi",
49    "sha256",
50    "md5",
51    "simhash",
52    "hamming_distance",
53    "hex",
54    "unhex",
55    "encode",
56    "decode",
57    "length",
58    "char_length",
59    "octet_length",
60    "upper",
61    "lower",
62    "initcap",
63    "substr",
64    "left",
65    "right",
66    "trim",
67    "ltrim",
68    "rtrim",
69    "replace",
70    "instr",
71    "strpos",
72    "concat",
73    "concat_ws",
74    "repeat",
75    "reverse",
76    "lpad",
77    "rpad",
78    "split_part",
79    "regexp_replace",
80    "regexp_match",
81    "regexp_matches",
82    "coalesce",
83    "nullif",
84    "ifnull",
85    "iif",
86    "greatest",
87    "least",
88    "typeof",
89    "pg_typeof",
90    "quote",
91];
92
93/// Registered scalar identities intentionally excluded from remote execution.
94pub const REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS: &[&str] = &[
95    "vector_similarity",
96    "vector_distance",
97    "vector_dims",
98    "vector_norm",
99    "random",
100    "now",
101    "gen_random_uuid",
102    "uuidv7",
103    "memory_stats",
104    "io_stats",
105    "clear_cache",
106];
107
108/// A complete pre-routing classification for a planned SQL statement.
109#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
110pub enum RemoteReadClassification {
111    /// The plan belongs to the explicitly supported v0.8 catalog.
112    Supported(RemoteReadDescriptor),
113    /// The statement remains valid for the legacy local executor, but is not
114    /// eligible for remote/multi-range execution.
115    LocalOnly(RemoteReadRejection),
116    /// A cluster read request must fail before opening a transport session.
117    UnsupportedRemote(RemoteReadRejection),
118}
119
120/// Bounded descriptor metadata derived from an accepted logical plan.
121///
122/// This is deliberately not executable SQL and carries no private planner
123/// tree.  The later transport task expands it into an expression codec with a
124/// separate compatibility test.
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
126pub struct RemoteReadDescriptor {
127    pub catalog_version: String,
128    pub table: String,
129    pub shape: RemoteReadShape,
130    pub operators: RemoteReadOperators,
131}
132
133/// The closed high-level result shape accepted by the catalog.
134#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
135pub enum RemoteReadShape {
136    Rows,
137    Aggregate { aggregates: Vec<RemoteAggregate> },
138}
139
140/// Aggregate identities available to the v0.8 catalog.
141#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
142#[serde(rename_all = "snake_case")]
143pub enum RemoteAggregate {
144    Count,
145    Sum,
146    Total,
147    Avg,
148    Min,
149    Max,
150    GroupConcat,
151    StringAgg,
152}
153
154/// Closed modifiers which a later normalized descriptor must preserve.
155#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
156pub struct RemoteReadOperators {
157    pub filter: bool,
158    pub projection: bool,
159    pub order_by: bool,
160    pub limit: bool,
161    pub offset: bool,
162    pub group_by: bool,
163    pub having: bool,
164    pub deterministic_scalar: bool,
165    pub aggregate_distinct: bool,
166}
167
168/// Stable explanation for a non-supported remote classification.
169#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
170pub struct RemoteReadRejection {
171    pub code: String,
172    pub reason: String,
173}
174
175/// Public support status emitted by the coverage matrix.
176#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
177#[serde(rename_all = "snake_case")]
178pub enum RemoteReadCoverageStatus {
179    RemoteSupported,
180    LocalOnly,
181    PreExecutionRejection,
182}
183
184/// One stable public category row in the v0.8 coverage matrix.
185#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
186pub struct RemoteReadCoverageEntry {
187    pub id: &'static str,
188    pub public_surface: &'static str,
189    #[serde(skip_serializing)]
190    pub identities: &'static [&'static str],
191    pub remote_status: RemoteReadCoverageStatus,
192    pub prerequisite: &'static str,
193    pub normal_outcome: &'static str,
194    pub failure_outcome: &'static str,
195}
196
197impl RemoteReadRejection {
198    fn local_only(code: &str, reason: &str) -> RemoteReadClassification {
199        RemoteReadClassification::LocalOnly(Self {
200            code: code.to_string(),
201            reason: reason.to_string(),
202        })
203    }
204
205    fn unsupported(code: &str, reason: &str) -> RemoteReadClassification {
206        RemoteReadClassification::UnsupportedRemote(Self {
207            code: code.to_string(),
208            reason: reason.to_string(),
209        })
210    }
211}
212
213/// Closed v0.8 remote-read catalog.
214#[derive(Debug, Default, Clone, Copy)]
215pub struct RemoteReadCatalogV0_8;
216
217impl RemoteReadCatalogV0_8 {
218    /// Classifies a fully planned statement before routing or transport.
219    pub fn classify(
220        &self,
221        plan: &LogicalPlan,
222        table_references: &[TableReference],
223    ) -> RemoteReadClassification {
224        classify(plan, table_references)
225    }
226
227    /// Returns every public SQL category from the same closed catalog used by
228    /// the classifier.  No local feature is inferred as remotely supported.
229    pub fn coverage_entries(&self) -> Vec<RemoteReadCoverageEntry> {
230        coverage_entries()
231    }
232}
233
234/// Returns every public SQL category from the v0.8 closed catalog.
235pub fn coverage_entries() -> Vec<RemoteReadCoverageEntry> {
236    use RemoteReadCoverageStatus::{LocalOnly, PreExecutionRejection, RemoteSupported};
237
238    vec![
239        RemoteReadCoverageEntry {
240            id: "select.one_table.read_only",
241            public_surface: "one logical table SELECT with projection, WHERE, ORDER BY, LIMIT, OFFSET",
242            identities: &[
243                "select",
244                "projection",
245                "where",
246                "order_by",
247                "limit",
248                "offset",
249            ],
250            remote_status: RemoteSupported,
251            prerequisite: "closed catalog, fenced retained read point, authorized range targets",
252            normal_outcome: "prepared globally equivalent result",
253            failure_outcome: "classified pre-execution or routed-read failure; no local fallback",
254        },
255        RemoteReadCoverageEntry {
256            id: "select.aggregate.basic",
257            public_surface: "one-table COUNT, SUM, TOTAL, AVG, MIN, MAX with GROUP BY/HAVING/DISTINCT",
258            identities: &[
259                "count", "sum", "total", "avg", "min", "max", "group_by", "having", "distinct",
260            ],
261            remote_status: RemoteSupported,
262            prerequisite: "closed aggregate descriptor and global finalization budget",
263            normal_outcome: "prepared globally equivalent aggregate result",
264            failure_outcome: "classified pre-execution or global preparation failure",
265        },
266        RemoteReadCoverageEntry {
267            id: "select.aggregate.ordered_string",
268            public_surface: "one-table GROUP_CONCAT and STRING_AGG with global ordered replay",
269            identities: &["group_concat", "string_agg"],
270            remote_status: RemoteSupported,
271            prerequisite: "closed aggregate descriptor and ordered raw-value finalization budget",
272            normal_outcome: "prepared globally ordered aggregate result",
273            failure_outcome: "classified pre-execution or global preparation failure",
274        },
275        RemoteReadCoverageEntry {
276            id: "scalar.deterministic",
277            public_surface: "explicit deterministic scalar function list in RemoteReadCatalogV0_8",
278            identities: REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS,
279            remote_status: RemoteSupported,
280            prerequisite: "one-table SELECT and each function identity listed by the v0.8 catalog",
281            normal_outcome: "evaluated as part of a prepared remotely supported read",
282            failure_outcome: "unlisted function is rejected before transport",
283        },
284        RemoteReadCoverageEntry {
285            id: "scalar.local_only",
286            public_surface: "vector, random/UUID, statistics, and cache-control scalar functions",
287            identities: REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS,
288            remote_status: LocalOnly,
289            prerequisite: "local execution profile",
290            normal_outcome: "v0.7.4 local SQL behavior",
291            failure_outcome: "remote profile receives an explicit local-only classification",
292        },
293        RemoteReadCoverageEntry {
294            id: "statement.ddl",
295            public_surface: "CREATE/DROP TABLE and CREATE/DROP INDEX",
296            identities: &["create_table", "drop_table", "create_index", "drop_index"],
297            remote_status: PreExecutionRejection,
298            prerequisite: "local schema-management workflow",
299            normal_outcome: "v0.7.4 local SQL behavior",
300            failure_outcome: "ddl_not_supported_remote before transport",
301        },
302        RemoteReadCoverageEntry {
303            id: "statement.dml",
304            public_surface: "INSERT, UPDATE, DELETE",
305            identities: &["insert", "update", "delete"],
306            remote_status: PreExecutionRejection,
307            prerequisite: "local transaction workflow",
308            normal_outcome: "v0.7.4 local SQL behavior",
309            failure_outcome: "dml_not_supported_remote before transport",
310        },
311        RemoteReadCoverageEntry {
312            id: "statement.pragma",
313            public_surface: "PRAGMA",
314            identities: &["pragma"],
315            remote_status: LocalOnly,
316            prerequisite: "local execution profile",
317            normal_outcome: "v0.7.4 local SQL behavior",
318            failure_outcome: "pragma_local_only before transport",
319        },
320        RemoteReadCoverageEntry {
321            id: "relation.join",
322            public_surface: "JOIN",
323            identities: &[
324                "inner_join",
325                "left_join",
326                "right_join",
327                "full_join",
328                "cross_join",
329            ],
330            remote_status: PreExecutionRejection,
331            prerequisite: "local execution profile",
332            normal_outcome: "v0.7.4 local SQL behavior",
333            failure_outcome: "join_not_supported_remote before transport",
334        },
335        RemoteReadCoverageEntry {
336            id: "relation.subquery",
337            public_surface: "scalar, IN, EXISTS, and quantified subqueries",
338            identities: &["scalar_subquery", "in_subquery", "exists", "quantified"],
339            remote_status: PreExecutionRejection,
340            prerequisite: "local execution profile",
341            normal_outcome: "v0.7.4 local SQL behavior",
342            failure_outcome: "subquery_not_supported_remote before transport",
343        },
344        RemoteReadCoverageEntry {
345            id: "relation.compound_window",
346            public_surface: "compound and window query forms",
347            identities: &["compound_query", "window_expression"],
348            remote_status: PreExecutionRejection,
349            prerequisite: "a future remote catalog version",
350            normal_outcome: "not a v0.8 remote-read form",
351            failure_outcome: "function_not_in_remote_catalog before transport",
352        },
353        RemoteReadCoverageEntry {
354            id: "relation.recursive_cte",
355            public_surface: "recursive common table expressions",
356            identities: &["with_recursive"],
357            remote_status: PreExecutionRejection,
358            prerequisite: "local execution profile",
359            normal_outcome: "bounded fixed-point evaluation by the local executor",
360            failure_outcome: "recursive_cte_not_supported_remote before transport",
361        },
362        RemoteReadCoverageEntry {
363            id: "transaction.multi_statement",
364            public_surface: "existing multi-statement Transaction API workflow",
365            identities: &["transaction_api"],
366            remote_status: LocalOnly,
367            prerequisite: "local transaction workflow",
368            normal_outcome: "v0.7.4 local transaction behavior",
369            failure_outcome: "remote profile receives an explicit pre-execution classification",
370        },
371    ]
372}
373
374/// Classifies a fully planned statement before routing or transport.
375pub fn classify(
376    plan: &LogicalPlan,
377    table_references: &[TableReference],
378) -> RemoteReadClassification {
379    if let Some(classification) = table_boundary(table_references) {
380        return classification;
381    }
382    if plan.contains_join() {
383        return RemoteReadRejection::unsupported(
384            "join_not_supported_remote",
385            "JOIN is outside the v0.8 remote-read catalog",
386        );
387    }
388
389    let mut analysis = Analysis::default();
390    if let Err(rejection) = validate_plan(plan, &mut analysis) {
391        return rejection;
392    }
393    let table = match single_table(table_references) {
394        Some(table) => table,
395        None => {
396            return RemoteReadRejection::unsupported(
397                "single_logical_table_required",
398                "remote reads require exactly one physical logical table",
399            );
400        }
401    };
402    if analysis.scan_count != 1 {
403        return RemoteReadRejection::unsupported(
404            "single_logical_table_required",
405            "remote reads require exactly one table scan",
406        );
407    }
408    if analysis
409        .scan_tables
410        .first()
411        .is_none_or(|scan_table| scan_table != &table)
412    {
413        return RemoteReadRejection::unsupported(
414            "table_reference_mismatch",
415            "planned scan table does not match the routing table reference",
416        );
417    }
418
419    RemoteReadClassification::Supported(RemoteReadDescriptor {
420        catalog_version: REMOTE_READ_CATALOG_VERSION.to_string(),
421        table,
422        shape: if analysis.aggregates.is_empty() {
423            RemoteReadShape::Rows
424        } else {
425            RemoteReadShape::Aggregate {
426                aggregates: analysis.aggregates,
427            }
428        },
429        operators: analysis.operators,
430    })
431}
432
433fn table_boundary(table_references: &[TableReference]) -> Option<RemoteReadClassification> {
434    if table_references
435        .iter()
436        .any(|reference| reference.access != TableReferenceAccess::Read)
437    {
438        return Some(RemoteReadRejection::unsupported(
439            "read_only_select_required",
440            "remote reads require a read-only SELECT",
441        ));
442    }
443    let tables: BTreeSet<_> = table_references
444        .iter()
445        .map(|reference| reference.table_name.as_str())
446        .collect();
447    if tables.len() > 1 {
448        return Some(RemoteReadRejection::unsupported(
449            "single_logical_table_required",
450            "remote reads cannot span multiple logical tables",
451        ));
452    }
453    None
454}
455
456fn single_table(table_references: &[TableReference]) -> Option<String> {
457    table_references
458        .first()
459        .map(|reference| reference.table_name.clone())
460}
461
462#[derive(Debug, Default)]
463struct Analysis {
464    scan_count: usize,
465    scan_tables: Vec<String>,
466    aggregates: Vec<RemoteAggregate>,
467    operators: RemoteReadOperators,
468}
469
470fn validate_plan(
471    plan: &LogicalPlan,
472    analysis: &mut Analysis,
473) -> Result<(), RemoteReadClassification> {
474    match plan {
475        LogicalPlan::Pragma { .. } => Err(RemoteReadRejection::local_only(
476            "pragma_local_only",
477            "PRAGMA remains available only to the local executor",
478        )),
479        LogicalPlan::Insert { .. }
480        | LogicalPlan::InsertSelect { .. }
481        | LogicalPlan::Update { .. }
482        | LogicalPlan::Delete { .. } => Err(RemoteReadRejection::unsupported(
483            "dml_not_supported_remote",
484            "DML is outside the read-only remote-read catalog",
485        )),
486        LogicalPlan::CreateTable { .. }
487        | LogicalPlan::DropTable { .. }
488        | LogicalPlan::CreateIndex { .. }
489        | LogicalPlan::DropIndex { .. } => Err(RemoteReadRejection::unsupported(
490            "ddl_not_supported_remote",
491            "DDL is outside the read-only remote-read catalog",
492        )),
493        LogicalPlan::Join { .. } => Err(RemoteReadRejection::unsupported(
494            "join_not_supported_remote",
495            "JOIN is outside the v0.8 remote-read catalog",
496        )),
497        LogicalPlan::Window { .. } => Err(RemoteReadRejection::unsupported(
498            "window_not_supported_remote",
499            "window functions are outside the v0.8 remote-read catalog",
500        )),
501        LogicalPlan::SetOperation { .. } => Err(RemoteReadRejection::unsupported(
502            "set_operation_not_supported_remote",
503            "set operations are outside the v0.8 remote-read catalog",
504        )),
505        LogicalPlan::RecursiveCte { .. } | LogicalPlan::RecursiveReference { .. } => {
506            Err(RemoteReadRejection::unsupported(
507                "recursive_cte_not_supported_remote",
508                "recursive common table expressions are outside the v0.8 remote-read catalog",
509            ))
510        }
511        LogicalPlan::Scan { table, projection } => {
512            analysis.scan_count += 1;
513            analysis.scan_tables.push(table.clone());
514            validate_projection(projection, false, analysis)
515        }
516        LogicalPlan::Filter { input, predicate } => {
517            analysis.operators.filter = true;
518            validate_expr(predicate, false, analysis)?;
519            validate_plan(input, analysis)
520        }
521        LogicalPlan::Project { input, projection } => {
522            validate_plan(input, analysis)?;
523            analysis.operators.projection = true;
524            validate_projection(projection, !analysis.aggregates.is_empty(), analysis)
525        }
526        LogicalPlan::Aggregate {
527            input,
528            group_keys,
529            aggregates,
530            having,
531            projection,
532        } => {
533            analysis.operators.group_by = !group_keys.is_empty();
534            analysis.operators.having = having.is_some();
535            for group_key in group_keys {
536                validate_expr(group_key, false, analysis)?;
537            }
538            for aggregate in aggregates {
539                let aggregate_name = remote_aggregate(&aggregate.function);
540                analysis.operators.aggregate_distinct |= aggregate.distinct;
541                if let Some(argument) = &aggregate.arg {
542                    validate_expr(argument, false, analysis)?;
543                }
544                analysis.aggregates.push(aggregate_name);
545            }
546            if let Some(having) = having {
547                validate_expr(having, true, analysis)?;
548            }
549            validate_projection(projection, true, analysis)?;
550            validate_plan(input, analysis)
551        }
552        LogicalPlan::Sort { input, order_by } => {
553            validate_plan(input, analysis)?;
554            analysis.operators.order_by = true;
555            for sort in order_by {
556                validate_expr(sort.expr(), !analysis.aggregates.is_empty(), analysis)?;
557            }
558            Ok(())
559        }
560        LogicalPlan::Limit {
561            input,
562            limit,
563            offset,
564        } => {
565            analysis.operators.limit |= limit.is_some();
566            analysis.operators.offset |= offset.is_some();
567            validate_plan(input, analysis)
568        }
569    }
570}
571
572fn validate_projection(
573    projection: &Projection,
574    allow_aggregate: bool,
575    analysis: &mut Analysis,
576) -> Result<(), RemoteReadClassification> {
577    if let Projection::Columns(columns) = projection {
578        analysis.operators.projection = true;
579        for column in columns {
580            validate_expr(&column.expr, allow_aggregate, analysis)?;
581        }
582    }
583    Ok(())
584}
585
586fn validate_expr(
587    expr: &TypedExpr,
588    allow_aggregate: bool,
589    analysis: &mut Analysis,
590) -> Result<(), RemoteReadClassification> {
591    match &expr.kind {
592        TypedExprKind::Literal(_) | TypedExprKind::ColumnRef { .. } => Ok(()),
593        TypedExprKind::VectorLiteral(_) => Err(RemoteReadRejection::local_only(
594            "vector_sql_local_only",
595            "vector SQL is not in the v0.8 remote-read catalog",
596        )),
597        TypedExprKind::BinaryOp { left, right, .. } => {
598            validate_expr(left, allow_aggregate, analysis)?;
599            validate_expr(right, allow_aggregate, analysis)
600        }
601        TypedExprKind::UnaryOp { operand, .. }
602        | TypedExprKind::Cast { expr: operand, .. }
603        | TypedExprKind::IsNull { expr: operand, .. } => {
604            validate_expr(operand, allow_aggregate, analysis)
605        }
606        TypedExprKind::Case {
607            operand,
608            branches,
609            else_expr,
610        } => {
611            if let Some(operand) = operand {
612                validate_expr(operand, allow_aggregate, analysis)?;
613            }
614            for branch in branches {
615                validate_expr(&branch.when, allow_aggregate, analysis)?;
616                validate_expr(&branch.then, allow_aggregate, analysis)?;
617            }
618            if let Some(else_expr) = else_expr {
619                validate_expr(else_expr, allow_aggregate, analysis)?;
620            }
621            Ok(())
622        }
623        TypedExprKind::Between {
624            expr, low, high, ..
625        } => {
626            validate_expr(expr, allow_aggregate, analysis)?;
627            validate_expr(low, allow_aggregate, analysis)?;
628            validate_expr(high, allow_aggregate, analysis)
629        }
630        TypedExprKind::Like {
631            expr,
632            pattern,
633            escape,
634            ..
635        } => {
636            validate_expr(expr, allow_aggregate, analysis)?;
637            validate_expr(pattern, allow_aggregate, analysis)?;
638            if let Some(escape) = escape {
639                validate_expr(escape, allow_aggregate, analysis)?;
640            }
641            Ok(())
642        }
643        TypedExprKind::InList { expr, list, .. } => {
644            validate_expr(expr, allow_aggregate, analysis)?;
645            for item in list {
646                validate_expr(item, allow_aggregate, analysis)?;
647            }
648            Ok(())
649        }
650        TypedExprKind::FunctionCall { name, args, .. } => {
651            if allow_aggregate && aggregate_function_name(name) {
652                for argument in args {
653                    validate_expr(argument, false, analysis)?;
654                }
655                return Ok(());
656            }
657            let normalized = name.to_ascii_lowercase();
658            if REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
659                return Err(RemoteReadRejection::local_only(
660                    "stateful_function_local_only",
661                    "the requested scalar function remains local-only",
662                ));
663            }
664            if !REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
665                return Err(RemoteReadRejection::unsupported(
666                    "function_not_in_remote_catalog",
667                    "function is not explicitly listed in the remote-read catalog",
668                ));
669            }
670            analysis.operators.deterministic_scalar = true;
671            for argument in args {
672                validate_expr(argument, false, analysis)?;
673            }
674            Ok(())
675        }
676        TypedExprKind::ScalarSubquery(_)
677        | TypedExprKind::InSubquery { .. }
678        | TypedExprKind::Exists { .. }
679        | TypedExprKind::Quantified { .. } => Err(RemoteReadRejection::unsupported(
680            "subquery_not_supported_remote",
681            "subqueries are outside the v0.8 remote-read catalog",
682        )),
683    }
684}
685
686fn remote_aggregate(function: &AggregateFunction) -> RemoteAggregate {
687    match function {
688        AggregateFunction::Count => RemoteAggregate::Count,
689        AggregateFunction::Sum => RemoteAggregate::Sum,
690        AggregateFunction::Total => RemoteAggregate::Total,
691        AggregateFunction::Avg => RemoteAggregate::Avg,
692        AggregateFunction::Min => RemoteAggregate::Min,
693        AggregateFunction::Max => RemoteAggregate::Max,
694        AggregateFunction::GroupConcat { .. } => RemoteAggregate::GroupConcat,
695        AggregateFunction::StringAgg { .. } => RemoteAggregate::StringAgg,
696    }
697}
698
699fn aggregate_function_name(name: &str) -> bool {
700    matches!(
701        name.to_ascii_lowercase().as_str(),
702        "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
703    )
704}
705
706trait SortExprExt {
707    fn expr(&self) -> &TypedExpr;
708}
709
710impl SortExprExt for crate::planner::SortExpr {
711    fn expr(&self) -> &TypedExpr {
712        &self.expr
713    }
714}
715
716#[cfg(test)]
717mod tests {
718    use super::*;
719    use crate::Span;
720    use crate::ast::expr::Literal;
721    use crate::catalog::ColumnMetadata;
722    use crate::planner::{Projection, RecursiveCteLimits, ResolvedType, SortExpr, TypedExpr};
723
724    fn references() -> Vec<TableReference> {
725        vec![TableReference::new(
726            "users",
727            TableReferenceAccess::Read,
728            crate::planner::TableReferenceSource::LogicalPlanScan,
729        )]
730    }
731
732    fn scan() -> LogicalPlan {
733        LogicalPlan::scan("users".to_string(), Projection::All(vec!["id".to_string()]))
734    }
735
736    fn column() -> TypedExpr {
737        TypedExpr::column_ref(
738            "users".to_string(),
739            "id".to_string(),
740            0,
741            ResolvedType::Integer,
742            Span::default(),
743        )
744    }
745
746    #[test]
747    fn classifies_closed_single_table_read_shape() {
748        let plan = LogicalPlan::limit(
749            LogicalPlan::sort(
750                LogicalPlan::filter(scan(), column()),
751                vec![SortExpr::asc(column())],
752            ),
753            Some(10),
754            Some(3),
755        );
756
757        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
758            panic!("single-table deterministic read must be accepted");
759        };
760        assert_eq!(descriptor.catalog_version, REMOTE_READ_CATALOG_VERSION);
761        assert_eq!(descriptor.table, "users");
762        assert_eq!(descriptor.shape, RemoteReadShape::Rows);
763        assert!(descriptor.operators.filter);
764        assert!(descriptor.operators.order_by);
765        assert!(descriptor.operators.limit);
766        assert!(descriptor.operators.offset);
767    }
768
769    #[test]
770    fn rejects_join_and_dml_before_transport() {
771        let join = LogicalPlan::join(scan(), scan(), crate::planner::JoinType::Inner, None, None);
772        assert!(matches!(
773            classify(&join, &references()),
774            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
775                if code == "join_not_supported_remote"
776        ));
777
778        let insert = LogicalPlan::insert("users".to_string(), vec!["id".to_string()], vec![]);
779        assert!(matches!(
780            classify(&insert, &references()),
781            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
782                if code == "dml_not_supported_remote"
783        ));
784    }
785
786    #[test]
787    fn rejects_ddl_and_pragma_before_a_remote_session_exists() {
788        let ddl = LogicalPlan::drop_table("users".to_string(), false);
789        assert!(matches!(
790            classify(&ddl, &[]),
791            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
792                if code == "ddl_not_supported_remote"
793        ));
794
795        let pragma = LogicalPlan::Pragma {
796            name: "cache_size".to_string(),
797            value: None,
798        };
799        assert!(matches!(
800            classify(&pragma, &[]),
801            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
802                if code == "pragma_local_only"
803        ));
804    }
805
806    #[test]
807    fn recursive_cte_is_rejected_before_remote_transport() {
808        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
809        let reference = LogicalPlan::RecursiveReference {
810            name: "counter".to_string(),
811            schema: schema.clone(),
812        };
813        let recursive = LogicalPlan::RecursiveCte {
814            name: "counter".to_string(),
815            anchor: Box::new(scan()),
816            recursive_term: Box::new(reference.clone()),
817            union_all: true,
818            schema,
819            limits: RecursiveCteLimits::default(),
820        };
821
822        for plan in [&recursive, &reference] {
823            assert!(matches!(
824                classify(plan, &references()),
825                RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, reason })
826                    if code == "recursive_cte_not_supported_remote"
827                        && reason.contains("outside the v0.8 remote-read catalog")
828            ));
829        }
830    }
831
832    #[test]
833    fn stateful_and_vector_expressions_remain_local_only() {
834        let random = TypedExpr::function_call(
835            "random".to_string(),
836            vec![],
837            false,
838            false,
839            ResolvedType::Double,
840            Span::default(),
841        );
842        let random_plan = LogicalPlan::filter(scan(), random);
843        assert!(matches!(
844            classify(&random_plan, &references()),
845            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
846                if code == "stateful_function_local_only"
847        ));
848
849        let vector_plan = LogicalPlan::filter(
850            scan(),
851            TypedExpr::vector_literal(vec![1.0, 2.0], 2, Span::default()),
852        );
853        assert!(matches!(
854            classify(&vector_plan, &references()),
855            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
856                if code == "vector_sql_local_only"
857        ));
858    }
859
860    #[test]
861    fn subqueries_are_rejected_and_descriptor_never_contains_plan() {
862        let subquery = TypedExpr::new(
863            TypedExprKind::ScalarSubquery(Box::new(scan())),
864            ResolvedType::Integer,
865            Span::default(),
866        );
867        assert!(matches!(
868            classify(&LogicalPlan::filter(scan(), subquery), &references()),
869            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
870                if code == "subquery_not_supported_remote"
871        ));
872
873        let encoded = serde_json::to_string(&classify(&scan(), &references())).unwrap();
874        assert!(!encoded.contains("LogicalPlan"));
875        assert!(!encoded.contains("column_index"));
876    }
877
878    #[test]
879    fn aggregate_catalog_includes_string_aggregates() {
880        let aggregate = crate::planner::AggregateExpr {
881            function: AggregateFunction::StringAgg {
882                separator: Some(",".to_string()),
883            },
884            arg: Some(column()),
885            distinct: true,
886            result_type: ResolvedType::Text,
887        };
888        let plan = LogicalPlan::aggregate(
889            scan(),
890            vec![column()],
891            vec![aggregate],
892            Some(TypedExpr::literal(
893                Literal::Boolean(true),
894                ResolvedType::Boolean,
895                Span::default(),
896            )),
897            Projection::All(vec![]),
898        );
899        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
900            panic!("listed aggregate must be accepted");
901        };
902        assert_eq!(
903            descriptor.shape,
904            RemoteReadShape::Aggregate {
905                aggregates: vec![RemoteAggregate::StringAgg]
906            }
907        );
908        assert!(descriptor.operators.group_by);
909        assert!(descriptor.operators.having);
910        assert!(descriptor.operators.aggregate_distinct);
911    }
912}
alopex_sql/distributed_read/catalog_v0_8.rs

alopex_sql/distributed_read/
catalog_v0_8.rs
