linkrs_query/executor/streaming/operators/vector_operator.rs
1use std::sync::Arc;
2
3use parking_lot::RwLock;
4
5use crate::executor::streaming::chunk::DataChunk;
6use crate::executor::streaming::executor::StreamingExecutor;
7use crate::executor::streaming::operators::source_operator::OperatorConfig;
8use crate::executor::streaming::operators::spec::VectorManageCommand;
9use crate::executor::streaming::runtime::ExecutionRuntime;
10use crate::executor::streaming::slot::SlotLayout;
11use crate::storage::QueryStorage;
12use linkrs_core::error::QueryError;
13#[cfg(feature = "vector")]
14use linkrs_core::Value;
15#[cfg(feature = "vector")]
16use linkrs_sync::VectorSyncCoordinator;
17
18#[cfg(feature = "vector")]
19fn make_manage_result(
20 output_layout: Arc<SlotLayout>,
21 action: &str,
22 name: Option<&str>,
23 status: &str,
24) -> DataChunk {
25 let name_val = name
26 .map(Value::string)
27 .unwrap_or(Value::Null(linkrs_core::NullType::Null));
28 DataChunk::new_with_layout(
29 vec![vec![Value::string(action), name_val, Value::string(status)]],
30 output_layout,
31 )
32}
33
34/// Candidate count for MATCH VECTOR searches.
35///
36/// The `MATCH VECTOR` grammar has no LIMIT clause yet (see the parser AST),
37/// so a fixed candidate window is used; wire this to syntax once LIMIT is
38/// added to the statement form.
39#[cfg(feature = "vector")]
40const DEFAULT_MATCH_TOP_K: usize = 100;
41
42#[derive(Debug)]
43pub enum VectorOperatorKind {
44 VectorManage {
45 storage: Option<Arc<RwLock<dyn QueryStorage>>>,
46 space_name: String,
47 command: VectorManageCommand,
48 #[cfg(feature = "vector")]
49 vector_coordinator: Option<Arc<VectorSyncCoordinator>>,
50 },
51 VectorSearch {
52 storage: Option<Arc<RwLock<dyn QueryStorage>>>,
53 space_name: String,
54 space_id: u64,
55 index_name: String,
56 query_vector: Vec<f32>,
57 /// Raw text for TEXT queries; resolved to a vector at execution time.
58 query_text: Option<String>,
59 top_k: u32,
60 tag_name: String,
61 field_name: String,
62 threshold: Option<f32>,
63 filter: Option<super::spec::SpecVectorFilter>,
64 offset: usize,
65 #[cfg(feature = "vector")]
66 ryw_config: Option<linkrs_core::types::ReadYourWritesConfig>,
67 #[cfg(feature = "vector")]
68 vector_coordinator: Option<Arc<VectorSyncCoordinator>>,
69 },
70 VectorLookup {
71 storage: Option<Arc<RwLock<dyn QueryStorage>>>,
72 space_name: String,
73 space_id: u64,
74 index_name: String,
75 query_vector: Vec<f32>,
76 /// Raw text for TEXT queries; resolved to a vector at execution time.
77 query_text: Option<String>,
78 top_k: u32,
79 tag_name: String,
80 field_name: String,
81 #[cfg(feature = "vector")]
82 ryw_config: Option<linkrs_core::types::ReadYourWritesConfig>,
83 #[cfg(feature = "vector")]
84 vector_coordinator: Option<Arc<VectorSyncCoordinator>>,
85 },
86 VectorMatch {
87 storage: Option<Arc<RwLock<dyn QueryStorage>>>,
88 space_name: String,
89 pattern: String,
90 field: String,
91 query_vector: Vec<f32>,
92 /// Raw text for TEXT queries; resolved to a vector at execution time.
93 query_text: Option<String>,
94 threshold: Option<f32>,
95 tag_name: String,
96 field_name: String,
97 space_id: u64,
98 #[cfg(feature = "vector")]
99 ryw_config: Option<linkrs_core::types::ReadYourWritesConfig>,
100 #[cfg(feature = "vector")]
101 vector_coordinator: Option<Arc<VectorSyncCoordinator>>,
102 },
103}
104
105/// Vector operator.
106///
107/// Wraps [`VectorOperatorKind`] with the runtime context injected at `open()`.
108/// Lifecycle state is owned exclusively by the executor; operators never
109/// write it.
110#[derive(Debug)]
111pub struct VectorOperator {
112 pub kind: VectorOperatorKind,
113 pub runtime: Option<Arc<ExecutionRuntime>>,
114 pub output_layout: Arc<SlotLayout>,
115 pub config: OperatorConfig,
116}
117
118impl VectorOperator {
119 /// Create a VectorOperator from an immutable spec.
120 pub fn from_spec(
121 spec: &super::spec::VectorSpec,
122 storage: Option<Arc<RwLock<dyn QueryStorage>>>,
123 #[cfg(feature = "vector")] vector_coordinator: Option<Arc<VectorSyncCoordinator>>,
124 output_layout: Arc<SlotLayout>,
125 ) -> Self {
126 let kind = match spec {
127 super::spec::VectorSpec::VectorManage {
128 space_name,
129 command,
130 } => VectorOperatorKind::VectorManage {
131 storage: storage.clone(),
132 space_name: space_name.clone(),
133 command: command.clone(),
134 #[cfg(feature = "vector")]
135 vector_coordinator: vector_coordinator.clone(),
136 },
137 super::spec::VectorSpec::VectorSearch {
138 space_name,
139 space_id,
140 index_name,
141 query_vector,
142 query_text,
143 top_k,
144 tag_name,
145 field_name,
146 threshold,
147 filter,
148 offset,
149 ..
150 } => VectorOperatorKind::VectorSearch {
151 storage: storage.clone(),
152 space_name: space_name.clone(),
153 space_id: *space_id,
154 index_name: index_name.clone(),
155 query_vector: query_vector.clone(),
156 query_text: query_text.clone(),
157 top_k: *top_k,
158 tag_name: tag_name.clone(),
159 field_name: field_name.clone(),
160 threshold: *threshold,
161 filter: filter.clone(),
162 offset: *offset,
163 #[cfg(feature = "vector")]
164 ryw_config: spec.ryw_config(),
165 #[cfg(feature = "vector")]
166 vector_coordinator: vector_coordinator.clone(),
167 },
168 super::spec::VectorSpec::VectorLookup {
169 space_name,
170 space_id,
171 index_name,
172 query_vector,
173 query_text,
174 top_k,
175 tag_name,
176 field_name,
177 ..
178 } => VectorOperatorKind::VectorLookup {
179 storage: storage.clone(),
180 space_name: space_name.clone(),
181 space_id: *space_id,
182 index_name: index_name.clone(),
183 query_vector: query_vector.clone(),
184 query_text: query_text.clone(),
185 top_k: *top_k,
186 tag_name: tag_name.clone(),
187 field_name: field_name.clone(),
188 #[cfg(feature = "vector")]
189 ryw_config: spec.ryw_config(),
190 #[cfg(feature = "vector")]
191 vector_coordinator: vector_coordinator.clone(),
192 },
193 super::spec::VectorSpec::VectorMatch {
194 space_name,
195 pattern,
196 field,
197 query_vector,
198 query_text,
199 threshold,
200 tag_name,
201 field_name,
202 space_id,
203 ..
204 } => VectorOperatorKind::VectorMatch {
205 storage: storage.clone(),
206 space_name: space_name.clone(),
207 pattern: pattern.clone(),
208 field: field.clone(),
209 query_vector: query_vector.clone(),
210 query_text: query_text.clone(),
211 threshold: *threshold,
212 tag_name: tag_name.clone(),
213 field_name: field_name.clone(),
214 space_id: *space_id,
215 #[cfg(feature = "vector")]
216 ryw_config: spec.ryw_config(),
217 #[cfg(feature = "vector")]
218 vector_coordinator: vector_coordinator.clone(),
219 },
220 };
221 Self::new(kind, output_layout)
222 }
223
224 pub fn new(kind: VectorOperatorKind, output_layout: Arc<SlotLayout>) -> Self {
225 Self {
226 kind,
227 runtime: None,
228 output_layout,
229 config: OperatorConfig::default(),
230 }
231 }
232
233 /// Inject the runtime and execution config (called once by the executor
234 /// before this operator produces any data).
235 pub fn inject_context(
236 &mut self,
237 runtime: Option<&Arc<ExecutionRuntime>>,
238 config: OperatorConfig,
239 ) {
240 if let Some(rt) = runtime {
241 self.runtime = Some(rt.clone());
242 }
243 self.config = config;
244 }
245
246 pub fn open(&mut self, input: &mut StreamingExecutor) -> Result<(), QueryError> {
247 match &mut self.kind {
248 VectorOperatorKind::VectorManage { .. }
249 | VectorOperatorKind::VectorSearch { .. }
250 | VectorOperatorKind::VectorLookup { .. }
251 | VectorOperatorKind::VectorMatch { .. } => {
252 input.open()?;
253 Ok(())
254 }
255 }
256 }
257
258 pub fn next(&mut self, input: &mut StreamingExecutor) -> Result<Option<DataChunk>, QueryError> {
259 match &mut self.kind {
260 VectorOperatorKind::VectorManage {
261 storage,
262 space_name,
263 command,
264 #[cfg(feature = "vector")]
265 vector_coordinator,
266 } => {
267 #[cfg(feature = "vector")]
268 let _ = (&storage, &space_name);
269
270 match command {
271 VectorManageCommand::Create {
272 index_name,
273 tag_name,
274 field_name,
275 vector_size,
276 distance,
277 space_id,
278 hnsw_m,
279 hnsw_ef_construct,
280 quantization,
281 quantile,
282 compression,
283 always_ram,
284 } => {
285 #[cfg(feature = "vector")]
286 {
287 if let Some(coordinator) = vector_coordinator {
288 // Vector indexes are vertex-only: point IDs,
289 // payloads, and rebuild sources are all modeled
290 // per vertex. Reject edge type names here as well
291 // so the no-metadata query path cannot create an
292 // index that can never receive data. A storage
293 // error fails open: the planner already rejects
294 // known edge types, and an unreadable catalog
295 // must not block vertex index creation.
296 if let Some(storage) = storage.as_ref() {
297 if !tag_name.is_empty() {
298 if let Ok(edge) =
299 storage.read().get_edge_type(space_name, tag_name)
300 {
301 if edge.is_some() {
302 return Err(QueryError::execution(format!(
303 "Vector indexes are vertex-only: '{}' is an edge type, create the index on a vertex tag instead",
304 tag_name
305 )));
306 }
307 }
308 }
309 }
310 let distance = match distance {
311 crate::parser::ast::vector::VectorDistance::Cosine => {
312 simvec::DistanceMetric::Cosine
313 }
314 crate::parser::ast::vector::VectorDistance::Euclidean => {
315 simvec::DistanceMetric::Euclid
316 }
317 crate::parser::ast::vector::VectorDistance::Dot => {
318 simvec::DistanceMetric::Dot
319 }
320 crate::parser::ast::vector::VectorDistance::Manhattan => {
321 simvec::DistanceMetric::Manhattan
322 }
323 };
324 // Build CollectionConfig with optional HNSW and quantization
325 // (mirrors Qdrant scalar/product/binary builders).
326 let mut config =
327 simvec::CollectionConfig::new(*vector_size, distance);
328 if hnsw_m.is_some() || hnsw_ef_construct.is_some() {
329 let mut hnsw = simvec::HnswConfig::default();
330 if let Some(m) = hnsw_m {
331 hnsw.m = *m;
332 }
333 if let Some(ef) = hnsw_ef_construct {
334 hnsw.ef_construct = *ef;
335 }
336 hnsw.validate().map_err(|e| {
337 QueryError::execution(format!(
338 "invalid HNSW parameters: {}",
339 e
340 ))
341 })?;
342 config = config.with_hnsw(hnsw);
343 }
344 if let Some(qkind) = quantization {
345 if compression.is_some()
346 && !matches!(
347 qkind,
348 crate::parser::ast::vector::QuantizationKind::Product
349 )
350 {
351 return Err(QueryError::execution(format!(
352 "compression requires quantization=PRODUCT, got quantization={:?}; omit compression or use PRODUCT",
353 qkind
354 )));
355 }
356 let quant_cfg = match qkind {
357 crate::parser::ast::vector::QuantizationKind::Scalar => {
358 let mut cfg = simvec::QuantizationConfig::scalar(
359 quantile.unwrap_or(0.99),
360 );
361 if let Some(ar) = always_ram {
362 cfg = cfg.with_always_ram(*ar);
363 }
364 cfg
365 }
366 crate::parser::ast::vector::QuantizationKind::Binary => {
367 let mut cfg = simvec::QuantizationConfig::binary();
368 if let Some(ar) = always_ram {
369 cfg = cfg.with_always_ram(*ar);
370 }
371 cfg
372 }
373 crate::parser::ast::vector::QuantizationKind::Product => {
374 let ratio = match compression {
375 Some(crate::parser::ast::vector::CompressionRatioKind::X4) => {
376 simvec::CompressionRatio::X4
377 }
378 Some(crate::parser::ast::vector::CompressionRatioKind::X8) => {
379 simvec::CompressionRatio::X8
380 }
381 Some(crate::parser::ast::vector::CompressionRatioKind::X16) => {
382 simvec::CompressionRatio::X16
383 }
384 Some(crate::parser::ast::vector::CompressionRatioKind::X32) => {
385 simvec::CompressionRatio::X32
386 }
387 Some(crate::parser::ast::vector::CompressionRatioKind::X64) => {
388 simvec::CompressionRatio::X64
389 }
390 None => simvec::CompressionRatio::X4,
391 };
392 let mut cfg =
393 simvec::QuantizationConfig::product(ratio);
394 if let Some(ar) = always_ram {
395 cfg = cfg.with_always_ram(*ar);
396 }
397 cfg
398 }
399 };
400 quant_cfg.validate(*vector_size).map_err(|e| {
401 QueryError::execution(format!(
402 "invalid quantization parameters: {}",
403 e
404 ))
405 })?;
406 config = config.with_quantization(quant_cfg);
407 }
408 let res = if config.quantization_config.is_some()
409 || config.hnsw_config.is_some()
410 {
411 crate::executor::streaming::helpers::runtime_bridge::wait(
412 "Vector create",
413 coordinator.create_index_with_config(
414 *space_id, tag_name, field_name, config,
415 ),
416 )
417 } else {
418 crate::executor::streaming::helpers::runtime_bridge::wait(
419 "Vector create",
420 coordinator.create_vector_index(
421 *space_id,
422 tag_name,
423 field_name,
424 *vector_size,
425 distance,
426 ),
427 )
428 };
429 match res {
430 Ok(_) => {
431 // Record the statement-level name so
432 // later SEARCH/LOOKUP/DROP statements
433 // resolve it back to (space, tag, field).
434 coordinator.set_index_name(
435 *space_id, tag_name, field_name, index_name,
436 );
437 Ok(Some(make_manage_result(
438 Arc::clone(&self.output_layout),
439 "create_vector_index",
440 Some(index_name.as_str()),
441 "created",
442 )))
443 }
444 Err(e) => Err(e),
445 }
446 } else {
447 Err(QueryError::execution(
448 "CREATE VECTOR INDEX cannot execute: no vector coordinator is configured",
449 ))
450 }
451 }
452 #[cfg(not(feature = "vector"))]
453 {
454 let _ = (
455 storage,
456 space_name,
457 index_name,
458 tag_name,
459 field_name,
460 vector_size,
461 distance,
462 space_id,
463 hnsw_m,
464 hnsw_ef_construct,
465 quantization,
466 quantile,
467 compression,
468 always_ram,
469 );
470 Err(QueryError::feature_disabled(
471 "vector",
472 "CREATE VECTOR INDEX",
473 ))
474 }
475 }
476 VectorManageCommand::Drop {
477 index_name,
478 if_exists,
479 space_id,
480 tag_name,
481 field_name,
482 } => {
483 #[cfg(feature = "vector")]
484 {
485 // The drop API is addressed by (space_id, tag,
486 // field). An unresolved location means the index
487 // was not found at planning time: `IF EXISTS`
488 // degrades to a no-op status row, otherwise it is
489 // a clear error.
490 if tag_name.is_empty() || field_name.is_empty() {
491 if *if_exists {
492 return Ok(Some(make_manage_result(
493 Arc::clone(&self.output_layout),
494 "drop_vector_index",
495 Some(index_name.as_str()),
496 "not_exists",
497 )));
498 }
499 return Err(QueryError::execution(format!(
500 "Vector index '{}' cannot be dropped: index location (tag/field) is not resolved",
501 index_name
502 )));
503 }
504 match vector_coordinator {
505 Some(coordinator) => {
506 crate::executor::streaming::helpers::runtime_bridge::wait(
507 "Vector drop",
508 coordinator
509 .drop_vector_index(*space_id, tag_name, field_name),
510 )?;
511 Ok(Some(make_manage_result(
512 Arc::clone(&self.output_layout),
513 "drop_vector_index",
514 Some(index_name.as_str()),
515 "dropped",
516 )))
517 }
518 None => Err(QueryError::execution(
519 "DROP VECTOR INDEX cannot execute: no vector coordinator is configured",
520 )),
521 }
522 }
523 #[cfg(not(feature = "vector"))]
524 {
525 let _ = (
526 storage, space_name, index_name, if_exists, space_id, tag_name,
527 field_name,
528 );
529 Err(QueryError::feature_disabled("vector", "DROP VECTOR INDEX"))
530 }
531 }
532 }
533 }
534
535 VectorOperatorKind::VectorSearch {
536 storage: _,
537 space_name: _,
538 space_id,
539 index_name: _,
540 query_vector,
541 query_text,
542 top_k,
543 tag_name,
544 field_name,
545 threshold,
546 filter,
547 offset,
548 #[cfg(feature = "vector")]
549 ryw_config,
550 #[cfg(feature = "vector")]
551 vector_coordinator,
552 ..
553 } => {
554 #[cfg(feature = "vector")]
555 {
556 if let Some(coordinator) = vector_coordinator {
557 // The planner resolves index_name to (tag, field)
558 // via the index registry; an empty location is an
559 // unknown index, not an empty match set.
560 if tag_name.is_empty() || field_name.is_empty() {
561 return Err(QueryError::execution(
562 "SEARCH VECTOR cannot execute: index location (tag/field) is not resolved",
563 ));
564 }
565 // Resolve TEXT query to a vector at execution time via
566 // the embedding service (deferred embed pattern).
567 let resolved_vector = if let Some(text) = query_text {
568 #[cfg(feature = "embedding")]
569 {
570 let embedded =
571 crate::executor::streaming::helpers::runtime_bridge::wait(
572 "TEXT vector embed",
573 coordinator.embed_text(text),
574 )
575 .map_err(|e| {
576 QueryError::execution(format!(
577 "TEXT vector embed failed: {}",
578 e
579 ))
580 })?;
581 embedded
582 }
583 #[cfg(not(feature = "embedding"))]
584 {
585 let _ = text;
586 return Err(QueryError::execution(
587 "TEXT vector query requires the 'embedding' feature to be enabled",
588 ));
589 }
590 } else {
591 query_vector.clone()
592 };
593 // Fetch enough candidates so skipping `offset` rows
594 // still leaves up to `top_k` results. An active
595 // rerank stage widens recall inside the coordinator
596 // and truncates back to this limit.
597 let recall_limit = (*top_k as usize).saturating_add(*offset);
598 let mut options = linkrs_sync::vector_sync::SearchOptions::new(
599 *space_id,
600 tag_name.clone(),
601 field_name.clone(),
602 resolved_vector,
603 recall_limit,
604 );
605 options.query_text = query_text.clone();
606 // A zero threshold is vacuous for similarity scores;
607 // keep it unset so behavior matches the no-THRESHOLD
608 // statement form.
609 match threshold {
610 Some(t) if *t > 0.0 => options.threshold = Some(*t),
611 _ => {}
612 }
613 if let Some(filter) = filter {
614 options.filter = Some(filter.clone());
615 }
616 if let Some(cfg) = ryw_config {
617 options.consistency =
618 linkrs_sync::vector_sync::SearchConsistency::ReadYourWrites {
619 timeout_ms: cfg.timeout_ms,
620 };
621 options.minimum_lsn = cfg.minimum_lsn;
622 }
623 let search_results =
624 crate::executor::streaming::helpers::runtime_bridge::wait(
625 "Vector search",
626 coordinator.search_with_options(options),
627 )?;
628 let mut rows = Vec::new();
629 for result in search_results.into_iter().skip(*offset) {
630 rows.push(vec![
631 Value::string(result.id.to_string()),
632 Value::Double(result.score as f64),
633 ]);
634 }
635 return if rows.is_empty() {
636 Ok(None)
637 } else {
638 Ok(Some(DataChunk::new_with_layout(
639 rows,
640 self.output_layout.clone(),
641 )))
642 };
643 }
644
645 // A missing coordinator is a configuration error, not an
646 // empty result: fail loudly instead of silently returning
647 // nothing (or unrelated input rows).
648 let _ = input;
649 Err(QueryError::execution(
650 "SEARCH VECTOR cannot execute: no vector coordinator is configured",
651 ))
652 }
653
654 #[cfg(not(feature = "vector"))]
655 {
656 let _ = (
657 &space_id,
658 &tag_name,
659 &field_name,
660 &query_vector,
661 query_text.as_deref(),
662 &top_k,
663 &threshold,
664 &filter,
665 &offset,
666 input,
667 );
668 Err(QueryError::feature_disabled("vector", "VECTOR SEARCH"))
669 }
670 }
671
672 VectorOperatorKind::VectorLookup {
673 storage: _,
674 space_name: _,
675 space_id,
676 index_name: _,
677 query_vector,
678 query_text,
679 top_k,
680 tag_name,
681 field_name,
682 #[cfg(feature = "vector")]
683 ryw_config,
684 #[cfg(feature = "vector")]
685 vector_coordinator,
686 ..
687 } => {
688 #[cfg(feature = "vector")]
689 {
690 if let Some(coordinator) = vector_coordinator {
691 // LOOKUP VECTOR resolves to the same index location
692 // as SEARCH VECTOR and reuses the identical search
693 // path, producing (id, score) rows.
694 if tag_name.is_empty() || field_name.is_empty() {
695 return Err(QueryError::execution(
696 "LOOKUP VECTOR cannot execute: index location (tag/field) is not resolved",
697 ));
698 }
699 // Resolve TEXT query to a vector at execution time.
700 let resolved_vector = if let Some(text) = query_text {
701 #[cfg(feature = "embedding")]
702 {
703 let embedded =
704 crate::executor::streaming::helpers::runtime_bridge::wait(
705 "TEXT vector embed",
706 coordinator.embed_text(text),
707 )
708 .map_err(|e| {
709 QueryError::execution(format!(
710 "TEXT vector embed failed: {}",
711 e
712 ))
713 })?;
714 embedded
715 }
716 #[cfg(not(feature = "embedding"))]
717 {
718 let _ = text;
719 return Err(QueryError::execution(
720 "TEXT vector query requires the 'embedding' feature to be enabled",
721 ));
722 }
723 } else {
724 query_vector.clone()
725 };
726 // The coordinator widens recall for an active rerank
727 // stage and truncates back to `top_k`.
728 let mut options = linkrs_sync::vector_sync::SearchOptions::new(
729 *space_id,
730 tag_name.clone(),
731 field_name.clone(),
732 resolved_vector,
733 *top_k as usize,
734 );
735 options.query_text = query_text.clone();
736 if let Some(cfg) = ryw_config {
737 options.consistency =
738 linkrs_sync::vector_sync::SearchConsistency::ReadYourWrites {
739 timeout_ms: cfg.timeout_ms,
740 };
741 options.minimum_lsn = cfg.minimum_lsn;
742 }
743 let search_results =
744 crate::executor::streaming::helpers::runtime_bridge::wait(
745 "Vector lookup",
746 coordinator.search_with_options(options),
747 )?;
748 let mut rows = Vec::new();
749 for result in search_results {
750 rows.push(vec![
751 Value::string(result.id.to_string()),
752 Value::Double(result.score as f64),
753 ]);
754 }
755 return if rows.is_empty() {
756 Ok(None)
757 } else {
758 Ok(Some(DataChunk::new_with_layout(
759 rows,
760 self.output_layout.clone(),
761 )))
762 };
763 }
764
765 let _ = input;
766 Err(QueryError::execution(
767 "LOOKUP VECTOR cannot execute: no vector coordinator is configured",
768 ))
769 }
770
771 #[cfg(not(feature = "vector"))]
772 {
773 let _ = (
774 &space_id,
775 &tag_name,
776 &field_name,
777 &query_vector,
778 query_text.as_deref(),
779 &top_k,
780 );
781 Err(QueryError::feature_disabled("vector", "VECTOR LOOKUP"))
782 }
783 }
784
785 VectorOperatorKind::VectorMatch {
786 storage: _,
787 space_name: _,
788 pattern: _,
789 field: _,
790 query_vector,
791 query_text,
792 threshold,
793 tag_name,
794 field_name,
795 space_id,
796 #[cfg(feature = "vector")]
797 ryw_config,
798 #[cfg(feature = "vector")]
799 vector_coordinator,
800 ..
801 } => {
802 #[cfg(feature = "vector")]
803 {
804 if let Some(coordinator) = vector_coordinator {
805 // Resolve TEXT query to a vector at execution time.
806 let resolved_vector = if let Some(text) = query_text {
807 #[cfg(feature = "embedding")]
808 {
809 let embedded =
810 crate::executor::streaming::helpers::runtime_bridge::wait(
811 "TEXT vector embed",
812 coordinator.embed_text(text),
813 )
814 .map_err(|e| {
815 QueryError::execution(format!(
816 "TEXT vector embed failed: {}",
817 e
818 ))
819 })?;
820 embedded
821 }
822 #[cfg(not(feature = "embedding"))]
823 {
824 let _ = text;
825 return Err(QueryError::execution(
826 "TEXT vector query requires the 'embedding' feature to be enabled",
827 ));
828 }
829 } else {
830 query_vector.clone()
831 };
832 let thr = threshold.unwrap_or(0.5);
833 // Threshold filtering and optional RYW consistency
834 // ride the shared search options; an active rerank
835 // stage is applied by the coordinator.
836 let mut options = linkrs_sync::vector_sync::SearchOptions::new(
837 *space_id,
838 tag_name.clone(),
839 field_name.clone(),
840 resolved_vector,
841 DEFAULT_MATCH_TOP_K,
842 )
843 .with_threshold(thr);
844 options.query_text = query_text.clone();
845 if let Some(cfg) = ryw_config {
846 options.consistency =
847 linkrs_sync::vector_sync::SearchConsistency::ReadYourWrites {
848 timeout_ms: cfg.timeout_ms,
849 };
850 options.minimum_lsn = cfg.minimum_lsn;
851 }
852 let search_results =
853 crate::executor::streaming::helpers::runtime_bridge::wait(
854 "Vector match",
855 coordinator.search_with_options(options),
856 )?;
857 let mut rows = Vec::new();
858 for result in search_results {
859 rows.push(vec![
860 Value::string(result.id.to_string()),
861 Value::Double(result.score as f64),
862 ]);
863 }
864 return if rows.is_empty() {
865 Ok(None)
866 } else {
867 Ok(Some(DataChunk::new_with_layout(
868 rows,
869 self.output_layout.clone(),
870 )))
871 };
872 }
873
874 let _ = input;
875 Err(QueryError::execution(
876 "MATCH VECTOR cannot execute: no vector coordinator is configured",
877 ))
878 }
879
880 #[cfg(not(feature = "vector"))]
881 {
882 let _ = (
883 &space_id,
884 &tag_name,
885 &field_name,
886 &query_vector,
887 query_text.as_deref(),
888 &threshold,
889 input,
890 );
891 Err(QueryError::feature_disabled("vector", "VECTOR MATCH"))
892 }
893 }
894 }
895 }
896
897 pub fn stop(&mut self) -> Result<(), QueryError> {
898 Ok(())
899 }
900
901 pub fn close(&mut self) -> Result<(), QueryError> {
902 Ok(())
903 }
904}