Skip to main content

reddb_server/runtime/
impl_graph_commands.rs

1//! Execution of GRAPH and SEARCH SQL-like commands.
2//!
3//! Maps parsed `GraphCommand` and `SearchCommand` AST nodes to the existing
4//! runtime graph analytics and search methods, returning results wrapped in
5//! `RuntimeQueryResult`.
6
7use super::*;
8use crate::storage::query::ast::GraphCommandOrderBy;
9use std::cmp::Ordering;
10
11impl RedDBRuntime {
12    /// Execute a GRAPH analytics command.
13    pub fn execute_graph_command(
14        &self,
15        raw_query: &str,
16        cmd: &GraphCommand,
17    ) -> RedDBResult<RuntimeQueryResult> {
18        match cmd {
19            GraphCommand::Neighborhood {
20                source,
21                depth,
22                direction,
23                edge_labels,
24            } => {
25                let dir = parse_direction(direction)?;
26                let res = self.graph_neighborhood(
27                    source,
28                    dir,
29                    *depth as usize,
30                    edge_labels.clone(),
31                    None,
32                )?;
33                let mut result = UnifiedResult::with_columns(vec![
34                    "node_id".into(),
35                    "label".into(),
36                    "node_type".into(),
37                    "depth".into(),
38                ]);
39                for visit in &res.nodes {
40                    let mut record = UnifiedRecord::new();
41                    record.set("node_id", Value::text(visit.node.id.clone()));
42                    record.set("label", Value::text(visit.node.label.clone()));
43                    record.set("node_type", Value::text(visit.node.node_type.clone()));
44                    record.set("depth", Value::Integer(visit.depth as i64));
45                    result.push(record);
46                }
47                Ok(RuntimeQueryResult {
48                    query: raw_query.to_string(),
49                    mode: QueryMode::Sql,
50                    statement: "graph_neighborhood",
51                    engine: "runtime-graph",
52                    result,
53                    affected_rows: 0,
54                    statement_type: "select",
55                    bookmark: None,
56                    notice: None,
57                })
58            }
59            GraphCommand::ShortestPath {
60                source,
61                target,
62                algorithm,
63                direction,
64                limit,
65                order_by,
66            } => {
67                let dir = parse_direction(direction)?;
68                let alg = parse_path_algorithm(algorithm)?;
69                let res = self.graph_shortest_path(source, target, dir, alg, None, None)?;
70                let mut result = UnifiedResult::with_columns(vec![
71                    "source".into(),
72                    "target".into(),
73                    "nodes_visited".into(),
74                    "negative_cycle_detected".into(),
75                    "path_found".into(),
76                    "hop_count".into(),
77                    "total_weight".into(),
78                ]);
79                let mut record = UnifiedRecord::new();
80                record.set("source", Value::text(res.source));
81                record.set("target", Value::text(res.target));
82                record.set("nodes_visited", Value::Integer(res.nodes_visited as i64));
83                record.set(
84                    "negative_cycle_detected",
85                    match res.negative_cycle_detected {
86                        Some(value) => Value::Boolean(value),
87                        None => Value::Null,
88                    },
89                );
90                if let Some(ref path) = res.path {
91                    record.set("path_found", Value::Boolean(true));
92                    record.set("hop_count", Value::Integer(path.hop_count as i64));
93                    record.set("total_weight", Value::Float(path.total_weight));
94                } else {
95                    record.set("path_found", Value::Boolean(false));
96                    record.set("hop_count", Value::Null);
97                    record.set("total_weight", Value::Null);
98                }
99                result.push(record);
100                apply_graph_order_and_limit(
101                    &mut result,
102                    "graph_shortest_path",
103                    order_by.as_ref(),
104                    limit.map(|n| n as usize),
105                )?;
106                Ok(RuntimeQueryResult {
107                    query: raw_query.to_string(),
108                    mode: QueryMode::Sql,
109                    statement: "graph_shortest_path",
110                    engine: "runtime-graph",
111                    result,
112                    affected_rows: 0,
113                    statement_type: "select",
114                    bookmark: None,
115                    notice: None,
116                })
117            }
118            GraphCommand::Properties { source } => {
119                if let Some(node_ref) = source {
120                    // Per-node property lookup (#423). Uses the same label
121                    // resolution as NEIGHBORHOOD/TRAVERSE so '<label>' and
122                    // '<numeric id>' both work.
123                    let graph =
124                        materialize_graph_with_projection(self.inner.db.store().as_ref(), None)?;
125                    let resolved = resolve_graph_node_id(&graph, node_ref)?;
126                    let stored = graph
127                        .get_node(&resolved)
128                        .ok_or_else(|| RedDBError::NotFound(node_ref.to_string()))?;
129                    let node_type = self
130                        .inner
131                        .db
132                        .store()
133                        .query_all(|entity| {
134                            entity.id.raw().to_string() == resolved
135                                && matches!(
136                                    entity.kind,
137                                    crate::storage::unified::EntityKind::GraphNode(_)
138                                )
139                        })
140                        .into_iter()
141                        .find_map(|(_, entity)| match entity.kind {
142                            crate::storage::unified::EntityKind::GraphNode(node) => {
143                                Some(node.node_type)
144                            }
145                            _ => None,
146                        })
147                        .unwrap_or_else(|| stored.node_type.clone());
148                    let all_props =
149                        materialize_graph_node_properties(self.inner.db.store().as_ref())?;
150                    let props = all_props.get(&resolved).cloned().unwrap_or_default();
151
152                    // Fixed columns first, then property keys in sorted order so
153                    // the schema is stable across snapshots / wire renders.
154                    let mut prop_keys: Vec<&String> = props.keys().collect();
155                    prop_keys.sort();
156                    let mut columns: Vec<String> = Vec::with_capacity(3 + prop_keys.len());
157                    columns.push("node_id".into());
158                    columns.push("label".into());
159                    columns.push("node_type".into());
160                    for k in &prop_keys {
161                        columns.push((*k).clone());
162                    }
163                    let mut result = UnifiedResult::with_columns(columns);
164                    let mut record = UnifiedRecord::new();
165                    record.set("node_id", Value::text(stored.id.clone()));
166                    record.set("label", Value::text(stored.label.clone()));
167                    record.set("node_type", Value::text(node_type));
168                    for k in &prop_keys {
169                        if let Some(v) = props.get(*k) {
170                            record.set(k.as_str(), v.clone());
171                        }
172                    }
173                    result.push(record);
174                    return Ok(RuntimeQueryResult {
175                        query: raw_query.to_string(),
176                        mode: QueryMode::Sql,
177                        statement: "graph_properties",
178                        engine: "runtime-graph",
179                        result,
180                        affected_rows: 0,
181                        statement_type: "select",
182                        bookmark: None,
183                        notice: None,
184                    });
185                }
186                let res = self.graph_properties(None)?;
187                let mut result = UnifiedResult::with_columns(vec![
188                    "node_count".into(),
189                    "edge_count".into(),
190                    "is_connected".into(),
191                    "is_complete".into(),
192                    "is_cyclic".into(),
193                    "density".into(),
194                ]);
195                let mut record = UnifiedRecord::new();
196                record.set("node_count", Value::Integer(res.node_count as i64));
197                record.set("edge_count", Value::Integer(res.edge_count as i64));
198                record.set("is_connected", Value::Boolean(res.is_connected));
199                record.set("is_complete", Value::Boolean(res.is_complete));
200                record.set("is_cyclic", Value::Boolean(res.is_cyclic));
201                record.set("density", Value::Float(res.density));
202                result.push(record);
203                Ok(RuntimeQueryResult {
204                    query: raw_query.to_string(),
205                    mode: QueryMode::Sql,
206                    statement: "graph_properties",
207                    engine: "runtime-graph",
208                    result,
209                    affected_rows: 0,
210                    statement_type: "select",
211                    bookmark: None,
212                    notice: None,
213                })
214            }
215            GraphCommand::Traverse {
216                source,
217                strategy,
218                depth,
219                direction,
220                edge_labels,
221            } => {
222                let dir = parse_direction(direction)?;
223                let strat = parse_traversal_strategy(strategy)?;
224                let res = self.graph_traverse(
225                    source,
226                    dir,
227                    *depth as usize,
228                    strat,
229                    edge_labels.clone(),
230                    None,
231                )?;
232                let mut result = UnifiedResult::with_columns(vec![
233                    "node_id".into(),
234                    "label".into(),
235                    "node_type".into(),
236                    "depth".into(),
237                ]);
238                for visit in &res.visits {
239                    let mut record = UnifiedRecord::new();
240                    record.set("node_id", Value::text(visit.node.id.clone()));
241                    record.set("label", Value::text(visit.node.label.clone()));
242                    record.set("node_type", Value::text(visit.node.node_type.clone()));
243                    record.set("depth", Value::Integer(visit.depth as i64));
244                    result.push(record);
245                }
246                Ok(RuntimeQueryResult {
247                    query: raw_query.to_string(),
248                    mode: QueryMode::Sql,
249                    statement: "graph_traverse",
250                    engine: "runtime-graph",
251                    result,
252                    affected_rows: 0,
253                    statement_type: "select",
254                    bookmark: None,
255                    notice: None,
256                })
257            }
258            GraphCommand::Centrality {
259                algorithm,
260                limit,
261                order_by,
262            } => {
263                let alg = parse_centrality_algorithm(algorithm)?;
264                // `limit = None` keeps historical implicit top-100 cap.
265                // `Some(0)` returns zero rows (standard SQL LIMIT 0 semantics).
266                let limit_usize = limit.map(|n| n as usize);
267                let order_needs_full_set = order_by
268                    .as_ref()
269                    .map(|order| order.ascending)
270                    .unwrap_or(false);
271                let top_k = if order_needs_full_set {
272                    usize::MAX
273                } else {
274                    limit_usize.unwrap_or(100).max(1)
275                };
276                let res = self.graph_centrality(alg, top_k, false, None, None, None, None)?;
277                let mut result = UnifiedResult::with_columns(vec![
278                    "node_id".into(),
279                    "label".into(),
280                    "score".into(),
281                ]);
282                for score in &res.scores {
283                    let mut record = UnifiedRecord::new();
284                    record.set("node_id", Value::text(score.node.id.clone()));
285                    record.set("label", Value::text(score.node.label.clone()));
286                    record.set("score", Value::Float(score.score));
287                    result.push(record);
288                }
289                for ds in &res.degree_scores {
290                    let mut record = UnifiedRecord::new();
291                    record.set("node_id", Value::text(ds.node.id.clone()));
292                    record.set("label", Value::text(ds.node.label.clone()));
293                    record.set("score", Value::Float(ds.total_degree as f64));
294                    result.push(record);
295                }
296                apply_graph_order_and_limit(
297                    &mut result,
298                    "graph_centrality",
299                    order_by.as_ref(),
300                    Some(limit_usize.unwrap_or(100)),
301                )?;
302                Ok(RuntimeQueryResult {
303                    query: raw_query.to_string(),
304                    mode: QueryMode::Sql,
305                    statement: "graph_centrality",
306                    engine: "runtime-graph",
307                    result,
308                    affected_rows: 0,
309                    statement_type: "select",
310                    bookmark: None,
311                    notice: None,
312                })
313            }
314            GraphCommand::Community {
315                algorithm,
316                max_iterations,
317                limit,
318                order_by,
319                return_assignments,
320            } => {
321                let alg = parse_community_algorithm(algorithm)?;
322                let res =
323                    self.graph_communities(alg, 1, Some(*max_iterations as usize), None, None)?;
324                let result = if *return_assignments {
325                    // Per-node node→community map (#660). Communities arrive
326                    // sorted (size desc, id asc); within each, sort nodes for a
327                    // deterministic row order. ORDER BY metrics target the
328                    // aggregate shape, so they don't apply here — only LIMIT
329                    // (row cap) is honoured.
330                    let mut result =
331                        UnifiedResult::with_columns(vec!["node_id".into(), "community_id".into()]);
332                    let row_cap = limit.map(|n| n as usize);
333                    'outer: for community in &res.communities {
334                        let mut nodes = community.nodes.clone();
335                        nodes.sort();
336                        for node_id in nodes {
337                            if row_cap.is_some_and(|cap| result.records.len() >= cap) {
338                                break 'outer;
339                            }
340                            let mut record = UnifiedRecord::new();
341                            record.set("node_id", Value::text(node_id));
342                            record.set("community_id", Value::text(community.id.clone()));
343                            result.push(record);
344                        }
345                    }
346                    result
347                } else {
348                    let mut result =
349                        UnifiedResult::with_columns(vec!["community_id".into(), "size".into()]);
350                    for community in &res.communities {
351                        let mut record = UnifiedRecord::new();
352                        record.set("community_id", Value::text(community.id.clone()));
353                        record.set("size", Value::Integer(community.size as i64));
354                        result.push(record);
355                    }
356                    apply_graph_order_and_limit(
357                        &mut result,
358                        "graph_community",
359                        order_by.as_ref(),
360                        limit.map(|n| n as usize),
361                    )?;
362                    result
363                };
364                Ok(RuntimeQueryResult {
365                    query: raw_query.to_string(),
366                    mode: QueryMode::Sql,
367                    statement: "graph_community",
368                    engine: "runtime-graph",
369                    result,
370                    affected_rows: 0,
371                    statement_type: "select",
372                    bookmark: None,
373                    notice: None,
374                })
375            }
376            GraphCommand::Components {
377                mode,
378                limit,
379                order_by,
380            } => {
381                let m = parse_components_mode(mode)?;
382                let res = self.graph_components(m, 1, None)?;
383                let mut result =
384                    UnifiedResult::with_columns(vec!["component_id".into(), "size".into()]);
385                for component in &res.components {
386                    let mut record = UnifiedRecord::new();
387                    record.set("component_id", Value::text(component.id.clone()));
388                    record.set("size", Value::Integer(component.size as i64));
389                    result.push(record);
390                }
391                apply_graph_order_and_limit(
392                    &mut result,
393                    "graph_components",
394                    order_by.as_ref(),
395                    limit.map(|n| n as usize),
396                )?;
397                Ok(RuntimeQueryResult {
398                    query: raw_query.to_string(),
399                    mode: QueryMode::Sql,
400                    statement: "graph_components",
401                    engine: "runtime-graph",
402                    result,
403                    affected_rows: 0,
404                    statement_type: "select",
405                    bookmark: None,
406                    notice: None,
407                })
408            }
409            GraphCommand::Cycles { max_length } => {
410                let res = self.graph_cycles(*max_length as usize, 100, None)?;
411                let mut result =
412                    UnifiedResult::with_columns(vec!["cycle_index".into(), "length".into()]);
413                for (i, cycle) in res.cycles.iter().enumerate() {
414                    let mut record = UnifiedRecord::new();
415                    record.set("cycle_index", Value::Integer(i as i64));
416                    record.set("length", Value::Integer(cycle.nodes.len() as i64));
417                    result.push(record);
418                }
419                Ok(RuntimeQueryResult {
420                    query: raw_query.to_string(),
421                    mode: QueryMode::Sql,
422                    statement: "graph_cycles",
423                    engine: "runtime-graph",
424                    result,
425                    affected_rows: 0,
426                    statement_type: "select",
427                    bookmark: None,
428                    notice: None,
429                })
430            }
431            GraphCommand::Clustering => {
432                let res = self.graph_clustering(100, true, None)?;
433                let mut result = UnifiedResult::with_columns(vec![
434                    "node_id".into(),
435                    "label".into(),
436                    "score".into(),
437                ]);
438                // First row: global coefficient
439                let mut global_record = UnifiedRecord::new();
440                global_record.set("node_id", Value::text("__global__"));
441                global_record.set("label", Value::text("global_clustering"));
442                global_record.set("score", Value::Float(res.global));
443                result.push(global_record);
444                for score in &res.local {
445                    let mut record = UnifiedRecord::new();
446                    record.set("node_id", Value::text(score.node.id.clone()));
447                    record.set("label", Value::text(score.node.label.clone()));
448                    record.set("score", Value::Float(score.score));
449                    result.push(record);
450                }
451                Ok(RuntimeQueryResult {
452                    query: raw_query.to_string(),
453                    mode: QueryMode::Sql,
454                    statement: "graph_clustering",
455                    engine: "runtime-graph",
456                    result,
457                    affected_rows: 0,
458                    statement_type: "select",
459                    bookmark: None,
460                    notice: None,
461                })
462            }
463            GraphCommand::TopologicalSort => {
464                let res = self.graph_topological_sort(None)?;
465                let mut result = UnifiedResult::with_columns(vec![
466                    "order".into(),
467                    "node_id".into(),
468                    "label".into(),
469                ]);
470                for (i, node) in res.ordered_nodes.iter().enumerate() {
471                    let mut record = UnifiedRecord::new();
472                    record.set("order", Value::Integer(i as i64));
473                    record.set("node_id", Value::text(node.id.clone()));
474                    record.set("label", Value::text(node.label.clone()));
475                    result.push(record);
476                }
477                Ok(RuntimeQueryResult {
478                    query: raw_query.to_string(),
479                    mode: QueryMode::Sql,
480                    statement: "graph_topological_sort",
481                    engine: "runtime-graph",
482                    result,
483                    affected_rows: 0,
484                    statement_type: "select",
485                    bookmark: None,
486                    notice: None,
487                })
488            }
489        }
490    }
491
492    /// Execute a SEARCH command.
493    pub fn execute_search_command(
494        &self,
495        raw_query: &str,
496        cmd: &SearchCommand,
497    ) -> RedDBResult<RuntimeQueryResult> {
498        match cmd {
499            SearchCommand::Similar {
500                vector,
501                text,
502                provider,
503                collection,
504                limit,
505                min_score,
506                vector_param,
507                limit_param,
508                min_score_param,
509                text_param,
510            } => {
511                if vector_param.is_some() {
512                    return Err(RedDBError::Query(
513                        "SEARCH SIMILAR $N vector parameter was not bound before execution"
514                            .to_string(),
515                    ));
516                }
517                if limit_param.is_some() {
518                    return Err(RedDBError::Query(
519                        "SEARCH SIMILAR LIMIT $N parameter was not bound before execution"
520                            .to_string(),
521                    ));
522                }
523                if min_score_param.is_some() {
524                    return Err(RedDBError::Query(
525                        "SEARCH SIMILAR MIN_SCORE $N parameter was not bound before execution"
526                            .to_string(),
527                    ));
528                }
529                if text_param.is_some() {
530                    return Err(RedDBError::Query(
531                        "SEARCH SIMILAR TEXT $N parameter was not bound before execution"
532                            .to_string(),
533                    ));
534                }
535                // If text provided, generate embedding first (semantic search)
536                let search_vector = if let Some(query_text) = text {
537                    // Embeddings follow the embeddings task pointer (ADR-0068
538                    // §5); an explicit `USING` overrides it for this query.
539                    use crate::application::ports::RuntimeEntityPort;
540                    let kv_getter = |key: &str| -> RedDBResult<Option<String>> {
541                        match self.get_kv("red_config", key)? {
542                            Some((Value::Text(s), _)) => Ok(Some(s.to_string())),
543                            _ => Ok(None),
544                        }
545                    };
546                    let provider = match provider.as_deref() {
547                        Some(p) => {
548                            let provider = crate::ai::parse_provider(p)?;
549                            crate::ai::ensure_provider_supports_embeddings(&provider)?;
550                            provider
551                        }
552                        None => crate::ai::resolve_embeddings_provider(&kv_getter)?,
553                    };
554                    // S3 / #711: planner-level provider gate. Runs
555                    // before the credential resolver so the resolver
556                    // audit event is not emitted when policy denies.
557                    crate::runtime::ai::provider_gate::enforce(self, &provider)?;
558                    let api_key = crate::ai::resolve_api_key_from_runtime(&provider, None, self)?;
559                    let model = crate::ai::resolve_embeddings_model(&provider, &kv_getter);
560                    let transport = crate::runtime::ai::transport::AiTransport::from_runtime(self);
561                    let request = crate::ai::OpenAiEmbeddingRequest {
562                        api_key,
563                        model,
564                        inputs: vec![query_text.clone()],
565                        dimensions: None,
566                        api_base: provider.resolve_api_base(),
567                    };
568                    let response = crate::runtime::ai::block_on_ai(async move {
569                        crate::ai::openai_embeddings_async(&transport, request).await
570                    })
571                    .and_then(|result| result)?;
572                    response.embeddings.into_iter().next().ok_or_else(|| {
573                        RedDBError::Query("embedding API returned no vectors".to_string())
574                    })?
575                } else {
576                    vector.clone()
577                };
578                // Issue #119: route through AuthorizedSearch so the
579                // candidate set is gated by `EffectiveScope.visible_collections`
580                // before any similarity score is computed.
581                let scope = self.ai_scope();
582                let results =
583                    if super::statement_frame::ReadFrame::visible_collections(&scope).is_some() {
584                        crate::runtime::authorized_search::AuthorizedSearch::execute_similar(
585                            self,
586                            &scope,
587                            collection,
588                            &search_vector,
589                            *limit,
590                            *min_score,
591                        )?
592                    } else {
593                        // Embedded / no-auth caller: keep legacy behaviour.
594                        self.search_similar(collection, &search_vector, *limit, *min_score)?
595                    };
596                let mut result =
597                    UnifiedResult::with_columns(vec!["entity_id".into(), "score".into()]);
598                for sr in &results {
599                    let mut record = UnifiedRecord::new();
600                    record.set("entity_id", Value::UnsignedInteger(sr.entity_id.raw()));
601                    record.set("score", Value::Float(sr.score as f64));
602                    result.push(record);
603                }
604                Ok(RuntimeQueryResult {
605                    query: raw_query.to_string(),
606                    mode: QueryMode::Sql,
607                    statement: "search_similar",
608                    engine: "runtime-search",
609                    result,
610                    affected_rows: 0,
611                    statement_type: "select",
612                    bookmark: None,
613                    notice: None,
614                })
615            }
616            SearchCommand::Text {
617                query,
618                collection,
619                limit,
620                fuzzy,
621                limit_param,
622            } => {
623                if limit_param.is_some() {
624                    return Err(RedDBError::Query(
625                        "SEARCH TEXT LIMIT $N parameter was not bound before execution".to_string(),
626                    ));
627                }
628                let collections = collection.as_ref().map(|c| vec![c.clone()]);
629                // Issue #119: gate the candidate set by visible_collections.
630                let scope = self.ai_scope();
631                let res =
632                    if super::statement_frame::ReadFrame::visible_collections(&scope).is_some() {
633                        crate::runtime::authorized_search::AuthorizedSearch::execute_text(
634                            self,
635                            &scope,
636                            query.clone(),
637                            collections,
638                            None,
639                            None,
640                            None,
641                            Some(*limit),
642                            *fuzzy,
643                        )?
644                    } else {
645                        self.search_text(
646                            query.clone(),
647                            collections,
648                            None,
649                            None,
650                            None,
651                            Some(*limit),
652                            *fuzzy,
653                        )?
654                    };
655                let mut result =
656                    UnifiedResult::with_columns(vec!["entity_id".into(), "score".into()]);
657                for item in &res.matches {
658                    let mut record = UnifiedRecord::new();
659                    record.set("entity_id", Value::UnsignedInteger(item.entity.id.raw()));
660                    record.set("score", Value::Float(item.score as f64));
661                    result.push(record);
662                }
663                Ok(RuntimeQueryResult {
664                    query: raw_query.to_string(),
665                    mode: QueryMode::Sql,
666                    statement: "search_text",
667                    engine: "runtime-search",
668                    result,
669                    affected_rows: 0,
670                    statement_type: "select",
671                    bookmark: None,
672                    notice: None,
673                })
674            }
675            SearchCommand::Hybrid {
676                vector,
677                query,
678                collection,
679                limit,
680                limit_param,
681            } => {
682                if limit_param.is_some() {
683                    return Err(RedDBError::Query(
684                        "SEARCH HYBRID LIMIT $N parameter was not bound before execution"
685                            .to_string(),
686                    ));
687                }
688                let res = self.search_hybrid(
689                    vector.clone(),
690                    query.clone(),
691                    Some(*limit),
692                    Some(vec![collection.clone()]),
693                    None,
694                    None,
695                    None,
696                    Vec::new(),
697                    None,
698                    None,
699                    Some(*limit),
700                )?;
701                let mut result =
702                    UnifiedResult::with_columns(vec!["entity_id".into(), "score".into()]);
703                for item in &res.matches {
704                    let mut record = UnifiedRecord::new();
705                    record.set("entity_id", Value::UnsignedInteger(item.entity.id.raw()));
706                    record.set("score", Value::Float(item.score as f64));
707                    result.push(record);
708                }
709                Ok(RuntimeQueryResult {
710                    query: raw_query.to_string(),
711                    mode: QueryMode::Sql,
712                    statement: "search_hybrid",
713                    engine: "runtime-search",
714                    result,
715                    affected_rows: 0,
716                    statement_type: "select",
717                    bookmark: None,
718                    notice: None,
719                })
720            }
721            SearchCommand::Multimodal {
722                query,
723                collection,
724                limit,
725                limit_param,
726            } => {
727                if limit_param.is_some() {
728                    return Err(RedDBError::Query(
729                        "SEARCH MULTIMODAL LIMIT $N parameter was not bound before execution"
730                            .to_string(),
731                    ));
732                }
733                let collections = collection.as_ref().map(|c| vec![c.clone()]);
734                let res =
735                    self.search_multimodal(query.clone(), collections, None, None, Some(*limit))?;
736                let mut result =
737                    UnifiedResult::with_columns(vec!["entity_id".into(), "score".into()]);
738                for item in &res.matches {
739                    let mut record = UnifiedRecord::new();
740                    record.set("entity_id", Value::UnsignedInteger(item.entity.id.raw()));
741                    record.set("score", Value::Float(item.score as f64));
742                    result.push(record);
743                }
744                Ok(RuntimeQueryResult {
745                    query: raw_query.to_string(),
746                    mode: QueryMode::Sql,
747                    statement: "search_multimodal",
748                    engine: "runtime-search",
749                    result,
750                    affected_rows: 0,
751                    statement_type: "select",
752                    bookmark: None,
753                    notice: None,
754                })
755            }
756            SearchCommand::Index {
757                index,
758                value,
759                collection,
760                limit,
761                exact,
762                limit_param,
763            } => {
764                if limit_param.is_some() {
765                    return Err(RedDBError::Query(
766                        "SEARCH INDEX LIMIT $N parameter was not bound before execution"
767                            .to_string(),
768                    ));
769                }
770                let collections = collection.as_ref().map(|c| vec![c.clone()]);
771                let res = self.search_index(
772                    index.clone(),
773                    value.clone(),
774                    *exact,
775                    collections,
776                    None,
777                    None,
778                    Some(*limit),
779                )?;
780                let mut result =
781                    UnifiedResult::with_columns(vec!["entity_id".into(), "score".into()]);
782                for item in &res.matches {
783                    let mut record = UnifiedRecord::new();
784                    record.set("entity_id", Value::UnsignedInteger(item.entity.id.raw()));
785                    record.set("score", Value::Float(item.score as f64));
786                    result.push(record);
787                }
788                Ok(RuntimeQueryResult {
789                    query: raw_query.to_string(),
790                    mode: QueryMode::Sql,
791                    statement: "search_index",
792                    engine: "runtime-search",
793                    result,
794                    affected_rows: 0,
795                    statement_type: "select",
796                    bookmark: None,
797                    notice: None,
798                })
799            }
800            SearchCommand::Context {
801                query,
802                field,
803                collection,
804                limit,
805                depth,
806                limit_param,
807            } => {
808                if limit_param.is_some() {
809                    return Err(RedDBError::Query(
810                        "SEARCH CONTEXT LIMIT $N parameter was not bound before execution"
811                            .to_string(),
812                    ));
813                }
814                use crate::application::SearchContextInput;
815                // Issue #119: route through AuthorizedSearch so the
816                // candidate set + every expansion bucket is bounded by
817                // `EffectiveScope.visible_collections`.
818                let input = SearchContextInput {
819                    query: query.clone(),
820                    field: field.clone(),
821                    vector: None,
822                    collections: collection.as_ref().map(|c| vec![c.clone()]),
823                    graph_depth: Some(*depth),
824                    graph_max_edges: None,
825                    max_cross_refs: None,
826                    follow_cross_refs: None,
827                    expand_graph: None,
828                    global_scan: None,
829                    reindex: None,
830                    limit: Some(*limit),
831                    min_score: None,
832                };
833                let scope = self.ai_scope();
834                let res =
835                    if super::statement_frame::ReadFrame::visible_collections(&scope).is_some() {
836                        crate::runtime::authorized_search::AuthorizedSearch::execute_context(
837                            self, &scope, input,
838                        )?
839                    } else {
840                        self.search_context(input)?
841                    };
842                let mut result = UnifiedResult::with_columns(vec![
843                    "entity_id".into(),
844                    "collection".into(),
845                    "score".into(),
846                    "discovery".into(),
847                    "kind".into(),
848                ]);
849                let all_entities = res
850                    .tables
851                    .iter()
852                    .map(|e| (e, "table"))
853                    .chain(res.graph.nodes.iter().map(|e| (e, "graph_node")))
854                    .chain(res.graph.edges.iter().map(|e| (e, "graph_edge")))
855                    .chain(res.vectors.iter().map(|e| (e, "vector")))
856                    .chain(res.documents.iter().map(|e| (e, "document")))
857                    .chain(res.key_values.iter().map(|e| (e, "kv")));
858                for (entity, kind) in all_entities {
859                    let mut record = UnifiedRecord::new();
860                    record.set("entity_id", Value::UnsignedInteger(entity.entity.id.raw()));
861                    record.set("collection", Value::text(entity.collection.clone()));
862                    record.set("score", Value::Float(entity.score as f64));
863                    record.set("discovery", Value::text(format!("{:?}", entity.discovery)));
864                    record.set("kind", Value::text(kind.to_string()));
865                    result.push(record);
866                }
867                Ok(RuntimeQueryResult {
868                    query: raw_query.to_string(),
869                    mode: QueryMode::Sql,
870                    statement: "search_context",
871                    engine: "runtime-context",
872                    result,
873                    affected_rows: 0,
874                    statement_type: "select",
875                    bookmark: None,
876                    notice: None,
877                })
878            }
879            SearchCommand::SpatialRadius {
880                center_lat,
881                center_lon,
882                radius_km,
883                collection,
884                column,
885                limit,
886                limit_param,
887            } => {
888                if limit_param.is_some() {
889                    return Err(RedDBError::Query(
890                        "SEARCH SPATIAL RADIUS LIMIT $N parameter was not bound before execution"
891                            .to_string(),
892                    ));
893                }
894                use crate::geo::haversine_km;
895                // When the geo column carries an H3 index, gather candidates
896                // from a covering kRing over the disk B-tree cell ids; the
897                // ring is a provable superset of the in-radius rows, so the
898                // haversine post-filter below yields byte-identical results
899                // to the full scan. Without an H3 index, scan every row
900                // exactly as before (PRD #1574 slice 3).
901                let h3_candidates = self.h3_radius_candidate_ids(
902                    collection,
903                    column,
904                    *center_lat,
905                    *center_lon,
906                    *radius_km,
907                );
908                let store = self.inner.db.store();
909                let entities = scan_collection_with_candidates(&store, collection, &h3_candidates);
910
911                let mut hits: Vec<(u64, f64)> = Vec::new();
912                let mut geo_values_seen = 0usize;
913                for entity in &entities {
914                    if let Some((lat, lon)) =
915                        self.extract_spatial_column(entity, collection, column)
916                    {
917                        geo_values_seen += 1;
918                        let dist = haversine_km(*center_lat, *center_lon, lat, lon);
919                        if dist <= *radius_km {
920                            hits.push((entity.id.raw(), dist));
921                        }
922                    }
923                }
924                hits.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
925                hits.truncate(*limit);
926
927                let mut result =
928                    UnifiedResult::with_columns(vec!["entity_id".into(), "distance_km".into()]);
929                for (id, dist) in &hits {
930                    let mut record = UnifiedRecord::new();
931                    record.set("entity_id", Value::UnsignedInteger(*id));
932                    record.set("distance_km", Value::Float(*dist));
933                    result.push(record);
934                }
935                let notice = self.spatial_zero_geo_notice(
936                    collection,
937                    column,
938                    &entities,
939                    geo_values_seen,
940                    result.records.is_empty(),
941                );
942                Ok(RuntimeQueryResult {
943                    query: raw_query.to_string(),
944                    mode: QueryMode::Sql,
945                    statement: "search_spatial_radius",
946                    engine: "runtime-spatial",
947                    result,
948                    affected_rows: 0,
949                    statement_type: "select",
950                    bookmark: None,
951                    notice,
952                })
953            }
954            SearchCommand::SpatialBbox {
955                min_lat,
956                min_lon,
957                max_lat,
958                max_lon,
959                collection,
960                column,
961                limit,
962                limit_param,
963            } => {
964                if limit_param.is_some() {
965                    return Err(RedDBError::Query(
966                        "SEARCH SPATIAL BBOX LIMIT $N parameter was not bound before execution"
967                            .to_string(),
968                    ));
969                }
970                // Cover the bbox with a kRing big enough to enclose its
971                // farthest corner from the centre, scan those cells off the
972                // H3 disk index, then apply the exact bbox filter. The ring
973                // is a superset of the box, so results match the full scan.
974                let h3_candidates = self.h3_bbox_candidate_ids(
975                    collection, column, *min_lat, *min_lon, *max_lat, *max_lon,
976                );
977                let store = self.inner.db.store();
978                let entities = scan_collection_with_candidates(&store, collection, &h3_candidates);
979
980                let mut result = UnifiedResult::with_columns(vec!["entity_id".into()]);
981                let mut count = 0;
982                let mut geo_values_seen = 0usize;
983                for entity in &entities {
984                    if count >= *limit {
985                        break;
986                    }
987                    if let Some((lat, lon)) =
988                        self.extract_spatial_column(entity, collection, column)
989                    {
990                        geo_values_seen += 1;
991                        if lat >= *min_lat && lat <= *max_lat && lon >= *min_lon && lon <= *max_lon
992                        {
993                            let mut record = UnifiedRecord::new();
994                            record.set("entity_id", Value::UnsignedInteger(entity.id.raw()));
995                            result.push(record);
996                            count += 1;
997                        }
998                    }
999                }
1000                let notice = self.spatial_zero_geo_notice(
1001                    collection,
1002                    column,
1003                    &entities,
1004                    geo_values_seen,
1005                    result.records.is_empty(),
1006                );
1007                Ok(RuntimeQueryResult {
1008                    query: raw_query.to_string(),
1009                    mode: QueryMode::Sql,
1010                    statement: "search_spatial_bbox",
1011                    engine: "runtime-spatial",
1012                    result,
1013                    affected_rows: 0,
1014                    statement_type: "select",
1015                    bookmark: None,
1016                    notice,
1017                })
1018            }
1019            SearchCommand::SpatialWithinPolygon {
1020                vertices,
1021                collection,
1022                column,
1023                limit,
1024                limit_param,
1025            } => {
1026                if limit_param.is_some() {
1027                    return Err(RedDBError::Query(
1028                        "SEARCH SPATIAL WITHIN POLYGON LIMIT $N parameter was not bound before execution"
1029                            .to_string(),
1030                    ));
1031                }
1032                let h3_candidates = self.h3_polygon_candidate_ids(collection, column, vertices);
1033                let store = self.inner.db.store();
1034                let entities = scan_collection_with_candidates(&store, collection, &h3_candidates);
1035
1036                let mut result = UnifiedResult::with_columns(vec!["entity_id".into()]);
1037                let mut count = 0;
1038                for entity in &entities {
1039                    if count >= *limit {
1040                        break;
1041                    }
1042                    if let Some((lat, lon)) =
1043                        self.extract_spatial_column(entity, collection, column)
1044                    {
1045                        if crate::geo::point_in_polygon_even_odd(lat, lon, vertices) {
1046                            let mut record = UnifiedRecord::new();
1047                            record.set("entity_id", Value::UnsignedInteger(entity.id.raw()));
1048                            result.push(record);
1049                            count += 1;
1050                        }
1051                    }
1052                }
1053                Ok(RuntimeQueryResult {
1054                    query: raw_query.to_string(),
1055                    mode: QueryMode::Sql,
1056                    statement: "search_spatial_within_polygon",
1057                    engine: "runtime-spatial",
1058                    result,
1059                    affected_rows: 0,
1060                    statement_type: "select",
1061                    bookmark: None,
1062                    notice: None,
1063                })
1064            }
1065            SearchCommand::SpatialNearest {
1066                lat,
1067                lon,
1068                k,
1069                collection,
1070                column,
1071                k_param,
1072            } => {
1073                if k_param.is_some() {
1074                    return Err(RedDBError::Query(
1075                        "SEARCH SPATIAL NEAREST K $N parameter was not bound before execution"
1076                            .to_string(),
1077                    ));
1078                }
1079                use crate::geo::haversine_km;
1080                // Expand rings outward off the H3 disk index until the K-th
1081                // nearest candidate is provably closer than any unscanned
1082                // cell, then post-filter exactly as the full scan does. When
1083                // no H3 index covers the column (or the cover can't be
1084                // proven within the ring cap), fall back to the full scan so
1085                // results stay byte-identical (PRD #1574 slice 3).
1086                let h3_candidates =
1087                    self.h3_nearest_candidate_ids(collection, column, *lat, *lon, *k);
1088                let store = self.inner.db.store();
1089                let entities = scan_collection_with_candidates(&store, collection, &h3_candidates);
1090
1091                let mut hits: Vec<(u64, f64)> = Vec::new();
1092                let mut geo_values_seen = 0usize;
1093                for entity in &entities {
1094                    if let Some((elat, elon)) =
1095                        self.extract_spatial_column(entity, collection, column)
1096                    {
1097                        geo_values_seen += 1;
1098                        let dist = haversine_km(*lat, *lon, elat, elon);
1099                        hits.push((entity.id.raw(), dist));
1100                    }
1101                }
1102                hits.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
1103                hits.truncate(*k);
1104
1105                let mut result =
1106                    UnifiedResult::with_columns(vec!["entity_id".into(), "distance_km".into()]);
1107                for (id, dist) in &hits {
1108                    let mut record = UnifiedRecord::new();
1109                    record.set("entity_id", Value::UnsignedInteger(*id));
1110                    record.set("distance_km", Value::Float(*dist));
1111                    result.push(record);
1112                }
1113                let notice = self.spatial_zero_geo_notice(
1114                    collection,
1115                    column,
1116                    &entities,
1117                    geo_values_seen,
1118                    result.records.is_empty(),
1119                );
1120                Ok(RuntimeQueryResult {
1121                    query: raw_query.to_string(),
1122                    mode: QueryMode::Sql,
1123                    statement: "search_spatial_nearest",
1124                    engine: "runtime-spatial",
1125                    result,
1126                    affected_rows: 0,
1127                    statement_type: "select",
1128                    bookmark: None,
1129                    notice,
1130                })
1131            }
1132        }
1133    }
1134}
1135
1136fn apply_graph_order_and_limit(
1137    result: &mut UnifiedResult,
1138    statement: &str,
1139    order_by: Option<&GraphCommandOrderBy>,
1140    limit: Option<usize>,
1141) -> RedDBResult<()> {
1142    if let Some(order) = order_by {
1143        let column = graph_order_metric_column(statement, &order.metric)?;
1144        let columns = result.columns.clone();
1145        result.records.sort_by(|left, right| {
1146            let cmp = compare_graph_values(left.get(column), right.get(column));
1147            let cmp = if order.ascending { cmp } else { cmp.reverse() };
1148            if cmp == Ordering::Equal {
1149                compare_graph_rows(left, right, &columns)
1150            } else {
1151                cmp
1152            }
1153        });
1154    }
1155    if let Some(limit) = limit {
1156        result.records.truncate(limit);
1157    }
1158    Ok(())
1159}
1160
1161fn graph_order_metric_column(statement: &str, metric: &str) -> RedDBResult<&'static str> {
1162    let metric = metric.to_ascii_lowercase();
1163    match (statement, metric.as_str()) {
1164        ("graph_centrality", "score" | "centrality_score") => Ok("score"),
1165        ("graph_community", "size" | "community_size") => Ok("size"),
1166        ("graph_components", "size" | "component_size") => Ok("size"),
1167        ("graph_shortest_path", "hop_count" | "total_weight" | "nodes_visited") => {
1168            Ok(match metric.as_str() {
1169                "total_weight" => "total_weight",
1170                "nodes_visited" => "nodes_visited",
1171                _ => "hop_count",
1172            })
1173        }
1174        _ => Err(RedDBError::Query(format!(
1175            "unsupported ORDER BY metric '{metric}' for GRAPH {}",
1176            statement.trim_start_matches("graph_")
1177        ))),
1178    }
1179}
1180
1181fn compare_graph_rows(left: &UnifiedRecord, right: &UnifiedRecord, columns: &[String]) -> Ordering {
1182    for column in columns {
1183        let cmp = compare_graph_values(left.get(column), right.get(column));
1184        if cmp != Ordering::Equal {
1185            return cmp;
1186        }
1187    }
1188    Ordering::Equal
1189}
1190
1191fn compare_graph_values(left: Option<&Value>, right: Option<&Value>) -> Ordering {
1192    match (left, right) {
1193        (None, None) => Ordering::Equal,
1194        (None, Some(_)) => Ordering::Less,
1195        (Some(_), None) => Ordering::Greater,
1196        (Some(Value::Null), Some(Value::Null)) => Ordering::Equal,
1197        (Some(Value::Null), Some(_)) => Ordering::Less,
1198        (Some(_), Some(Value::Null)) => Ordering::Greater,
1199        (Some(Value::Integer(left)), Some(Value::Integer(right))) => left.cmp(right),
1200        (Some(Value::UnsignedInteger(left)), Some(Value::UnsignedInteger(right))) => {
1201            left.cmp(right)
1202        }
1203        (Some(Value::Float(left)), Some(Value::Float(right))) => {
1204            left.partial_cmp(right).unwrap_or(Ordering::Equal)
1205        }
1206        (Some(Value::Integer(left)), Some(Value::Float(right))) => {
1207            (*left as f64).partial_cmp(right).unwrap_or(Ordering::Equal)
1208        }
1209        (Some(Value::Float(left)), Some(Value::Integer(right))) => left
1210            .partial_cmp(&(*right as f64))
1211            .unwrap_or(Ordering::Equal),
1212        (Some(Value::UnsignedInteger(left)), Some(Value::Float(right))) => {
1213            (*left as f64).partial_cmp(right).unwrap_or(Ordering::Equal)
1214        }
1215        (Some(Value::Float(left)), Some(Value::UnsignedInteger(right))) => left
1216            .partial_cmp(&(*right as f64))
1217            .unwrap_or(Ordering::Equal),
1218        (Some(Value::Integer(left)), Some(Value::UnsignedInteger(right))) => {
1219            (*left as i128).cmp(&(*right as i128))
1220        }
1221        (Some(Value::UnsignedInteger(left)), Some(Value::Integer(right))) => {
1222            (*left as i128).cmp(&(*right as i128))
1223        }
1224        (Some(Value::Timestamp(left)), Some(Value::Timestamp(right))) => left.cmp(right),
1225        (Some(Value::Text(left)), Some(Value::Text(right))) => left.cmp(right),
1226        (Some(Value::Boolean(left)), Some(Value::Boolean(right))) => left.cmp(right),
1227        (Some(left), Some(right)) => format!("{left:?}").cmp(&format!("{right:?}")),
1228    }
1229}
1230
1231// =============================================================================
1232// Conversion helpers for string -> enum
1233// =============================================================================
1234
1235fn parse_direction(s: &str) -> RedDBResult<RuntimeGraphDirection> {
1236    match s.to_lowercase().as_str() {
1237        "outgoing" | "out" => Ok(RuntimeGraphDirection::Outgoing),
1238        "incoming" | "in" => Ok(RuntimeGraphDirection::Incoming),
1239        "both" | "any" => Ok(RuntimeGraphDirection::Both),
1240        _ => Err(RedDBError::Query(format!(
1241            "unknown direction: '{s}', expected outgoing|incoming|both"
1242        ))),
1243    }
1244}
1245
1246fn parse_path_algorithm(s: &str) -> RedDBResult<RuntimeGraphPathAlgorithm> {
1247    match s.to_lowercase().as_str() {
1248        "bfs" => Ok(RuntimeGraphPathAlgorithm::Bfs),
1249        "dijkstra" => Ok(RuntimeGraphPathAlgorithm::Dijkstra),
1250        "astar" | "a*" => Ok(RuntimeGraphPathAlgorithm::AStar),
1251        "bellman_ford" | "bellmanford" => Ok(RuntimeGraphPathAlgorithm::BellmanFord),
1252        _ => Err(RedDBError::Query(format!(
1253            "unknown path algorithm: '{s}', expected bfs|dijkstra|astar|bellman_ford"
1254        ))),
1255    }
1256}
1257
1258fn parse_traversal_strategy(s: &str) -> RedDBResult<RuntimeGraphTraversalStrategy> {
1259    match s.to_lowercase().as_str() {
1260        "bfs" => Ok(RuntimeGraphTraversalStrategy::Bfs),
1261        "dfs" => Ok(RuntimeGraphTraversalStrategy::Dfs),
1262        _ => Err(RedDBError::Query(format!(
1263            "unknown traversal strategy: '{s}', expected bfs|dfs"
1264        ))),
1265    }
1266}
1267
1268fn parse_centrality_algorithm(s: &str) -> RedDBResult<RuntimeGraphCentralityAlgorithm> {
1269    match s.to_lowercase().as_str() {
1270        "degree" => Ok(RuntimeGraphCentralityAlgorithm::Degree),
1271        "closeness" => Ok(RuntimeGraphCentralityAlgorithm::Closeness),
1272        "betweenness" => Ok(RuntimeGraphCentralityAlgorithm::Betweenness),
1273        "eigenvector" => Ok(RuntimeGraphCentralityAlgorithm::Eigenvector),
1274        "pagerank" | "page_rank" => Ok(RuntimeGraphCentralityAlgorithm::PageRank),
1275        _ => Err(RedDBError::Query(format!(
1276            "unknown centrality algorithm: '{s}', expected degree|closeness|betweenness|eigenvector|pagerank"
1277        ))),
1278    }
1279}
1280
1281fn parse_community_algorithm(s: &str) -> RedDBResult<RuntimeGraphCommunityAlgorithm> {
1282    match s.to_lowercase().as_str() {
1283        "label_propagation" | "labelpropagation" => {
1284            Ok(RuntimeGraphCommunityAlgorithm::LabelPropagation)
1285        }
1286        "louvain" => Ok(RuntimeGraphCommunityAlgorithm::Louvain),
1287        _ => Err(RedDBError::Query(format!(
1288            "unknown community algorithm: '{s}', expected label_propagation|louvain"
1289        ))),
1290    }
1291}
1292
1293fn parse_components_mode(s: &str) -> RedDBResult<RuntimeGraphComponentsMode> {
1294    match s.to_lowercase().as_str() {
1295        "connected" => Ok(RuntimeGraphComponentsMode::Connected),
1296        "weak" | "weakly_connected" => Ok(RuntimeGraphComponentsMode::Weak),
1297        "strong" | "strongly_connected" => Ok(RuntimeGraphComponentsMode::Strong),
1298        _ => Err(RedDBError::Query(format!(
1299            "unknown components mode: '{s}', expected connected|weak|strong"
1300        ))),
1301    }
1302}
1303
1304/// Extract (latitude, longitude) from an entity.
1305///
1306/// Looks for GeoPoint values in the entity data (row columns or node properties)
1307/// or dedicated lat/lon fields. Returns degrees.
1308fn extract_geo_from_entity(entity: &UnifiedEntity) -> Option<(f64, f64)> {
1309    match &entity.data {
1310        EntityData::Row(row) => {
1311            if let Some(ref named) = row.named {
1312                for value in named.values() {
1313                    if let Some(point) = crate::geo::recognize_geo_value(value) {
1314                        return Some(point);
1315                    }
1316                }
1317                if let Some(point) = crate::geo::recognize_geo_fields(|key| named.get(key)) {
1318                    return Some(point);
1319                }
1320            }
1321            for value in &row.columns {
1322                if let Some(point) = crate::geo::recognize_geo_value(value) {
1323                    return Some(point);
1324                }
1325            }
1326            None
1327        }
1328        EntityData::Node(node) => {
1329            for value in node.properties.values() {
1330                if let Some(point) = crate::geo::recognize_geo_value(value) {
1331                    return Some(point);
1332                }
1333            }
1334            crate::geo::recognize_geo_fields(|key| node.properties.get(key))
1335        }
1336        _ => None,
1337    }
1338}
1339
1340fn index_fields_from_entity(entity: &UnifiedEntity) -> Vec<(String, Value)> {
1341    match &entity.data {
1342        EntityData::Row(row) => row
1343            .iter_fields()
1344            .map(|(field, value)| (field.to_string(), value.clone()))
1345            .collect(),
1346        EntityData::Node(node) => node
1347            .properties
1348            .iter()
1349            .map(|(field, value)| (field.clone(), value.clone()))
1350            .collect(),
1351        EntityData::Edge(edge) => edge
1352            .properties
1353            .iter()
1354            .map(|(field, value)| (field.clone(), value.clone()))
1355            .collect(),
1356        _ => Vec::new(),
1357    }
1358}
1359
1360/// Collect the entities a spatial post-filter runs over. With an H3 cover
1361/// (`Some`), the collection scan is restricted to the candidate entity ids
1362/// while preserving `query_all` order, so the downstream stable sort and
1363/// early-LIMIT behaviour stay byte-identical to the full scan. Without a
1364/// cover (`None`), every row is scanned exactly as before (PRD #1574 slice 3).
1365fn scan_collection_with_candidates(
1366    store: &std::sync::Arc<crate::storage::unified::store::UnifiedStore>,
1367    collection: &str,
1368    candidates: &Option<std::collections::HashSet<u64>>,
1369) -> Vec<UnifiedEntity> {
1370    let resolver =
1371        crate::runtime::table_row_mvcc_resolver::TableRowMvccReadResolver::current_statement();
1372    match candidates {
1373        Some(ids) => store
1374            .get_collection(collection)
1375            .map(|m| {
1376                m.query_all(|e| {
1377                    ids.contains(&e.id.raw()) && resolver.resolve_read_candidate(e).is_some()
1378                })
1379            })
1380            .unwrap_or_default(),
1381        None => store
1382            .get_collection(collection)
1383            .map(|m| m.query_all(|e| resolver.resolve_read_candidate(e).is_some()))
1384            .unwrap_or_default(),
1385    }
1386}
1387
1388/// Cells covering a disk of `radius_km` around `(lat, lon)` at `resolution`.
1389///
1390/// Sizes the kRing as `ceil(radius_km / edge_km) + 1`: one grid step spans
1391/// at least one edge length, so the division over-estimates the ring count,
1392/// and the extra ring is the boundary margin that prevents the geohash-style
1393/// edge miss. Returns an empty vec — signalling the caller to fall back to
1394/// the full scan — for an un-encodable coordinate or a cover so large it
1395/// would be cheaper to scan the collection than to enumerate the cells.
1396fn h3_cover_cells(lat: f64, lon: f64, radius_km: f64, resolution: u8) -> Vec<u64> {
1397    let cell = crate::geo::h3::lat_lng_to_cell(lat, lon, resolution);
1398    if cell == 0 {
1399        return Vec::new();
1400    }
1401    let edge_km = crate::geo::h3::edge_length_km(resolution).max(f64::MIN_POSITIVE);
1402    // Cap the cover so a large radius over a fine resolution cannot blow up
1403    // into hundreds of millions of cells; beyond it the full scan is cheaper.
1404    const MAX_COVER_RING: u32 = 128;
1405    let k_f = (radius_km / edge_km).ceil() + 1.0;
1406    if !k_f.is_finite() || k_f > f64::from(MAX_COVER_RING) {
1407        return Vec::new();
1408    }
1409    crate::geo::h3::grid_disk(cell, k_f as u32)
1410}
1411
1412impl RedDBRuntime {
1413    fn extract_spatial_column(
1414        &self,
1415        entity: &UnifiedEntity,
1416        collection: &str,
1417        column: &str,
1418    ) -> Option<(f64, f64)> {
1419        let fields = index_fields_from_entity(entity);
1420        if let Some(value) = self
1421            .inner
1422            .index_store
1423            .resolve_index_field_value(&fields, column)
1424        {
1425            return crate::geo::recognize_geo_value(&value);
1426        }
1427
1428        let is_document_collection =
1429            self.inner
1430                .db
1431                .collection_contract(collection)
1432                .is_some_and(|contract| {
1433                    contract.declared_model == crate::catalog::CollectionModel::Document
1434                });
1435        if is_document_collection {
1436            return None;
1437        }
1438
1439        match entity.data {
1440            EntityData::Row(_) | EntityData::Node(_) => extract_geo_from_entity(entity),
1441            _ => None,
1442        }
1443    }
1444
1445    fn spatial_zero_geo_notice(
1446        &self,
1447        collection: &str,
1448        column: &str,
1449        scanned_entities: &[UnifiedEntity],
1450        geo_values_seen: usize,
1451        result_is_empty: bool,
1452    ) -> Option<String> {
1453        if !result_is_empty || geo_values_seen > 0 {
1454            return None;
1455        }
1456        let (entity_count, geo_count) = if scanned_entities.is_empty() {
1457            let store = self.inner.db.store();
1458            let entities = scan_collection_with_candidates(&store, collection, &None);
1459            let geo_count = entities
1460                .iter()
1461                .filter(|entity| {
1462                    self.extract_spatial_column(entity, collection, column)
1463                        .is_some()
1464                })
1465                .count();
1466            (entities.len(), geo_count)
1467        } else {
1468            (scanned_entities.len(), geo_values_seen)
1469        };
1470        if entity_count == 0 || geo_count > 0 {
1471            return None;
1472        }
1473        Some(format!(
1474            "no entity in '{}' has an indexable geo value in column '{}' (expected {}).",
1475            collection,
1476            column,
1477            crate::geo::RECOGNIZED_GEO_SHAPES
1478        ))
1479    }
1480
1481    /// Resolution of the H3 index registered on `(collection, column)`, if
1482    /// the column is covered by one. `None` for any other (or no) index.
1483    fn h3_index_resolution(&self, collection: &str, column: &str) -> Option<u8> {
1484        match self
1485            .inner
1486            .index_store
1487            .find_index_for_column(collection, column)?
1488            .method
1489        {
1490            super::index_store::IndexMethodKind::H3 { resolution } => Some(resolution),
1491            _ => None,
1492        }
1493    }
1494
1495    /// Map a set of H3 cell ids to the entity ids stored under them in the
1496    /// disk B-tree (sorted) index via per-cell point lookups. `None` when no
1497    /// cells were supplied or the column has no sorted index entry.
1498    fn h3_cell_candidate_ids(
1499        &self,
1500        collection: &str,
1501        column: &str,
1502        cells: &[u64],
1503    ) -> Option<std::collections::HashSet<u64>> {
1504        if cells.is_empty() {
1505            return None;
1506        }
1507        let keys: Vec<_> = cells
1508            .iter()
1509            .filter_map(|c| {
1510                crate::storage::schema::value_to_canonical_key(&Value::UnsignedInteger(*c))
1511            })
1512            .collect();
1513        let ids = self.inner.index_store.sorted.in_lookup_limited(
1514            collection,
1515            column,
1516            &keys,
1517            usize::MAX,
1518        )?;
1519        Some(ids.into_iter().map(|id| id.raw()).collect())
1520    }
1521
1522    /// Covering-kRing candidate ids for a RADIUS query, or `None` to fall
1523    /// back to the full scan (no H3 index, un-encodable centre, or a cover
1524    /// too large to enumerate).
1525    fn h3_radius_candidate_ids(
1526        &self,
1527        collection: &str,
1528        column: &str,
1529        center_lat: f64,
1530        center_lon: f64,
1531        radius_km: f64,
1532    ) -> Option<std::collections::HashSet<u64>> {
1533        let resolution = self.h3_index_resolution(collection, column)?;
1534        let cells = h3_cover_cells(center_lat, center_lon, radius_km, resolution);
1535        self.h3_cell_candidate_ids(collection, column, &cells)
1536    }
1537
1538    /// Covering-kRing candidate ids for a BBOX query: cover a disk centred on
1539    /// the box whose radius reaches its farthest corner, then let the exact
1540    /// bbox filter trim it. `None` falls back to the full scan.
1541    fn h3_bbox_candidate_ids(
1542        &self,
1543        collection: &str,
1544        column: &str,
1545        min_lat: f64,
1546        min_lon: f64,
1547        max_lat: f64,
1548        max_lon: f64,
1549    ) -> Option<std::collections::HashSet<u64>> {
1550        let resolution = self.h3_index_resolution(collection, column)?;
1551        let center_lat = (min_lat + max_lat) / 2.0;
1552        let center_lon = (min_lon + max_lon) / 2.0;
1553        let radius_km = [
1554            (min_lat, min_lon),
1555            (min_lat, max_lon),
1556            (max_lat, min_lon),
1557            (max_lat, max_lon),
1558        ]
1559        .into_iter()
1560        .map(|(la, lo)| crate::geo::haversine_km(center_lat, center_lon, la, lo))
1561        .fold(0.0_f64, f64::max);
1562        let cells = h3_cover_cells(center_lat, center_lon, radius_km, resolution);
1563        self.h3_cell_candidate_ids(collection, column, &cells)
1564    }
1565
1566    /// Native H3 polygon-to-cells candidate ids for a WITHIN POLYGON query.
1567    /// The H3 cover is only a pruning superset; exact even-odd
1568    /// point-in-polygon filtering decides correctness. `None` falls back to
1569    /// the full scan when there is no H3 index or the cover is too large.
1570    fn h3_polygon_candidate_ids(
1571        &self,
1572        collection: &str,
1573        column: &str,
1574        vertices: &[(f64, f64)],
1575    ) -> Option<std::collections::HashSet<u64>> {
1576        let resolution = self.h3_index_resolution(collection, column)?;
1577        let cells = crate::geo::h3::polygon_to_cover_cells(
1578            vertices,
1579            resolution,
1580            crate::geo::h3::MAX_POLYGON_COVER_CELLS,
1581        )?;
1582        self.h3_cell_candidate_ids(collection, column, &cells)
1583    }
1584
1585    /// Ring-expansion candidate ids for a NEAREST-K query. Expands the cover
1586    /// outward until the K-th nearest candidate is provably closer than any
1587    /// point in an unscanned cell, returning that superset. `None` (→ full
1588    /// scan) when the column has no H3 index, the centre is un-encodable, or
1589    /// the proof is not reached within the ring cap (e.g. fewer than K rows).
1590    fn h3_nearest_candidate_ids(
1591        &self,
1592        collection: &str,
1593        column: &str,
1594        lat: f64,
1595        lon: f64,
1596        k: usize,
1597    ) -> Option<std::collections::HashSet<u64>> {
1598        let resolution = self.h3_index_resolution(collection, column)?;
1599        if k == 0 {
1600            return None;
1601        }
1602        let center = crate::geo::h3::lat_lng_to_cell(lat, lon, resolution);
1603        if center == 0 {
1604            return None;
1605        }
1606        let edge_km = crate::geo::h3::edge_length_km(resolution).max(f64::MIN_POSITIVE);
1607        // Bound the expansion so an index holding fewer than K points cannot
1608        // loop forever; past the cap the exact full scan takes over.
1609        const MAX_RING: u32 = 64;
1610        let store = self.inner.db.store();
1611        for r in 0..=MAX_RING {
1612            let cells = crate::geo::h3::grid_disk(center, r);
1613            let Some(ids) = self.h3_cell_candidate_ids(collection, column, &cells) else {
1614                continue;
1615            };
1616            if ids.len() < k {
1617                continue;
1618            }
1619            let candidates = Some(ids);
1620            let mut dists: Vec<f64> =
1621                scan_collection_with_candidates(&store, collection, &candidates)
1622                    .iter()
1623                    .filter_map(|e| {
1624                        extract_geo_from_entity(e)
1625                            .map(|(elat, elon)| crate::geo::haversine_km(lat, lon, elat, elon))
1626                    })
1627                    .collect();
1628            if dists.len() < k {
1629                continue;
1630            }
1631            dists.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
1632            let d_k = dists[k - 1];
1633            // A point in any unscanned cell (grid distance > r) lies at least
1634            // `r * edge_km` from the centre — a conservative lower bound. Once
1635            // that already meets or exceeds the current K-th distance, the
1636            // cover is a proven superset of the true nearest-K.
1637            if f64::from(r) * edge_km >= d_k {
1638                return candidates;
1639            }
1640        }
1641        None
1642    }
1643}