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    "extract",
40    "date_part",
41    "date_trunc",
42    "to_char",
43    "to_timestamp",
44    "strftime",
45    "julianday",
46    "unixepoch",
47    "cbrt",
48    "cot",
49    "log2",
50    "acosh",
51    "asinh",
52    "atanh",
53    "cosh",
54    "sinh",
55    "tanh",
56    "isnan",
57    "sin",
58    "cos",
59    "tan",
60    "asin",
61    "acos",
62    "atan",
63    "atan2",
64    "degrees",
65    "radians",
66    "pi",
67    "sha256",
68    "md5",
69    "simhash",
70    "hamming_distance",
71    "hex",
72    "unhex",
73    "encode",
74    "decode",
75    "length",
76    "char_length",
77    "octet_length",
78    "ascii",
79    "chr",
80    "bit_length",
81    "starts_with",
82    "ends_with",
83    "translate",
84    "levenshtein",
85    "upper",
86    "lower",
87    "initcap",
88    "substr",
89    "left",
90    "right",
91    "trim",
92    "ltrim",
93    "rtrim",
94    "replace",
95    "instr",
96    "strpos",
97    "concat",
98    "concat_ws",
99    "repeat",
100    "reverse",
101    "lpad",
102    "rpad",
103    "split_part",
104    "regexp_replace",
105    "regexp_match",
106    "regexp_matches",
107    "regexp_like",
108    "coalesce",
109    "nullif",
110    "ifnull",
111    "iif",
112    "greatest",
113    "least",
114    "typeof",
115    "pg_typeof",
116    "quote",
117];
118
119/// Registered scalar identities intentionally excluded from remote execution.
120pub const REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS: &[&str] = &[
121    "age",
122    "array_append",
123    "array_cat",
124    "array_length",
125    "array_position",
126    "array_positions",
127    "array_prepend",
128    "array_remove",
129    "array_replace",
130    "array_slice",
131    "array_subscript",
132    "array_to_string",
133    "array_value",
134    "current_date",
135    "current_time",
136    "date",
137    "date_add",
138    "date_sub",
139    "datetime",
140    "json",
141    "json_array",
142    "json_array_length",
143    "json_extract",
144    "json_insert",
145    "json_object",
146    "json_remove",
147    "json_replace",
148    "json_set",
149    "json_type",
150    "json_valid",
151    "jsonb_build_array",
152    "jsonb_build_object",
153    "jsonb_extract",
154    "jsonb_extract_path",
155    "jsonb_extract_path_text",
156    "jsonb_extract_text",
157    "jsonb_insert",
158    "jsonb_set",
159    "make_date",
160    "make_interval",
161    "make_time",
162    "make_timestamp",
163    "map",
164    "list_value",
165    "plainto_tsquery",
166    "string_to_array",
167    "struct_pack",
168    "time",
169    "to_date",
170    "to_tsquery",
171    "to_tsvector",
172    "ts_headline",
173    "ts_rank",
174    "vector_similarity",
175    "vector_distance",
176    "vector_dims",
177    "vector_norm",
178    "random",
179    "now",
180    "current_timestamp",
181    "gen_random_uuid",
182    "uuidv7",
183    "memory_stats",
184    "io_stats",
185    "clear_cache",
186    "websearch_to_tsquery",
187];
188
189/// A complete pre-routing classification for a planned SQL statement.
190#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
191pub enum RemoteReadClassification {
192    /// The plan belongs to the explicitly supported v0.8 catalog.
193    Supported(RemoteReadDescriptor),
194    /// The statement remains valid for the legacy local executor, but is not
195    /// eligible for remote/multi-range execution.
196    LocalOnly(RemoteReadRejection),
197    /// A cluster read request must fail before opening a transport session.
198    UnsupportedRemote(RemoteReadRejection),
199}
200
201/// Bounded descriptor metadata derived from an accepted logical plan.
202///
203/// This is deliberately not executable SQL and carries no private planner
204/// tree.  The later transport task expands it into an expression codec with a
205/// separate compatibility test.
206#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207pub struct RemoteReadDescriptor {
208    pub catalog_version: String,
209    pub table: String,
210    pub shape: RemoteReadShape,
211    pub operators: RemoteReadOperators,
212}
213
214/// The closed high-level result shape accepted by the catalog.
215#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
216pub enum RemoteReadShape {
217    Rows,
218    Aggregate { aggregates: Vec<RemoteAggregate> },
219}
220
221/// Aggregate identities available to the v0.8 catalog.
222#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
223#[serde(rename_all = "snake_case")]
224pub enum RemoteAggregate {
225    Count,
226    Sum,
227    Total,
228    Avg,
229    Min,
230    Max,
231    GroupConcat,
232    StringAgg,
233}
234
235/// Closed modifiers which a later normalized descriptor must preserve.
236#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
237pub struct RemoteReadOperators {
238    pub filter: bool,
239    pub projection: bool,
240    pub order_by: bool,
241    pub limit: bool,
242    pub offset: bool,
243    pub group_by: bool,
244    pub having: bool,
245    pub deterministic_scalar: bool,
246    pub aggregate_distinct: bool,
247}
248
249/// Stable explanation for a non-supported remote classification.
250#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
251pub struct RemoteReadRejection {
252    pub code: String,
253    pub reason: String,
254}
255
256/// Public support status emitted by the coverage matrix.
257#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
258#[serde(rename_all = "snake_case")]
259pub enum RemoteReadCoverageStatus {
260    RemoteSupported,
261    LocalOnly,
262    PreExecutionRejection,
263}
264
265/// One stable public category row in the v0.8 coverage matrix.
266#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
267pub struct RemoteReadCoverageEntry {
268    pub id: &'static str,
269    pub public_surface: &'static str,
270    #[serde(skip_serializing)]
271    pub identities: &'static [&'static str],
272    pub remote_status: RemoteReadCoverageStatus,
273    pub prerequisite: &'static str,
274    pub normal_outcome: &'static str,
275    pub failure_outcome: &'static str,
276}
277
278impl RemoteReadRejection {
279    fn local_only(code: &str, reason: &str) -> RemoteReadClassification {
280        RemoteReadClassification::LocalOnly(Self {
281            code: code.to_string(),
282            reason: reason.to_string(),
283        })
284    }
285
286    fn unsupported(code: &str, reason: &str) -> RemoteReadClassification {
287        RemoteReadClassification::UnsupportedRemote(Self {
288            code: code.to_string(),
289            reason: reason.to_string(),
290        })
291    }
292}
293
294/// Closed v0.8 remote-read catalog.
295#[derive(Debug, Default, Clone, Copy)]
296pub struct RemoteReadCatalogV0_8;
297
298impl RemoteReadCatalogV0_8 {
299    /// Classifies a fully planned statement before routing or transport.
300    pub fn classify(
301        &self,
302        plan: &LogicalPlan,
303        table_references: &[TableReference],
304    ) -> RemoteReadClassification {
305        classify(plan, table_references)
306    }
307
308    /// Returns every public SQL category from the same closed catalog used by
309    /// the classifier.  No local feature is inferred as remotely supported.
310    pub fn coverage_entries(&self) -> Vec<RemoteReadCoverageEntry> {
311        coverage_entries()
312    }
313}
314
315/// Returns every public SQL category from the v0.8 closed catalog.
316pub fn coverage_entries() -> Vec<RemoteReadCoverageEntry> {
317    use RemoteReadCoverageStatus::{LocalOnly, PreExecutionRejection, RemoteSupported};
318
319    vec![
320        RemoteReadCoverageEntry {
321            id: "select.one_table.read_only",
322            public_surface: "one logical table SELECT with projection, WHERE, ORDER BY, LIMIT, OFFSET",
323            identities: &[
324                "select",
325                "projection",
326                "where",
327                "order_by",
328                "limit",
329                "offset",
330            ],
331            remote_status: RemoteSupported,
332            prerequisite: "closed catalog, fenced retained read point, authorized range targets",
333            normal_outcome: "prepared globally equivalent result",
334            failure_outcome: "classified pre-execution or routed-read failure; no local fallback",
335        },
336        RemoteReadCoverageEntry {
337            id: "select.aggregate.basic",
338            public_surface: "one-table COUNT, SUM, TOTAL, AVG, MIN, MAX with GROUP BY/HAVING/DISTINCT",
339            identities: &[
340                "count", "sum", "total", "avg", "min", "max", "group_by", "having", "distinct",
341            ],
342            remote_status: RemoteSupported,
343            prerequisite: "closed aggregate descriptor and global finalization budget",
344            normal_outcome: "prepared globally equivalent aggregate result",
345            failure_outcome: "classified pre-execution or global preparation failure",
346        },
347        RemoteReadCoverageEntry {
348            id: "select.aggregate.ordered_string",
349            public_surface: "one-table GROUP_CONCAT and STRING_AGG with global ordered replay",
350            identities: &["group_concat", "string_agg"],
351            remote_status: RemoteSupported,
352            prerequisite: "closed aggregate descriptor and ordered raw-value finalization budget",
353            normal_outcome: "prepared globally ordered aggregate result",
354            failure_outcome: "classified pre-execution or global preparation failure",
355        },
356        RemoteReadCoverageEntry {
357            id: "scalar.deterministic",
358            public_surface: "explicit deterministic scalar function list in RemoteReadCatalogV0_8",
359            identities: REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS,
360            remote_status: RemoteSupported,
361            prerequisite: "one-table SELECT and each function identity listed by the v0.8 catalog",
362            normal_outcome: "evaluated as part of a prepared remotely supported read",
363            failure_outcome: "unlisted function is rejected before transport",
364        },
365        RemoteReadCoverageEntry {
366            id: "scalar.local_only",
367            public_surface: "JSON, nested, full-text, temporal, vector, random/UUID, statistics, and cache-control scalar functions",
368            identities: REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS,
369            remote_status: LocalOnly,
370            prerequisite: "local execution profile",
371            normal_outcome: "v0.7.4 local SQL behavior",
372            failure_outcome: "remote profile receives an explicit local-only classification",
373        },
374        RemoteReadCoverageEntry {
375            id: "statement.ddl",
376            public_surface: "CREATE/DROP TABLE and CREATE/DROP INDEX",
377            identities: &["create_table", "drop_table", "create_index", "drop_index"],
378            remote_status: PreExecutionRejection,
379            prerequisite: "local schema-management workflow",
380            normal_outcome: "v0.7.4 local SQL behavior",
381            failure_outcome: "ddl_not_supported_remote before transport",
382        },
383        RemoteReadCoverageEntry {
384            id: "statement.dml",
385            public_surface: "INSERT, UPDATE, DELETE",
386            identities: &["insert", "update", "delete"],
387            remote_status: PreExecutionRejection,
388            prerequisite: "local transaction workflow",
389            normal_outcome: "v0.7.4 local SQL behavior",
390            failure_outcome: "dml_not_supported_remote before transport",
391        },
392        RemoteReadCoverageEntry {
393            id: "statement.pragma",
394            public_surface: "PRAGMA",
395            identities: &["pragma"],
396            remote_status: LocalOnly,
397            prerequisite: "local execution profile",
398            normal_outcome: "v0.7.4 local SQL behavior",
399            failure_outcome: "pragma_local_only before transport",
400        },
401        RemoteReadCoverageEntry {
402            id: "relation.join",
403            public_surface: "JOIN",
404            identities: &[
405                "inner_join",
406                "left_join",
407                "right_join",
408                "full_join",
409                "cross_join",
410            ],
411            remote_status: PreExecutionRejection,
412            prerequisite: "local execution profile",
413            normal_outcome: "v0.7.4 local SQL behavior",
414            failure_outcome: "join_not_supported_remote before transport",
415        },
416        RemoteReadCoverageEntry {
417            id: "relation.subquery",
418            public_surface: "scalar, IN, EXISTS, and quantified subqueries",
419            identities: &["scalar_subquery", "in_subquery", "exists", "quantified"],
420            remote_status: PreExecutionRejection,
421            prerequisite: "local execution profile",
422            normal_outcome: "v0.7.4 local SQL behavior",
423            failure_outcome: "subquery_not_supported_remote before transport",
424        },
425        RemoteReadCoverageEntry {
426            id: "relation.compound_window",
427            public_surface: "compound and window query forms",
428            identities: &["compound_query", "window_expression"],
429            remote_status: PreExecutionRejection,
430            prerequisite: "a future remote catalog version",
431            normal_outcome: "not a v0.8 remote-read form",
432            failure_outcome: "function_not_in_remote_catalog before transport",
433        },
434        RemoteReadCoverageEntry {
435            id: "scalar.standard_predicate",
436            public_surface: "truth, distinctness, and row-value predicates",
437            identities: &["truth_predicate", "is_distinct_from", "row_comparison"],
438            remote_status: LocalOnly,
439            prerequisite: "local execution profile",
440            normal_outcome: "three-valued evaluation by the local executor",
441            failure_outcome: "standard_predicate_local_only before transport",
442        },
443        RemoteReadCoverageEntry {
444            id: "scalar.try_cast",
445            public_surface: "TRY_CAST safe conversion",
446            identities: &["try_cast"],
447            remote_status: LocalOnly,
448            prerequisite: "local execution profile",
449            normal_outcome: "NULL-on-conversion-failure evaluation by the local executor",
450            failure_outcome: "try_cast_local_only before transport",
451        },
452        RemoteReadCoverageEntry {
453            id: "pagination.fetch_with_ties",
454            public_surface: "FETCH ... WITH TIES peer-preserving row limits",
455            identities: &["fetch_with_ties"],
456            remote_status: LocalOnly,
457            prerequisite: "local execution profile",
458            normal_outcome: "peer-preserving limit evaluation by the local executor",
459            failure_outcome: "fetch_with_ties_local_only before transport",
460        },
461        RemoteReadCoverageEntry {
462            id: "relation.distinct_on",
463            public_surface: "SELECT DISTINCT ON deterministic first-row deduplication",
464            identities: &["distinct_on"],
465            remote_status: LocalOnly,
466            prerequisite: "local execution profile",
467            normal_outcome: "deterministic per-key first-row evaluation by the local executor",
468            failure_outcome: "distinct_on_local_only before transport",
469        },
470        RemoteReadCoverageEntry {
471            id: "scalar.aggregate_filter",
472            public_surface: "aggregate FILTER (WHERE ...) per-aggregate row filtering",
473            identities: &["aggregate_filter"],
474            remote_status: LocalOnly,
475            prerequisite: "local execution profile",
476            normal_outcome: "per-aggregate predicate filtering by the local executor",
477            failure_outcome: "aggregate_filter_local_only before transport",
478        },
479        RemoteReadCoverageEntry {
480            id: "scalar.ordered_aggregate",
481            public_surface: "aggregate-local ORDER BY and WITHIN GROUP ordered-set aggregates",
482            identities: &["aggregate_order_by", "within_group", "percentile_disc"],
483            remote_status: LocalOnly,
484            prerequisite: "local execution profile",
485            normal_outcome: "ordered aggregate evaluation by the local executor",
486            failure_outcome: "ordered_aggregate_local_only before transport",
487        },
488        RemoteReadCoverageEntry {
489            id: "scalar.nested_aggregate",
490            public_surface: "ARRAY_AGG and JSON/JSONB collection aggregates",
491            identities: &[
492                "array_agg",
493                "json_group_array",
494                "json_group_object",
495                "jsonb_agg",
496                "jsonb_object_agg",
497            ],
498            remote_status: LocalOnly,
499            prerequisite: "local execution profile",
500            normal_outcome: "nested collection aggregation by the local executor",
501            failure_outcome: "nested_aggregate_local_only before transport",
502        },
503        RemoteReadCoverageEntry {
504            id: "aggregate.grouping_sets",
505            public_surface: "GROUPING SETS / ROLLUP / CUBE multi-set aggregation",
506            identities: &["rollup", "cube", "grouping_sets", "grouping", "grouping_id"],
507            remote_status: LocalOnly,
508            prerequisite: "local execution profile",
509            normal_outcome: "single-pass multi-set aggregation by the local executor",
510            failure_outcome: "grouping_sets_local_only before transport",
511        },
512        RemoteReadCoverageEntry {
513            id: "relation.lateral_join",
514            public_surface: "LATERAL joins over a correlated relation",
515            identities: &["lateral", "cross_join_lateral", "left_join_lateral"],
516            remote_status: PreExecutionRejection,
517            prerequisite: "local execution profile",
518            normal_outcome: "per-left-row correlated evaluation by the local executor",
519            failure_outcome: "lateral_join_not_supported_remote before transport",
520        },
521        RemoteReadCoverageEntry {
522            id: "relation.table_function",
523            public_surface: "FROM-clause table functions",
524            identities: &["unnest", "generate_series", "fts_search"],
525            remote_status: LocalOnly,
526            prerequisite: "local execution profile",
527            normal_outcome: "row generation by the local executor",
528            failure_outcome: "table_function_not_supported_remote before transport",
529        },
530        RemoteReadCoverageEntry {
531            id: "relation.recursive_cte",
532            public_surface: "recursive common table expressions",
533            identities: &["with_recursive"],
534            remote_status: PreExecutionRejection,
535            prerequisite: "local execution profile",
536            normal_outcome: "bounded fixed-point evaluation by the local executor",
537            failure_outcome: "recursive_cte_not_supported_remote before transport",
538        },
539        RemoteReadCoverageEntry {
540            id: "transaction.multi_statement",
541            public_surface: "existing multi-statement Transaction API workflow",
542            identities: &["transaction_api"],
543            remote_status: LocalOnly,
544            prerequisite: "local transaction workflow",
545            normal_outcome: "v0.7.4 local transaction behavior",
546            failure_outcome: "remote profile receives an explicit pre-execution classification",
547        },
548    ]
549}
550
551/// Classifies a fully planned statement before routing or transport.
552pub fn classify(
553    plan: &LogicalPlan,
554    table_references: &[TableReference],
555) -> RemoteReadClassification {
556    if let Some(classification) = table_boundary(table_references) {
557        return classification;
558    }
559    if plan.contains_join() {
560        return RemoteReadRejection::unsupported(
561            "join_not_supported_remote",
562            "JOIN is outside the v0.8 remote-read catalog",
563        );
564    }
565
566    let mut analysis = Analysis::default();
567    if let Err(rejection) = validate_plan(plan, &mut analysis) {
568        return rejection;
569    }
570    let table = match single_table(table_references) {
571        Some(table) => table,
572        None => {
573            return RemoteReadRejection::unsupported(
574                "single_logical_table_required",
575                "remote reads require exactly one physical logical table",
576            );
577        }
578    };
579    if analysis.scan_count != 1 {
580        return RemoteReadRejection::unsupported(
581            "single_logical_table_required",
582            "remote reads require exactly one table scan",
583        );
584    }
585    if analysis
586        .scan_tables
587        .first()
588        .is_none_or(|scan_table| scan_table != &table)
589    {
590        return RemoteReadRejection::unsupported(
591            "table_reference_mismatch",
592            "planned scan table does not match the routing table reference",
593        );
594    }
595
596    RemoteReadClassification::Supported(RemoteReadDescriptor {
597        catalog_version: REMOTE_READ_CATALOG_VERSION.to_string(),
598        table,
599        shape: if analysis.aggregates.is_empty() {
600            RemoteReadShape::Rows
601        } else {
602            RemoteReadShape::Aggregate {
603                aggregates: analysis.aggregates,
604            }
605        },
606        operators: analysis.operators,
607    })
608}
609
610fn table_boundary(table_references: &[TableReference]) -> Option<RemoteReadClassification> {
611    if table_references
612        .iter()
613        .any(|reference| reference.access != TableReferenceAccess::Read)
614    {
615        return Some(RemoteReadRejection::unsupported(
616            "read_only_select_required",
617            "remote reads require a read-only SELECT",
618        ));
619    }
620    let tables: BTreeSet<_> = table_references
621        .iter()
622        .map(|reference| reference.table_name.as_str())
623        .collect();
624    if tables.len() > 1 {
625        return Some(RemoteReadRejection::unsupported(
626            "single_logical_table_required",
627            "remote reads cannot span multiple logical tables",
628        ));
629    }
630    None
631}
632
633fn single_table(table_references: &[TableReference]) -> Option<String> {
634    table_references
635        .first()
636        .map(|reference| reference.table_name.clone())
637}
638
639#[derive(Debug, Default)]
640struct Analysis {
641    scan_count: usize,
642    scan_tables: Vec<String>,
643    aggregates: Vec<RemoteAggregate>,
644    operators: RemoteReadOperators,
645}
646
647fn validate_plan(
648    plan: &LogicalPlan,
649    analysis: &mut Analysis,
650) -> Result<(), RemoteReadClassification> {
651    match plan {
652        LogicalPlan::Pragma { .. } => Err(RemoteReadRejection::local_only(
653            "pragma_local_only",
654            "PRAGMA remains available only to the local executor",
655        )),
656        LogicalPlan::Insert { .. }
657        | LogicalPlan::InsertSelect { .. }
658        | LogicalPlan::Update { .. }
659        | LogicalPlan::Delete { .. } => Err(RemoteReadRejection::unsupported(
660            "dml_not_supported_remote",
661            "DML is outside the read-only remote-read catalog",
662        )),
663        LogicalPlan::CreateTable { .. }
664        | LogicalPlan::DropTable { .. }
665        | LogicalPlan::CreateIndex { .. }
666        | LogicalPlan::DropIndex { .. } => Err(RemoteReadRejection::unsupported(
667            "ddl_not_supported_remote",
668            "DDL is outside the read-only remote-read catalog",
669        )),
670        LogicalPlan::Join { .. } => Err(RemoteReadRejection::unsupported(
671            "join_not_supported_remote",
672            "JOIN is outside the v0.8 remote-read catalog",
673        )),
674        LogicalPlan::LateralJoin { .. } => Err(RemoteReadRejection::unsupported(
675            "lateral_join_not_supported_remote",
676            "LATERAL joins are outside the v0.8 remote-read catalog",
677        )),
678        LogicalPlan::TableFunction { .. } => Err(RemoteReadRejection::local_only(
679            "table_function_not_supported_remote",
680            "FROM-clause table functions are evaluated by the local executor",
681        )),
682        LogicalPlan::Window { .. } => Err(RemoteReadRejection::unsupported(
683            "window_not_supported_remote",
684            "window functions are outside the v0.8 remote-read catalog",
685        )),
686        LogicalPlan::SetOperation { .. } => Err(RemoteReadRejection::unsupported(
687            "set_operation_not_supported_remote",
688            "set operations are outside the v0.8 remote-read catalog",
689        )),
690        LogicalPlan::Values { .. } => Err(RemoteReadRejection::local_only(
691            "values_local_only",
692            "VALUES relations are evaluated by the local executor",
693        )),
694        LogicalPlan::RecursiveCte { .. } | LogicalPlan::RecursiveReference { .. } => {
695            Err(RemoteReadRejection::unsupported(
696                "recursive_cte_not_supported_remote",
697                "recursive common table expressions are outside the v0.8 remote-read catalog",
698            ))
699        }
700        LogicalPlan::Scan { table, projection } => {
701            analysis.scan_count += 1;
702            analysis.scan_tables.push(table.clone());
703            validate_projection(projection, false, analysis)
704        }
705        LogicalPlan::Filter { input, predicate } => {
706            analysis.operators.filter = true;
707            validate_expr(predicate, false, analysis)?;
708            validate_plan(input, analysis)
709        }
710        LogicalPlan::Project { input, projection } => {
711            validate_plan(input, analysis)?;
712            analysis.operators.projection = true;
713            validate_projection(projection, !analysis.aggregates.is_empty(), analysis)
714        }
715        LogicalPlan::Aggregate {
716            input,
717            group_keys,
718            aggregates,
719            having,
720            projection,
721            grouping_sets,
722        } => {
723            if grouping_sets.is_some() {
724                return Err(RemoteReadRejection::local_only(
725                    "grouping_sets_local_only",
726                    "GROUPING SETS/ROLLUP/CUBE aggregation is not in the v0.8 remote-read catalog",
727                ));
728            }
729            analysis.operators.group_by = !group_keys.is_empty();
730            analysis.operators.having = having.is_some();
731            for group_key in group_keys {
732                validate_expr(group_key, false, analysis)?;
733            }
734            for aggregate in aggregates {
735                if matches!(
736                    aggregate.function,
737                    AggregateFunction::ArrayAgg
738                        | AggregateFunction::JsonGroupArray
739                        | AggregateFunction::JsonGroupObject
740                        | AggregateFunction::JsonbAgg
741                        | AggregateFunction::JsonbObjectAgg
742                ) {
743                    return Err(RemoteReadRejection::local_only(
744                        "nested_aggregate_local_only",
745                        "JSON and nested aggregates are not in the v0.8 remote-read catalog",
746                    ));
747                }
748                if aggregate.filter.is_some() {
749                    return Err(RemoteReadRejection::local_only(
750                        "aggregate_filter_local_only",
751                        "aggregate FILTER (WHERE ...) is not in the v0.8 remote-read catalog",
752                    ));
753                }
754                let Some(aggregate_name) = remote_aggregate(&aggregate.function) else {
755                    return Err(RemoteReadRejection::local_only(
756                        "ordered_aggregate_local_only",
757                        "ordered-set aggregates are not in the v0.8 remote-read catalog",
758                    ));
759                };
760                if !aggregate.order_by.is_empty() {
761                    return Err(RemoteReadRejection::local_only(
762                        "ordered_aggregate_local_only",
763                        "aggregate-local ORDER BY is not in the v0.8 remote-read catalog",
764                    ));
765                }
766                analysis.operators.aggregate_distinct |= aggregate.distinct;
767                if let Some(argument) = &aggregate.arg {
768                    validate_expr(argument, false, analysis)?;
769                }
770                analysis.aggregates.push(aggregate_name);
771            }
772            if let Some(having) = having {
773                validate_expr(having, true, analysis)?;
774            }
775            validate_projection(projection, true, analysis)?;
776            validate_plan(input, analysis)
777        }
778        LogicalPlan::Sort { input, order_by } => {
779            validate_plan(input, analysis)?;
780            analysis.operators.order_by = true;
781            for sort in order_by {
782                validate_expr(sort.expr(), !analysis.aggregates.is_empty(), analysis)?;
783            }
784            Ok(())
785        }
786        LogicalPlan::Limit {
787            input,
788            limit,
789            offset,
790            ties,
791        } => {
792            if ties.is_some() {
793                return Err(RemoteReadRejection::local_only(
794                    "fetch_with_ties_local_only",
795                    "FETCH ... WITH TIES is not in the v0.8 remote-read catalog",
796                ));
797            }
798            analysis.operators.limit |= limit.is_some();
799            analysis.operators.offset |= offset.is_some();
800            validate_plan(input, analysis)
801        }
802        LogicalPlan::DistinctOn { .. } => Err(RemoteReadRejection::local_only(
803            "distinct_on_local_only",
804            "SELECT DISTINCT ON is not in the v0.8 remote-read catalog",
805        )),
806    }
807}
808
809fn validate_projection(
810    projection: &Projection,
811    allow_aggregate: bool,
812    analysis: &mut Analysis,
813) -> Result<(), RemoteReadClassification> {
814    if let Projection::Columns(columns) = projection {
815        analysis.operators.projection = true;
816        for column in columns {
817            validate_expr(&column.expr, allow_aggregate, analysis)?;
818        }
819    }
820    Ok(())
821}
822
823fn validate_expr(
824    expr: &TypedExpr,
825    allow_aggregate: bool,
826    analysis: &mut Analysis,
827) -> Result<(), RemoteReadClassification> {
828    match &expr.kind {
829        TypedExprKind::Literal(_) | TypedExprKind::ColumnRef { .. } => Ok(()),
830        TypedExprKind::VectorLiteral(_) => Err(RemoteReadRejection::local_only(
831            "vector_sql_local_only",
832            "vector SQL is not in the v0.8 remote-read catalog",
833        )),
834        TypedExprKind::TryCast { .. } => Err(RemoteReadRejection::local_only(
835            "try_cast_local_only",
836            "TRY_CAST is not in the v0.8 remote-read catalog",
837        )),
838        TypedExprKind::BinaryOp { left, right, .. } => {
839            validate_expr(left, allow_aggregate, analysis)?;
840            validate_expr(right, allow_aggregate, analysis)
841        }
842        TypedExprKind::UnaryOp { operand, .. }
843        | TypedExprKind::Cast { expr: operand, .. }
844        | TypedExprKind::IsNull { expr: operand, .. } => {
845            validate_expr(operand, allow_aggregate, analysis)
846        }
847        TypedExprKind::Case {
848            operand,
849            branches,
850            else_expr,
851        } => {
852            if let Some(operand) = operand {
853                validate_expr(operand, allow_aggregate, analysis)?;
854            }
855            for branch in branches {
856                validate_expr(&branch.when, allow_aggregate, analysis)?;
857                validate_expr(&branch.then, allow_aggregate, analysis)?;
858            }
859            if let Some(else_expr) = else_expr {
860                validate_expr(else_expr, allow_aggregate, analysis)?;
861            }
862            Ok(())
863        }
864        TypedExprKind::Between {
865            expr, low, high, ..
866        } => {
867            validate_expr(expr, allow_aggregate, analysis)?;
868            validate_expr(low, allow_aggregate, analysis)?;
869            validate_expr(high, allow_aggregate, analysis)
870        }
871        TypedExprKind::Like {
872            expr,
873            pattern,
874            escape,
875            ..
876        } => {
877            validate_expr(expr, allow_aggregate, analysis)?;
878            validate_expr(pattern, allow_aggregate, analysis)?;
879            if let Some(escape) = escape {
880                validate_expr(escape, allow_aggregate, analysis)?;
881            }
882            Ok(())
883        }
884        TypedExprKind::InList { expr, list, .. } => {
885            validate_expr(expr, allow_aggregate, analysis)?;
886            for item in list {
887                validate_expr(item, allow_aggregate, analysis)?;
888            }
889            Ok(())
890        }
891        TypedExprKind::FunctionCall { name, args, .. } => {
892            if allow_aggregate && aggregate_function_name(name) {
893                for argument in args {
894                    validate_expr(argument, false, analysis)?;
895                }
896                return Ok(());
897            }
898            let normalized = name.to_ascii_lowercase();
899            if normalized.starts_with("__alopex_truth_") || normalized.starts_with("__alopex_row_")
900            {
901                return Err(RemoteReadRejection::local_only(
902                    "standard_predicate_local_only",
903                    "standard predicates are not in the v0.8 remote-read catalog",
904                ));
905            }
906            if REMOTE_LOCAL_ONLY_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
907                return Err(RemoteReadRejection::local_only(
908                    "stateful_function_local_only",
909                    "the requested scalar function remains local-only",
910                ));
911            }
912            if !REMOTE_DETERMINISTIC_SCALAR_FUNCTIONS.contains(&normalized.as_str()) {
913                return Err(RemoteReadRejection::unsupported(
914                    "function_not_in_remote_catalog",
915                    "function is not explicitly listed in the remote-read catalog",
916                ));
917            }
918            analysis.operators.deterministic_scalar = true;
919            for argument in args {
920                validate_expr(argument, false, analysis)?;
921            }
922            Ok(())
923        }
924        TypedExprKind::ScalarSubquery(_)
925        | TypedExprKind::InSubquery { .. }
926        | TypedExprKind::Exists { .. }
927        | TypedExprKind::Quantified { .. } => Err(RemoteReadRejection::unsupported(
928            "subquery_not_supported_remote",
929            "subqueries are outside the v0.8 remote-read catalog",
930        )),
931    }
932}
933
934/// Aggregate identities admitted to the closed v0.8 catalog. Ordered-set
935/// aggregates (PERCENTILE_DISC, issue #148) return `None` and classify as
936/// `ordered_aggregate_local_only`.
937fn remote_aggregate(function: &AggregateFunction) -> Option<RemoteAggregate> {
938    match function {
939        AggregateFunction::Count => Some(RemoteAggregate::Count),
940        AggregateFunction::Sum => Some(RemoteAggregate::Sum),
941        AggregateFunction::Total => Some(RemoteAggregate::Total),
942        AggregateFunction::Avg => Some(RemoteAggregate::Avg),
943        AggregateFunction::Min => Some(RemoteAggregate::Min),
944        AggregateFunction::Max => Some(RemoteAggregate::Max),
945        AggregateFunction::GroupConcat { .. } => Some(RemoteAggregate::GroupConcat),
946        AggregateFunction::StringAgg { .. } => Some(RemoteAggregate::StringAgg),
947        AggregateFunction::PercentileDisc { .. } => None,
948        _ => None,
949    }
950}
951
952fn aggregate_function_name(name: &str) -> bool {
953    matches!(
954        name.to_ascii_lowercase().as_str(),
955        "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
956    )
957}
958
959trait SortExprExt {
960    fn expr(&self) -> &TypedExpr;
961}
962
963impl SortExprExt for crate::planner::SortExpr {
964    fn expr(&self) -> &TypedExpr {
965        &self.expr
966    }
967}
968
969#[cfg(test)]
970mod tests {
971    use super::*;
972    use crate::Span;
973    use crate::ast::expr::Literal;
974    use crate::catalog::ColumnMetadata;
975    use crate::planner::{Projection, RecursiveCteLimits, ResolvedType, SortExpr, TypedExpr};
976
977    fn references() -> Vec<TableReference> {
978        vec![TableReference::new(
979            "users",
980            TableReferenceAccess::Read,
981            crate::planner::TableReferenceSource::LogicalPlanScan,
982        )]
983    }
984
985    fn scan() -> LogicalPlan {
986        LogicalPlan::scan("users".to_string(), Projection::All(vec!["id".to_string()]))
987    }
988
989    fn column() -> TypedExpr {
990        TypedExpr::column_ref(
991            "users".to_string(),
992            "id".to_string(),
993            0,
994            ResolvedType::Integer,
995            Span::default(),
996        )
997    }
998
999    #[test]
1000    fn classifies_closed_single_table_read_shape() {
1001        let plan = LogicalPlan::limit(
1002            LogicalPlan::sort(
1003                LogicalPlan::filter(scan(), column()),
1004                vec![SortExpr::asc(column())],
1005            ),
1006            Some(10),
1007            Some(3),
1008        );
1009
1010        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
1011            panic!("single-table deterministic read must be accepted");
1012        };
1013        assert_eq!(descriptor.catalog_version, REMOTE_READ_CATALOG_VERSION);
1014        assert_eq!(descriptor.table, "users");
1015        assert_eq!(descriptor.shape, RemoteReadShape::Rows);
1016        assert!(descriptor.operators.filter);
1017        assert!(descriptor.operators.order_by);
1018        assert!(descriptor.operators.limit);
1019        assert!(descriptor.operators.offset);
1020    }
1021
1022    #[test]
1023    fn rejects_join_and_dml_before_transport() {
1024        let join = LogicalPlan::join(scan(), scan(), crate::planner::JoinType::Inner, None, None);
1025        assert!(matches!(
1026            classify(&join, &references()),
1027            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
1028                if code == "join_not_supported_remote"
1029        ));
1030
1031        let insert = LogicalPlan::insert("users".to_string(), vec!["id".to_string()], vec![]);
1032        assert!(matches!(
1033            classify(&insert, &references()),
1034            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
1035                if code == "dml_not_supported_remote"
1036        ));
1037    }
1038
1039    #[test]
1040    fn rejects_ddl_and_pragma_before_a_remote_session_exists() {
1041        let ddl = LogicalPlan::drop_table("users".to_string(), false);
1042        assert!(matches!(
1043            classify(&ddl, &[]),
1044            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
1045                if code == "ddl_not_supported_remote"
1046        ));
1047
1048        let pragma = LogicalPlan::Pragma {
1049            name: "cache_size".to_string(),
1050            value: None,
1051        };
1052        assert!(matches!(
1053            classify(&pragma, &[]),
1054            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1055                if code == "pragma_local_only"
1056        ));
1057    }
1058
1059    #[test]
1060    fn values_relations_remain_local_only() {
1061        let plan = LogicalPlan::Values {
1062            rows: vec![vec![TypedExpr::literal(
1063                Literal::Number("1".to_string()),
1064                ResolvedType::Integer,
1065                Span::default(),
1066            )]],
1067            schema: vec![ColumnMetadata::new("column1", ResolvedType::Integer)],
1068        };
1069        assert!(matches!(
1070            classify(&plan, &[]),
1071            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1072                if code == "values_local_only"
1073        ));
1074    }
1075
1076    #[test]
1077    fn recursive_cte_is_rejected_before_remote_transport() {
1078        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
1079        let reference = LogicalPlan::RecursiveReference {
1080            name: "counter".to_string(),
1081            schema: schema.clone(),
1082        };
1083        let recursive = LogicalPlan::RecursiveCte {
1084            name: "counter".to_string(),
1085            anchor: Box::new(scan()),
1086            recursive_term: Box::new(reference.clone()),
1087            union_all: true,
1088            schema,
1089            limits: RecursiveCteLimits::default(),
1090        };
1091
1092        for plan in [&recursive, &reference] {
1093            assert!(matches!(
1094                classify(plan, &references()),
1095                RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, reason })
1096                    if code == "recursive_cte_not_supported_remote"
1097                        && reason.contains("outside the v0.8 remote-read catalog")
1098            ));
1099        }
1100    }
1101
1102    #[test]
1103    fn explicitly_excluded_scalar_and_vector_expressions_remain_local_only() {
1104        for name in ["random", "json_valid", "date_add", "array_length"] {
1105            let expression = TypedExpr::function_call(
1106                name.to_string(),
1107                vec![],
1108                false,
1109                false,
1110                ResolvedType::Null,
1111                Span::default(),
1112            );
1113            assert!(matches!(
1114                classify(&LogicalPlan::filter(scan(), expression), &references()),
1115                RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1116                    if code == "stateful_function_local_only"
1117            ));
1118        }
1119
1120        let vector_plan = LogicalPlan::filter(
1121            scan(),
1122            TypedExpr::vector_literal(vec![1.0, 2.0], 2, Span::default()),
1123        );
1124        assert!(matches!(
1125            classify(&vector_plan, &references()),
1126            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1127                if code == "vector_sql_local_only"
1128        ));
1129    }
1130
1131    #[test]
1132    fn standard_predicates_remain_local_only() {
1133        let predicate = TypedExpr::function_call(
1134            "__alopex_truth_true:0".to_string(),
1135            vec![TypedExpr::literal(
1136                Literal::Boolean(true),
1137                ResolvedType::Boolean,
1138                Span::default(),
1139            )],
1140            false,
1141            false,
1142            ResolvedType::Boolean,
1143            Span::default(),
1144        );
1145        assert!(matches!(
1146            classify(&LogicalPlan::filter(scan(), predicate), &references()),
1147            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1148                if code == "standard_predicate_local_only"
1149        ));
1150    }
1151
1152    #[test]
1153    fn try_cast_remains_local_only() {
1154        let expression = TypedExpr::try_cast(
1155            TypedExpr::literal(
1156                Literal::String("42".to_string()),
1157                ResolvedType::Text,
1158                Span::default(),
1159            ),
1160            ResolvedType::Integer,
1161            Span::default(),
1162        );
1163        assert!(matches!(
1164            classify(&LogicalPlan::filter(scan(), expression), &references()),
1165            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1166                if code == "try_cast_local_only"
1167        ));
1168    }
1169
1170    #[test]
1171    fn grouping_sets_remain_local_only() {
1172        let plan = LogicalPlan::Aggregate {
1173            input: Box::new(scan()),
1174            group_keys: vec![column()],
1175            aggregates: Vec::new(),
1176            having: None,
1177            projection: Projection::All(vec!["value".to_string()]),
1178            grouping_sets: Some(vec![0b0, 0b1]),
1179        };
1180        assert!(matches!(
1181            classify(&plan, &references()),
1182            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1183                if code == "grouping_sets_local_only"
1184        ));
1185    }
1186
1187    #[test]
1188    fn fetch_with_ties_remains_local_only() {
1189        let sorted = LogicalPlan::sort(scan(), vec![SortExpr::asc(column())]);
1190        let ties = if let LogicalPlan::Sort { order_by, .. } = &sorted {
1191            Some(order_by.clone())
1192        } else {
1193            unreachable!("sort constructed above")
1194        };
1195        let plan = LogicalPlan::Limit {
1196            input: Box::new(sorted),
1197            limit: Some(2),
1198            offset: None,
1199            ties,
1200        };
1201        assert!(matches!(
1202            classify(&plan, &references()),
1203            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1204                if code == "fetch_with_ties_local_only"
1205        ));
1206
1207        // Plain FETCH ... ONLY desugars to limit/offset and stays remote-supported.
1208        let desugared = LogicalPlan::limit(
1209            LogicalPlan::sort(scan(), vec![SortExpr::asc(column())]),
1210            Some(2),
1211            Some(1),
1212        );
1213        assert!(matches!(
1214            classify(&desugared, &references()),
1215            RemoteReadClassification::Supported(_)
1216        ));
1217
1218        assert!(
1219            coverage_entries()
1220                .iter()
1221                .any(|entry| entry.id == "pagination.fetch_with_ties"
1222                    && matches!(entry.remote_status, RemoteReadCoverageStatus::LocalOnly))
1223        );
1224    }
1225
1226    #[test]
1227    fn subqueries_are_rejected_and_descriptor_never_contains_plan() {
1228        let subquery = TypedExpr::new(
1229            TypedExprKind::ScalarSubquery(Box::new(scan())),
1230            ResolvedType::Integer,
1231            Span::default(),
1232        );
1233        assert!(matches!(
1234            classify(&LogicalPlan::filter(scan(), subquery), &references()),
1235            RemoteReadClassification::UnsupportedRemote(RemoteReadRejection { code, .. })
1236                if code == "subquery_not_supported_remote"
1237        ));
1238
1239        let encoded = serde_json::to_string(&classify(&scan(), &references())).unwrap();
1240        assert!(!encoded.contains("LogicalPlan"));
1241        assert!(!encoded.contains("column_index"));
1242    }
1243
1244    #[test]
1245    fn aggregate_catalog_includes_string_aggregates() {
1246        let aggregate = crate::planner::AggregateExpr {
1247            function: AggregateFunction::StringAgg {
1248                separator: Some(",".to_string()),
1249            },
1250            arg: Some(column()),
1251            extra_args: Vec::new(),
1252            distinct: true,
1253            result_type: ResolvedType::Text,
1254            filter: None,
1255            order_by: Vec::new(),
1256        };
1257        let plan = LogicalPlan::aggregate(
1258            scan(),
1259            vec![column()],
1260            vec![aggregate],
1261            Some(TypedExpr::literal(
1262                Literal::Boolean(true),
1263                ResolvedType::Boolean,
1264                Span::default(),
1265            )),
1266            Projection::All(vec![]),
1267        );
1268        let RemoteReadClassification::Supported(descriptor) = classify(&plan, &references()) else {
1269            panic!("listed aggregate must be accepted");
1270        };
1271        assert_eq!(
1272            descriptor.shape,
1273            RemoteReadShape::Aggregate {
1274                aggregates: vec![RemoteAggregate::StringAgg]
1275            }
1276        );
1277        assert!(descriptor.operators.group_by);
1278        assert!(descriptor.operators.having);
1279        assert!(descriptor.operators.aggregate_distinct);
1280    }
1281
1282    #[test]
1283    fn nested_aggregates_remain_local_only() {
1284        let aggregate = crate::planner::AggregateExpr {
1285            function: AggregateFunction::ArrayAgg,
1286            arg: Some(column()),
1287            extra_args: Vec::new(),
1288            distinct: false,
1289            result_type: ResolvedType::Array(Box::new(ResolvedType::Integer)),
1290            filter: None,
1291            order_by: Vec::new(),
1292        };
1293        let plan = LogicalPlan::aggregate(
1294            scan(),
1295            Vec::new(),
1296            vec![aggregate],
1297            None,
1298            Projection::All(vec![]),
1299        );
1300        assert!(matches!(
1301            classify(&plan, &references()),
1302            RemoteReadClassification::LocalOnly(RemoteReadRejection { code, .. })
1303                if code == "nested_aggregate_local_only"
1304        ));
1305    }
1306}
alopex_sql/distributed_read/catalog_v0_8.rs

alopex_sql/distributed_read/
catalog_v0_8.rs
