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 "gen_random_uuid",
101 "uuidv7",
102 "memory_stats",
103 "io_stats",
104 "clear_cache",
105];
106
107#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
109pub enum RemoteReadClassification {
110 Supported(RemoteReadDescriptor),
112 LocalOnly(RemoteReadRejection),
115 UnsupportedRemote(RemoteReadRejection),
117}
118
119#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
134pub enum RemoteReadShape {
135 Rows,
136 Aggregate { aggregates: Vec<RemoteAggregate> },
137}
138
139#[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#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
169pub struct RemoteReadRejection {
170 pub code: String,
171 pub reason: String,
172}
173
174#[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#[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#[derive(Debug, Default, Clone, Copy)]
214pub struct RemoteReadCatalogV0_8;
215
216impl RemoteReadCatalogV0_8 {
217 pub fn classify(
219 &self,
220 plan: &LogicalPlan,
221 table_references: &[TableReference],
222 ) -> RemoteReadClassification {
223 classify(plan, table_references)
224 }
225
226 pub fn coverage_entries(&self) -> Vec<RemoteReadCoverageEntry> {
229 coverage_entries()
230 }
231}
232
233pub 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
364pub 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}