Skip to main content

alopex_sql/distributed_read/
catalog_v0_8.rs

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