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: "scalar.standard_predicate",
355            public_surface: "truth, distinctness, and row-value predicates",
356            identities: &["truth_predicate", "is_distinct_from", "row_comparison"],
357            remote_status: LocalOnly,
358            prerequisite: "local execution profile",
359            normal_outcome: "three-valued evaluation by the local executor",
360            failure_outcome: "standard_predicate_local_only before transport",
361        },
362        RemoteReadCoverageEntry {
363            id: "scalar.try_cast",
364            public_surface: "TRY_CAST safe conversion",
365            identities: &["try_cast"],
366            remote_status: LocalOnly,
367            prerequisite: "local execution profile",
368            normal_outcome: "NULL-on-conversion-failure evaluation by the local executor",
369            failure_outcome: "try_cast_local_only before transport",
370        },
371        RemoteReadCoverageEntry {
372            id: "pagination.fetch_with_ties",
373            public_surface: "FETCH ... WITH TIES peer-preserving row limits",
374            identities: &["fetch_with_ties"],
375            remote_status: LocalOnly,
376            prerequisite: "local execution profile",
377            normal_outcome: "peer-preserving limit evaluation by the local executor",
378            failure_outcome: "fetch_with_ties_local_only before transport",
379        },
380        RemoteReadCoverageEntry {
381            id: "relation.distinct_on",
382            public_surface: "SELECT DISTINCT ON deterministic first-row deduplication",
383            identities: &["distinct_on"],
384            remote_status: LocalOnly,
385            prerequisite: "local execution profile",
386            normal_outcome: "deterministic per-key first-row evaluation by the local executor",
387            failure_outcome: "distinct_on_local_only before transport",
388        },
389        RemoteReadCoverageEntry {
390            id: "scalar.aggregate_filter",
391            public_surface: "aggregate FILTER (WHERE ...) per-aggregate row filtering",
392            identities: &["aggregate_filter"],
393            remote_status: LocalOnly,
394            prerequisite: "local execution profile",
395            normal_outcome: "per-aggregate predicate filtering by the local executor",
396            failure_outcome: "aggregate_filter_local_only before transport",
397        },
398        RemoteReadCoverageEntry {
399            id: "scalar.ordered_aggregate",
400            public_surface: "aggregate-local ORDER BY and WITHIN GROUP ordered-set aggregates",
401            identities: &["aggregate_order_by", "within_group", "percentile_disc"],
402            remote_status: LocalOnly,
403            prerequisite: "local execution profile",
404            normal_outcome: "ordered aggregate evaluation by the local executor",
405            failure_outcome: "ordered_aggregate_local_only before transport",
406        },
407        RemoteReadCoverageEntry {
408            id: "aggregate.grouping_sets",
409            public_surface: "GROUPING SETS / ROLLUP / CUBE multi-set aggregation",
410            identities: &["rollup", "cube", "grouping_sets", "grouping", "grouping_id"],
411            remote_status: LocalOnly,
412            prerequisite: "local execution profile",
413            normal_outcome: "single-pass multi-set aggregation by the local executor",
414            failure_outcome: "grouping_sets_local_only before transport",
415        },
416        RemoteReadCoverageEntry {
417            id: "relation.lateral_join",
418            public_surface: "LATERAL joins over a correlated relation",
419            identities: &["lateral", "cross_join_lateral", "left_join_lateral"],
420            remote_status: PreExecutionRejection,
421            prerequisite: "local execution profile",
422            normal_outcome: "per-left-row correlated evaluation by the local executor",
423            failure_outcome: "lateral_join_not_supported_remote before transport",
424        },
425        RemoteReadCoverageEntry {
426            id: "relation.table_function",
427            public_surface: "FROM-clause table functions",
428            identities: &["unnest", "generate_series"],
429            remote_status: LocalOnly,
430            prerequisite: "local execution profile",
431            normal_outcome: "row generation by the local executor",
432            failure_outcome: "table_function_not_supported_remote before transport",
433        },
434        RemoteReadCoverageEntry {
435            id: "relation.recursive_cte",
436            public_surface: "recursive common table expressions",
437            identities: &["with_recursive"],
438            remote_status: PreExecutionRejection,
439            prerequisite: "local execution profile",
440            normal_outcome: "bounded fixed-point evaluation by the local executor",
441            failure_outcome: "recursive_cte_not_supported_remote before transport",
442        },
443        RemoteReadCoverageEntry {
444            id: "transaction.multi_statement",
445            public_surface: "existing multi-statement Transaction API workflow",
446            identities: &["transaction_api"],
447            remote_status: LocalOnly,
448            prerequisite: "local transaction workflow",
449            normal_outcome: "v0.7.4 local transaction behavior",
450            failure_outcome: "remote profile receives an explicit pre-execution classification",
451        },
452    ]
453}
454
455/// Classifies a fully planned statement before routing or transport.
456pub fn classify(
457    plan: &LogicalPlan,
458    table_references: &[TableReference],
459) -> RemoteReadClassification {
460    if let Some(classification) = table_boundary(table_references) {
461        return classification;
462    }
463    if plan.contains_join() {
464        return RemoteReadRejection::unsupported(
465            "join_not_supported_remote",
466            "JOIN is outside the v0.8 remote-read catalog",
467        );
468    }
469
470    let mut analysis = Analysis::default();
471    if let Err(rejection) = validate_plan(plan, &mut analysis) {
472        return rejection;
473    }
474    let table = match single_table(table_references) {
475        Some(table) => table,
476        None => {
477            return RemoteReadRejection::unsupported(
478                "single_logical_table_required",
479                "remote reads require exactly one physical logical table",
480            );
481        }
482    };
483    if analysis.scan_count != 1 {
484        return RemoteReadRejection::unsupported(
485            "single_logical_table_required",
486            "remote reads require exactly one table scan",
487        );
488    }
489    if analysis
490        .scan_tables
491        .first()
492        .is_none_or(|scan_table| scan_table != &table)
493    {
494        return RemoteReadRejection::unsupported(
495            "table_reference_mismatch",
496            "planned scan table does not match the routing table reference",
497        );
498    }
499
500    RemoteReadClassification::Supported(RemoteReadDescriptor {
501        catalog_version: REMOTE_READ_CATALOG_VERSION.to_string(),
502        table,
503        shape: if analysis.aggregates.is_empty() {
504            RemoteReadShape::Rows
505        } else {
506            RemoteReadShape::Aggregate {
507                aggregates: analysis.aggregates,
508            }
509        },
510        operators: analysis.operators,
511    })
512}
513
514fn table_boundary(table_references: &[TableReference]) -> Option<RemoteReadClassification> {
515    if table_references
516        .iter()
517        .any(|reference| reference.access != TableReferenceAccess::Read)
518    {
519        return Some(RemoteReadRejection::unsupported(
520            "read_only_select_required",
521            "remote reads require a read-only SELECT",
522        ));
523    }
524    let tables: BTreeSet<_> = table_references
525        .iter()
526        .map(|reference| reference.table_name.as_str())
527        .collect();
528    if tables.len() > 1 {
529        return Some(RemoteReadRejection::unsupported(
530            "single_logical_table_required",
531            "remote reads cannot span multiple logical tables",
532        ));
533    }
534    None
535}
536
537fn single_table(table_references: &[TableReference]) -> Option<String> {
538    table_references
539        .first()
540        .map(|reference| reference.table_name.clone())
541}
542
543#[derive(Debug, Default)]
544struct Analysis {
545    scan_count: usize,
546    scan_tables: Vec<String>,
547    aggregates: Vec<RemoteAggregate>,
548    operators: RemoteReadOperators,
549}
550
551fn validate_plan(
552    plan: &LogicalPlan,
553    analysis: &mut Analysis,
554) -> Result<(), RemoteReadClassification> {
555    match plan {
556        LogicalPlan::Pragma { .. } => Err(RemoteReadRejection::local_only(
557            "pragma_local_only",
558            "PRAGMA remains available only to the local executor",
559        )),
560        LogicalPlan::Insert { .. }
561        | LogicalPlan::InsertSelect { .. }
562        | LogicalPlan::Update { .. }
563        | LogicalPlan::Delete { .. } => Err(RemoteReadRejection::unsupported(
564            "dml_not_supported_remote",
565            "DML is outside the read-only remote-read catalog",
566        )),
567        LogicalPlan::CreateTable { .. }
568        | LogicalPlan::DropTable { .. }
569        | LogicalPlan::CreateIndex { .. }
570        | LogicalPlan::DropIndex { .. } => Err(RemoteReadRejection::unsupported(
571            "ddl_not_supported_remote",
572            "DDL is outside the read-only remote-read catalog",
573        )),
574        LogicalPlan::Join { .. } => Err(RemoteReadRejection::unsupported(
575            "join_not_supported_remote",
576            "JOIN is outside the v0.8 remote-read catalog",
577        )),
578        LogicalPlan::LateralJoin { .. } => Err(RemoteReadRejection::unsupported(
579            "lateral_join_not_supported_remote",
580            "LATERAL joins are outside the v0.8 remote-read catalog",
581        )),
582        LogicalPlan::TableFunction { .. } => Err(RemoteReadRejection::local_only(
583            "table_function_not_supported_remote",
584            "FROM-clause table functions are evaluated by the local executor",
585        )),
586        LogicalPlan::Window { .. } => Err(RemoteReadRejection::unsupported(
587            "window_not_supported_remote",
588            "window functions are outside the v0.8 remote-read catalog",
589        )),
590        LogicalPlan::SetOperation { .. } => Err(RemoteReadRejection::unsupported(
591            "set_operation_not_supported_remote",
592            "set operations are outside the v0.8 remote-read catalog",
593        )),
594        LogicalPlan::Values { .. } => Err(RemoteReadRejection::local_only(
595            "values_local_only",
596            "VALUES relations are evaluated by the local executor",
597        )),
598        LogicalPlan::RecursiveCte { .. } | LogicalPlan::RecursiveReference { .. } => {
599            Err(RemoteReadRejection::unsupported(
600                "recursive_cte_not_supported_remote",
601                "recursive common table expressions are outside the v0.8 remote-read catalog",
602            ))
603        }
604        LogicalPlan::Scan { table, projection } => {
605            analysis.scan_count += 1;
606            analysis.scan_tables.push(table.clone());
607            validate_projection(projection, false, analysis)
608        }
609        LogicalPlan::Filter { input, predicate } => {
610            analysis.operators.filter = true;
611            validate_expr(predicate, false, analysis)?;
612            validate_plan(input, analysis)
613        }
614        LogicalPlan::Project { input, projection } => {
615            validate_plan(input, analysis)?;
616            analysis.operators.projection = true;
617            validate_projection(projection, !analysis.aggregates.is_empty(), analysis)
618        }
619        LogicalPlan::Aggregate {
620            input,
621            group_keys,
622            aggregates,
623            having,
624            projection,
625            grouping_sets,
626        } => {
627            if grouping_sets.is_some() {
628                return Err(RemoteReadRejection::local_only(
629                    "grouping_sets_local_only",
630                    "GROUPING SETS/ROLLUP/CUBE aggregation is not in the v0.8 remote-read catalog",
631                ));
632            }
633            analysis.operators.group_by = !group_keys.is_empty();
634            analysis.operators.having = having.is_some();
635            for group_key in group_keys {
636                validate_expr(group_key, false, analysis)?;
637            }
638            for aggregate in aggregates {
639                if aggregate.filter.is_some() {
640                    return Err(RemoteReadRejection::local_only(
641                        "aggregate_filter_local_only",
642                        "aggregate FILTER (WHERE ...) is not in the v0.8 remote-read catalog",
643                    ));
644                }
645                let Some(aggregate_name) = remote_aggregate(&aggregate.function) else {
646                    return Err(RemoteReadRejection::local_only(
647                        "ordered_aggregate_local_only",
648                        "ordered-set aggregates are not in the v0.8 remote-read catalog",
649                    ));
650                };
651                if !aggregate.order_by.is_empty() {
652                    return Err(RemoteReadRejection::local_only(
653                        "ordered_aggregate_local_only",
654                        "aggregate-local ORDER BY is not in the v0.8 remote-read catalog",
655                    ));
656                }
657                analysis.operators.aggregate_distinct |= aggregate.distinct;
658                if let Some(argument) = &aggregate.arg {
659                    validate_expr(argument, false, analysis)?;
660                }
661                analysis.aggregates.push(aggregate_name);
662            }
663            if let Some(having) = having {
664                validate_expr(having, true, analysis)?;
665            }
666            validate_projection(projection, true, analysis)?;
667            validate_plan(input, analysis)
668        }
669        LogicalPlan::Sort { input, order_by } => {
670            validate_plan(input, analysis)?;
671            analysis.operators.order_by = true;
672            for sort in order_by {
673                validate_expr(sort.expr(), !analysis.aggregates.is_empty(), analysis)?;
674            }
675            Ok(())
676        }
677        LogicalPlan::Limit {
678            input,
679            limit,
680            offset,
681            ties,
682        } => {
683            if ties.is_some() {
684                return Err(RemoteReadRejection::local_only(
685                    "fetch_with_ties_local_only",
686                    "FETCH ... WITH TIES is not in the v0.8 remote-read catalog",
687                ));
688            }
689            analysis.operators.limit |= limit.is_some();
690            analysis.operators.offset |= offset.is_some();
691            validate_plan(input, analysis)
692        }
693        LogicalPlan::DistinctOn { .. } => Err(RemoteReadRejection::local_only(
694            "distinct_on_local_only",
695            "SELECT DISTINCT ON is not in the v0.8 remote-read catalog",
696        )),
697    }
698}
699
700fn validate_projection(
701    projection: &Projection,
702    allow_aggregate: bool,
703    analysis: &mut Analysis,
704) -> Result<(), RemoteReadClassification> {
705    if let Projection::Columns(columns) = projection {
706        analysis.operators.projection = true;
707        for column in columns {
708            validate_expr(&column.expr, allow_aggregate, analysis)?;
709        }
710    }
711    Ok(())
712}
713
714fn validate_expr(
715    expr: &TypedExpr,
716    allow_aggregate: bool,
717    analysis: &mut Analysis,
718) -> Result<(), RemoteReadClassification> {
719    match &expr.kind {
720        TypedExprKind::Literal(_) | TypedExprKind::ColumnRef { .. } => Ok(()),
721        TypedExprKind::VectorLiteral(_) => Err(RemoteReadRejection::local_only(
722            "vector_sql_local_only",
723            "vector SQL is not in the v0.8 remote-read catalog",
724        )),
725        TypedExprKind::TryCast { .. } => Err(RemoteReadRejection::local_only(
726            "try_cast_local_only",
727            "TRY_CAST is not in the v0.8 remote-read catalog",
728        )),
729        TypedExprKind::BinaryOp { left, right, .. } => {
730            validate_expr(left, allow_aggregate, analysis)?;
731            validate_expr(right, allow_aggregate, analysis)
732        }
733        TypedExprKind::UnaryOp { operand, .. }
734        | TypedExprKind::Cast { expr: operand, .. }
735        | TypedExprKind::IsNull { expr: operand, .. } => {
736            validate_expr(operand, allow_aggregate, analysis)
737        }
738        TypedExprKind::Case {
739            operand,
740            branches,
741            else_expr,
742        } => {
743            if let Some(operand) = operand {
744                validate_expr(operand, allow_aggregate, analysis)?;
745            }
746            for branch in branches {
747                validate_expr(&branch.when, allow_aggregate, analysis)?;
748                validate_expr(&branch.then, allow_aggregate, analysis)?;
749            }
750            if let Some(else_expr) = else_expr {
751                validate_expr(else_expr, allow_aggregate, analysis)?;
752            }
753            Ok(())
754        }
755        TypedExprKind::Between {
756            expr, low, high, ..
757        } => {
758            validate_expr(expr, allow_aggregate, analysis)?;
759            validate_expr(low, allow_aggregate, analysis)?;
760            validate_expr(high, allow_aggregate, analysis)
761        }
762        TypedExprKind::Like {
763            expr,
764            pattern,
765            escape,
766            ..
767        } => {
768            validate_expr(expr, allow_aggregate, analysis)?;
769            validate_expr(pattern, allow_aggregate, analysis)?;
770            if let Some(escape) = escape {
771                validate_expr(escape, allow_aggregate, analysis)?;
772            }
773            Ok(())
774        }
775        TypedExprKind::InList { expr, list, .. } => {
776            validate_expr(expr, allow_aggregate, analysis)?;
777            for item in list {
778                validate_expr(item, allow_aggregate, analysis)?;
779            }
780            Ok(())
781        }
782        TypedExprKind::FunctionCall { name, args, .. } => {
783            if allow_aggregate && aggregate_function_name(name) {
784                for argument in args {
785                    validate_expr(argument, false, analysis)?;
786                }
787                return Ok(());
788            }
789            let normalized = name.to_ascii_lowercase();
790            if normalized.starts_with("__alopex_truth_") || normalized.starts_with("__alopex_row_")
791            {
792                return Err(RemoteReadRejection::local_only(
793                    "standard_predicate_local_only",
794                    "standard predicates are not in the v0.8 remote-read catalog",
795                ));
796            }
797            if REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
798                return Err(RemoteReadRejection::local_only(
799                    "stateful_function_local_only",
800                    "the requested scalar function remains local-only",
801                ));
802            }
803            if !REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
804                return Err(RemoteReadRejection::unsupported(
805                    "function_not_in_remote_catalog",
806                    "function is not explicitly listed in the remote-read catalog",
807                ));
808            }
809            analysis.operators.deterministic_scalar = true;
810            for argument in args {
811                validate_expr(argument, false, analysis)?;
812            }
813            Ok(())
814        }
815        TypedExprKind::ScalarSubquery(_)
816        | TypedExprKind::InSubquery { .. }
817        | TypedExprKind::Exists { .. }
818        | TypedExprKind::Quantified { .. } => Err(RemoteReadRejection::unsupported(
819            "subquery_not_supported_remote",
820            "subqueries are outside the v0.8 remote-read catalog",
821        )),
822    }
823}
824
825/// Aggregate identities admitted to the closed v0.8 catalog. Ordered-set
826/// aggregates (PERCENTILE_DISC, issue #148) return `None` and classify as
827/// `ordered_aggregate_local_only`.
828fn remote_aggregate(function: &AggregateFunction) -> Option<RemoteAggregate> {
829    match function {
830        AggregateFunction::Count => Some(RemoteAggregate::Count),
831        AggregateFunction::Sum => Some(RemoteAggregate::Sum),
832        AggregateFunction::Total => Some(RemoteAggregate::Total),
833        AggregateFunction::Avg => Some(RemoteAggregate::Avg),
834        AggregateFunction::Min => Some(RemoteAggregate::Min),
835        AggregateFunction::Max => Some(RemoteAggregate::Max),
836        AggregateFunction::GroupConcat { .. } => Some(RemoteAggregate::GroupConcat),
837        AggregateFunction::StringAgg { .. } => Some(RemoteAggregate::StringAgg),
838        AggregateFunction::PercentileDisc { .. } => None,
839    }
840}
841
842fn aggregate_function_name(name: &str) -> bool {
843    matches!(
844        name.to_ascii_lowercase().as_str(),
845        "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
846    )
847}
848
849trait SortExprExt {
850    fn expr(&self) -> &TypedExpr;
851}
852
853impl SortExprExt for crate::planner::SortExpr {
854    fn expr(&self) -> &TypedExpr {
855        &self.expr
856    }
857}
858
859#[cfg(test)]
860mod tests {
861    use super::*;
862    use crate::Span;
863    use crate::ast::expr::Literal;
864    use crate::catalog::ColumnMetadata;
865    use crate::planner::{Projection, RecursiveCteLimits, ResolvedType, SortExpr, TypedExpr};
866
867    fn references() -> Vec<TableReference> {
868        vec![TableReference::new(
869            "users",
870            TableReferenceAccess::Read,
871            crate::planner::TableReferenceSource::LogicalPlanScan,
872        )]
873    }
874
875    fn scan() -> LogicalPlan {
876        LogicalPlan::scan("users".to_string(), Projection::All(vec!["id".to_string()]))
877    }
878
879    fn column() -> TypedExpr {
880        TypedExpr::column_ref(
881            "users".to_string(),
882            "id".to_string(),
883            0,
884            ResolvedType::Integer,
885            Span::default(),
886        )
887    }
888
889    #[test]
890    fn classifies_closed_single_table_read_shape() {
891        let plan = LogicalPlan::limit(
892            LogicalPlan::sort(
893                LogicalPlan::filter(scan(), column()),
894                vec![SortExpr::asc(column())],
895            ),
896            Some(10),
897            Some(3),
898        );
899
900        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
901            panic!("single-table deterministic read must be accepted");
902        };
903        assert_eq!(descriptor.catalog_version, REMOTE_READ_CATALOG_VERSION);
904        assert_eq!(descriptor.table, "users");
905        assert_eq!(descriptor.shape, RemoteReadShape::Rows);
906        assert!(descriptor.operators.filter);
907        assert!(descriptor.operators.order_by);
908        assert!(descriptor.operators.limit);
909        assert!(descriptor.operators.offset);
910    }
911
912    #[test]
913    fn rejects_join_and_dml_before_transport() {
914        let join = LogicalPlan::join(scan(), scan(), crate::planner::JoinType::Inner, None, None);
915        assert!(matches!(
916            classify(&join, &references()),
917            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
918                if code == "join_not_supported_remote"
919        ));
920
921        let insert = LogicalPlan::insert("users".to_string(), vec!["id".to_string()], vec![]);
922        assert!(matches!(
923            classify(&insert, &references()),
924            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
925                if code == "dml_not_supported_remote"
926        ));
927    }
928
929    #[test]
930    fn rejects_ddl_and_pragma_before_a_remote_session_exists() {
931        let ddl = LogicalPlan::drop_table("users".to_string(), false);
932        assert!(matches!(
933            classify(&ddl, &[]),
934            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
935                if code == "ddl_not_supported_remote"
936        ));
937
938        let pragma = LogicalPlan::Pragma {
939            name: "cache_size".to_string(),
940            value: None,
941        };
942        assert!(matches!(
943            classify(&pragma, &[]),
944            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
945                if code == "pragma_local_only"
946        ));
947    }
948
949    #[test]
950    fn values_relations_remain_local_only() {
951        let plan = LogicalPlan::Values {
952            rows: vec![vec![TypedExpr::literal(
953                Literal::Number("1".to_string()),
954                ResolvedType::Integer,
955                Span::default(),
956            )]],
957            schema: vec![ColumnMetadata::new("column1", ResolvedType::Integer)],
958        };
959        assert!(matches!(
960            classify(&plan, &[]),
961            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
962                if code == "values_local_only"
963        ));
964    }
965
966    #[test]
967    fn recursive_cte_is_rejected_before_remote_transport() {
968        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
969        let reference = LogicalPlan::RecursiveReference {
970            name: "counter".to_string(),
971            schema: schema.clone(),
972        };
973        let recursive = LogicalPlan::RecursiveCte {
974            name: "counter".to_string(),
975            anchor: Box::new(scan()),
976            recursive_term: Box::new(reference.clone()),
977            union_all: true,
978            schema,
979            limits: RecursiveCteLimits::default(),
980        };
981
982        for plan in [&recursive, &reference] {
983            assert!(matches!(
984                classify(plan, &references()),
985                RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, reason })
986                    if code == "recursive_cte_not_supported_remote"
987                        && reason.contains("outside the v0.8 remote-read catalog")
988            ));
989        }
990    }
991
992    #[test]
993    fn stateful_and_vector_expressions_remain_local_only() {
994        let random = TypedExpr::function_call(
995            "random".to_string(),
996            vec![],
997            false,
998            false,
999            ResolvedType::Double,
1000            Span::default(),
1001        );
1002        let random_plan = LogicalPlan::filter(scan(), random);
1003        assert!(matches!(
1004            classify(&random_plan, &references()),
1005            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1006                if code == "stateful_function_local_only"
1007        ));
1008
1009        let vector_plan = LogicalPlan::filter(
1010            scan(),
1011            TypedExpr::vector_literal(vec![1.0, 2.0], 2, Span::default()),
1012        );
1013        assert!(matches!(
1014            classify(&vector_plan, &references()),
1015            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1016                if code == "vector_sql_local_only"
1017        ));
1018    }
1019
1020    #[test]
1021    fn standard_predicates_remain_local_only() {
1022        let predicate = TypedExpr::function_call(
1023            "__alopex_truth_true:0".to_string(),
1024            vec![TypedExpr::literal(
1025                Literal::Boolean(true),
1026                ResolvedType::Boolean,
1027                Span::default(),
1028            )],
1029            false,
1030            false,
1031            ResolvedType::Boolean,
1032            Span::default(),
1033        );
1034        assert!(matches!(
1035            classify(&LogicalPlan::filter(scan(), predicate), &references()),
1036            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1037                if code == "standard_predicate_local_only"
1038        ));
1039    }
1040
1041    #[test]
1042    fn try_cast_remains_local_only() {
1043        let expression = TypedExpr::try_cast(
1044            TypedExpr::literal(
1045                Literal::String("42".to_string()),
1046                ResolvedType::Text,
1047                Span::default(),
1048            ),
1049            ResolvedType::Integer,
1050            Span::default(),
1051        );
1052        assert!(matches!(
1053            classify(&LogicalPlan::filter(scan(), expression), &references()),
1054            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1055                if code == "try_cast_local_only"
1056        ));
1057    }
1058
1059    #[test]
1060    fn grouping_sets_remain_local_only() {
1061        let plan = LogicalPlan::Aggregate {
1062            input: Box::new(scan()),
1063            group_keys: vec![column()],
1064            aggregates: Vec::new(),
1065            having: None,
1066            projection: Projection::All(vec!["value".to_string()]),
1067            grouping_sets: Some(vec![0b0, 0b1]),
1068        };
1069        assert!(matches!(
1070            classify(&plan, &references()),
1071            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1072                if code == "grouping_sets_local_only"
1073        ));
1074    }
1075
1076    #[test]
1077    fn fetch_with_ties_remains_local_only() {
1078        let sorted = LogicalPlan::sort(scan(), vec![SortExpr::asc(column())]);
1079        let ties = if let LogicalPlan::Sort { order_by, .. } = &sorted {
1080            Some(order_by.clone())
1081        } else {
1082            unreachable!("sort constructed above")
1083        };
1084        let plan = LogicalPlan::Limit {
1085            input: Box::new(sorted),
1086            limit: Some(2),
1087            offset: None,
1088            ties,
1089        };
1090        assert!(matches!(
1091            classify(&plan, &references()),
1092            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1093                if code == "fetch_with_ties_local_only"
1094        ));
1095
1096        // Plain FETCH ... ONLY desugars to limit/offset and stays remote-supported.
1097        let desugared = LogicalPlan::limit(
1098            LogicalPlan::sort(scan(), vec![SortExpr::asc(column())]),
1099            Some(2),
1100            Some(1),
1101        );
1102        assert!(matches!(
1103            classify(&desugared, &references()),
1104            RemoteReadClassification::Supported(_)
1105        ));
1106
1107        assert!(
1108            coverage_entries()
1109                .iter()
1110                .any(|entry| entry.id == "pagination.fetch_with_ties"
1111                    && matches!(entry.remote_status, RemoteReadCoverageStatus::LocalOnly))
1112        );
1113    }
1114
1115    #[test]
1116    fn subqueries_are_rejected_and_descriptor_never_contains_plan() {
1117        let subquery = TypedExpr::new(
1118            TypedExprKind::ScalarSubquery(Box::new(scan())),
1119            ResolvedType::Integer,
1120            Span::default(),
1121        );
1122        assert!(matches!(
1123            classify(&LogicalPlan::filter(scan(), subquery), &references()),
1124            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
1125                if code == "subquery_not_supported_remote"
1126        ));
1127
1128        let encoded = serde_json::to_string(&classify(&scan(), &references())).unwrap();
1129        assert!(!encoded.contains("LogicalPlan"));
1130        assert!(!encoded.contains("column_index"));
1131    }
1132
1133    #[test]
1134    fn aggregate_catalog_includes_string_aggregates() {
1135        let aggregate = crate::planner::AggregateExpr {
1136            function: AggregateFunction::StringAgg {
1137                separator: Some(",".to_string()),
1138            },
1139            arg: Some(column()),
1140            distinct: true,
1141            result_type: ResolvedType::Text,
1142            filter: None,
1143            order_by: Vec::new(),
1144        };
1145        let plan = LogicalPlan::aggregate(
1146            scan(),
1147            vec![column()],
1148            vec![aggregate],
1149            Some(TypedExpr::literal(
1150                Literal::Boolean(true),
1151                ResolvedType::Boolean,
1152                Span::default(),
1153            )),
1154            Projection::All(vec![]),
1155        );
1156        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
1157            panic!("listed aggregate must be accepted");
1158        };
1159        assert_eq!(
1160            descriptor.shape,
1161            RemoteReadShape::Aggregate {
1162                aggregates: vec![RemoteAggregate::StringAgg]
1163            }
1164        );
1165        assert!(descriptor.operators.group_by);
1166        assert!(descriptor.operators.having);
1167        assert!(descriptor.operators.aggregate_distinct);
1168    }
1169}
alopex_sql/distributed_read/catalog_v0_8.rs

alopex_sql/distributed_read/
catalog_v0_8.rs
