1use std::collections::BTreeSet;
9
10use serde::{Deserialize, Serialize};
11
12use crate::planner::{
13 AggregateFunction, LogicalPlan, Projection, TableReference, TableReferenceAccess, TypedExpr,
14 TypedExprKind,
15};
16
17pub const REMOTE_READ_CATALOG_VERSION: &str = "v0.8";
19
20pub 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
93pub 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
110pub enum RemoteReadClassification {
111 Supported(RemoteReadDescriptor),
113 LocalOnly(RemoteReadRejection),
116 UnsupportedRemote(RemoteReadRejection),
118}
119
120#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
135pub enum RemoteReadShape {
136 Rows,
137 Aggregate { aggregates: Vec<RemoteAggregate> },
138}
139
140#[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#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
170pub struct RemoteReadRejection {
171 pub code: String,
172 pub reason: String,
173}
174
175#[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#[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#[derive(Debug, Default, Clone, Copy)]
215pub struct RemoteReadCatalogV0_8;
216
217impl RemoteReadCatalogV0_8 {
218 pub fn classify(
220 &self,
221 plan: &LogicalPlan,
222 table_references: &[TableReference],
223 ) -> RemoteReadClassification {
224 classify(plan, table_references)
225 }
226
227 pub fn coverage_entries(&self) -> Vec<RemoteReadCoverageEntry> {
230 coverage_entries()
231 }
232}
233
234pub 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
365pub 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}