1use super::*;
8use crate::storage::query::ast::GraphCommandOrderBy;
9use std::cmp::Ordering;
10
11impl RedDBRuntime {
12 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 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 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 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 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 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 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 let search_vector = if let Some(query_text) = text {
537 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 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 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 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 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 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 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 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 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
1231fn 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
1304fn 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
1360fn 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
1388fn 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 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 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 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 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 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 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 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 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 if f64::from(r) * edge_km >= d_k {
1638 return candidates;
1639 }
1640 }
1641 None
1642 }
1643}