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