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

alopex_sql/distributed_read/
catalog_v0_8.rs
