linkrs_query/executor/streaming/operators/
fulltext_operator.rs

1use std::sync::Arc;
2
3use parking_lot::RwLock;
4
5use crate::executor::streaming::chunk::DataChunk;
6use crate::executor::streaming::executor::StreamingExecutor;
7#[cfg(feature = "fulltext")]
8use crate::executor::streaming::operators::ddl_operator::make_single_row;
9use crate::executor::streaming::operators::source_operator::OperatorConfig;
10use crate::executor::streaming::operators::spec::FulltextManageCommand;
11use crate::executor::streaming::runtime::ExecutionRuntime;
12use crate::executor::streaming::slot::SlotLayout;
13use crate::storage::QueryStorage;
14use linkrs_core::error::QueryError;
15use linkrs_core::types::expr::Expression;
16#[cfg(feature = "fulltext")]
17use linkrs_core::Value;
18#[cfg(feature = "fulltext")]
19use linkrs_fulltext::manager::FulltextIndexManager;
20
21#[cfg(feature = "fulltext")]
22use crate::executor::streaming::chunk::{ColumnInfo, Schema};
23
24#[cfg(not(feature = "fulltext"))]
25fn fulltext_command_info(cmd: &FulltextManageCommand) -> (&'static str, Option<&str>) {
26    use crate::executor::streaming::operators::spec::FulltextManageCommand::*;
27    match cmd {
28        Create { index_name, .. } => ("create_fulltext_index", Some(index_name.as_str())),
29        Drop { index_name, .. } => ("drop_fulltext_index", Some(index_name.as_str())),
30        Alter { index_name, .. } => ("alter_fulltext_index", Some(index_name.as_str())),
31        Show { pattern, .. } => ("show_fulltext_index", pattern.as_deref()),
32        Describe { index_name, .. } => ("describe_fulltext_index", Some(index_name.as_str())),
33    }
34}
35
36#[cfg(feature = "fulltext")]
37fn make_manage_result(
38    output_layout: Arc<SlotLayout>,
39    action: &str,
40    name: Option<&str>,
41    status: &str,
42) -> DataChunk {
43    let name_val = name
44        .map(Value::string)
45        .unwrap_or(Value::Null(linkrs_core::NullType::Null));
46    DataChunk::new_with_layout(
47        vec![vec![Value::string(action), name_val, Value::string(status)]],
48        output_layout,
49    )
50}
51
52/// Map the planner-level engine type to the fulltext manager's engine type.
53///
54/// Both enums currently carry only BM25; the match stays exhaustive so a
55/// future engine variant fails to compile here instead of silently falling
56/// back to a default.
57#[cfg(feature = "fulltext")]
58fn map_engine_type(engine: linkrs_core::types::FulltextEngineType) -> linkrs_fulltext::EngineType {
59    match engine {
60        linkrs_core::types::FulltextEngineType::Bm25 => linkrs_fulltext::EngineType::Bm25,
61    }
62}
63
64/// Render parsed index options as the opaque engine-config document stored
65/// in fulltext index metadata.
66///
67/// Returns `None` when no option was given so the manager keeps its compiled
68/// defaults; per-field `analyzer` / `boost` are parser-rejected today, so a
69/// present field entry only ever carries its name.
70#[cfg(feature = "fulltext")]
71fn fulltext_engine_config(
72    options: &crate::parser::ast::fulltext::IndexOptions,
73) -> Option<serde_json::Value> {
74    let has_bm25 = options.bm25_config.is_some();
75    if !has_bm25 && options.common_options.is_empty() {
76        return None;
77    }
78    let mut config = serde_json::Map::new();
79    if let Some(bm25) = &options.bm25_config {
80        if let Some(k1) = bm25.k1 {
81            config.insert("k1".to_string(), serde_json::json!(k1));
82        }
83        if let Some(b) = bm25.b {
84            config.insert("b".to_string(), serde_json::json!(b));
85        }
86        if let Some(analyzer) = &bm25.analyzer {
87            config.insert("analyzer".to_string(), serde_json::json!(analyzer));
88        }
89        if let Some(store_original) = bm25.store_original {
90            config.insert(
91                "store_original".to_string(),
92                serde_json::json!(store_original),
93            );
94        }
95        if !bm25.field_weights.is_empty() {
96            config.insert(
97                "field_weights".to_string(),
98                serde_json::json!(bm25.field_weights),
99            );
100        }
101    }
102    for (key, value) in &options.common_options {
103        config.insert(key.clone(), plain_option_json(value));
104    }
105    Some(serde_json::Value::Object(config))
106}
107
108/// Render an option value as plain JSON.
109///
110/// `Value` serializes externally tagged (`{"String": ...}`); option metadata
111/// stays human-readable by unwrapping the common scalar cases and falling
112/// back to the tagged form for complex values.
113#[cfg(feature = "fulltext")]
114fn plain_option_json(value: &linkrs_core::Value) -> serde_json::Value {
115    use linkrs_core::Value;
116    match value {
117        Value::String(s) => serde_json::Value::String(s.to_string()),
118        Value::FixedString(s) => serde_json::Value::String(s.clone()),
119        Value::SmallInt(v) => serde_json::json!(*v),
120        Value::Int(v) => serde_json::json!(*v),
121        Value::BigInt(v) => serde_json::json!(*v),
122        Value::Float(v) => serde_json::json!(*v),
123        Value::Double(v) => serde_json::json!(*v),
124        Value::Bool(v) => serde_json::Value::Bool(*v),
125        Value::Null(_) => serde_json::Value::Null,
126        _ => serde_json::to_value(value).unwrap_or(serde_json::Value::Null),
127    }
128}
129
130#[derive(Debug)]
131pub enum FulltextOperatorKind {
132    FulltextManage {
133        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
134        space_name: String,
135        command: FulltextManageCommand,
136        #[cfg(feature = "fulltext")]
137        fulltext_manager: Option<Arc<FulltextIndexManager>>,
138    },
139    FulltextSearch {
140        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
141        space_name: String,
142        space_id: u64,
143        index_name: String,
144        structured_query: linkrs_fulltext::query::FulltextQuery,
145        tag_name: String,
146        field_name: String,
147        limit: Option<usize>,
148        #[cfg(feature = "fulltext")]
149        fulltext_manager: Option<Arc<FulltextIndexManager>>,
150    },
151    FulltextLookup {
152        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
153        space_name: String,
154        space_id: u64,
155        index_name: String,
156        structured_query: linkrs_fulltext::query::FulltextQuery,
157        tag_name: String,
158        field_name: String,
159        limit: Option<usize>,
160        #[cfg(feature = "fulltext")]
161        fulltext_manager: Option<Arc<FulltextIndexManager>>,
162    },
163    MatchFulltext {
164        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
165        space_name: String,
166        space_id: u64,
167        match_expr: Expression,
168        match_field: Option<String>,
169        /// Structured query built from the fulltext match condition.
170        structured_query: linkrs_fulltext::query::FulltextQuery,
171        tag_name: String,
172        field_name: String,
173        limit: Option<usize>,
174        #[cfg(feature = "fulltext")]
175        fulltext_manager: Option<Arc<FulltextIndexManager>>,
176    },
177}
178
179/// Fulltext operator.
180///
181/// Wraps [`FulltextOperatorKind`] with the runtime context injected at
182/// `open()`. Lifecycle state is owned exclusively by the executor; operators
183/// never write it.
184#[derive(Debug)]
185pub struct FulltextOperator {
186    pub kind: FulltextOperatorKind,
187    pub runtime: Option<Arc<ExecutionRuntime>>,
188    pub output_layout: Arc<SlotLayout>,
189    pub config: OperatorConfig,
190}
191
192impl FulltextOperator {
193    /// Create a FulltextOperator from an immutable spec.
194    pub fn from_spec(
195        spec: &super::spec::FulltextSpec,
196        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
197        #[cfg(feature = "fulltext")] fulltext_manager: Option<Arc<FulltextIndexManager>>,
198        output_layout: Arc<SlotLayout>,
199    ) -> Self {
200        let kind = match spec {
201            super::spec::FulltextSpec::FulltextManage {
202                space_name,
203                command,
204            } => FulltextOperatorKind::FulltextManage {
205                storage: storage.clone(),
206                space_name: space_name.clone(),
207                command: command.clone(),
208                #[cfg(feature = "fulltext")]
209                fulltext_manager: fulltext_manager.clone(),
210            },
211            super::spec::FulltextSpec::FulltextSearch {
212                space_name,
213                space_id,
214                index_name,
215                structured_query,
216                tag_name,
217                field_name,
218                limit,
219            } => FulltextOperatorKind::FulltextSearch {
220                storage: storage.clone(),
221                space_name: space_name.clone(),
222                space_id: *space_id,
223                index_name: index_name.clone(),
224                structured_query: structured_query.clone(),
225                tag_name: tag_name.clone(),
226                field_name: field_name.clone(),
227                limit: *limit,
228                #[cfg(feature = "fulltext")]
229                fulltext_manager: fulltext_manager.clone(),
230            },
231            super::spec::FulltextSpec::FulltextLookup {
232                space_name,
233                space_id,
234                index_name,
235                structured_query,
236                tag_name,
237                field_name,
238                limit,
239            } => FulltextOperatorKind::FulltextLookup {
240                storage: storage.clone(),
241                space_name: space_name.clone(),
242                space_id: *space_id,
243                index_name: index_name.clone(),
244                structured_query: structured_query.clone(),
245                tag_name: tag_name.clone(),
246                field_name: field_name.clone(),
247                limit: *limit,
248                #[cfg(feature = "fulltext")]
249                fulltext_manager: fulltext_manager.clone(),
250            },
251            super::spec::FulltextSpec::MatchFulltext {
252                space_name,
253                space_id,
254                match_expr,
255                match_field,
256                structured_query,
257                tag_name,
258                field_name,
259                limit,
260            } => FulltextOperatorKind::MatchFulltext {
261                storage: storage.clone(),
262                space_name: space_name.clone(),
263                space_id: *space_id,
264                match_expr: match_expr.clone(),
265                match_field: match_field.clone(),
266                structured_query: structured_query.clone(),
267                tag_name: tag_name.clone(),
268                field_name: field_name.clone(),
269                limit: *limit,
270                #[cfg(feature = "fulltext")]
271                fulltext_manager: fulltext_manager.clone(),
272            },
273        };
274        Self::new(kind, output_layout)
275    }
276
277    pub fn new(kind: FulltextOperatorKind, output_layout: Arc<SlotLayout>) -> Self {
278        Self {
279            kind,
280            runtime: None,
281            output_layout,
282            config: OperatorConfig::default(),
283        }
284    }
285
286    /// Inject the runtime and execution config (called once by the executor
287    /// before this operator produces any data).
288    pub fn inject_context(
289        &mut self,
290        runtime: Option<&Arc<ExecutionRuntime>>,
291        config: OperatorConfig,
292    ) {
293        if let Some(rt) = runtime {
294            self.runtime = Some(rt.clone());
295        }
296        self.config = config;
297    }
298
299    pub fn open(&mut self, input: &mut StreamingExecutor) -> Result<(), QueryError> {
300        match &mut self.kind {
301            FulltextOperatorKind::FulltextManage { .. }
302            | FulltextOperatorKind::FulltextSearch { .. }
303            | FulltextOperatorKind::FulltextLookup { .. }
304            | FulltextOperatorKind::MatchFulltext { .. } => {
305                input.open()?;
306                Ok(())
307            }
308        }
309    }
310
311    pub fn next(&mut self, input: &mut StreamingExecutor) -> Result<Option<DataChunk>, QueryError> {
312        match &mut self.kind {
313            FulltextOperatorKind::FulltextManage {
314                storage: _storage,
315                space_name: _space_name,
316                command,
317                #[cfg(feature = "fulltext")]
318                fulltext_manager,
319            } => {
320                #[cfg(feature = "fulltext")]
321                {
322                    use crate::executor::streaming::operators::spec::FulltextManageCommand::*;
323                    let result = match command {
324                        Create {
325                            index_name,
326                            schema_name,
327                            fields,
328                            engine_type,
329                            options,
330                            if_not_exists,
331                            space_id,
332                        } => {
333                            if let Some(manager) = fulltext_manager {
334                                let engine_config = fulltext_engine_config(options);
335                                let engine = map_engine_type(*engine_type);
336                                for field in fields {
337                                    if manager.has_index(
338                                        *space_id,
339                                        schema_name,
340                                        field.field_name.as_str(),
341                                    ) {
342                                        if *if_not_exists {
343                                            continue;
344                                        }
345                                        return Err(QueryError::execution(format!(
346                                            "Fulltext index on {}.{}({}) already exists",
347                                            space_id, schema_name, field.field_name,
348                                        )));
349                                    }
350                                    crate::executor::streaming::helpers::runtime_bridge::wait(
351                                        "Fulltext create",
352                                        manager.create_index_with_engine_config(
353                                            *space_id,
354                                            schema_name,
355                                            field.field_name.as_str(),
356                                            index_name.as_str(),
357                                            Some(engine),
358                                            engine_config.clone(),
359                                        ),
360                                    )?;
361                                }
362                                Some(make_manage_result(
363                                    Arc::clone(&self.output_layout),
364                                    "create_fulltext_index",
365                                    Some(index_name.as_str()),
366                                    "created",
367                                ))
368                            } else {
369                                return Err(QueryError::execution(
370                                    "CREATE FULLTEXT INDEX cannot execute: no fulltext manager is configured",
371                                ));
372                            }
373                        }
374                        Drop {
375                            index_name,
376                            if_exists,
377                        } => {
378                            if let Some(manager) = fulltext_manager {
379                                let matching: Vec<_> = manager
380                                    .list_indexes()
381                                    .into_iter()
382                                    .filter(|metadata| metadata.index_name == *index_name)
383                                    .collect();
384                                if matching.is_empty() && !*if_exists {
385                                    return Err(QueryError::execution(format!(
386                                        "Fulltext index not found: {}",
387                                        index_name
388                                    )));
389                                }
390                                for metadata in matching {
391                                    crate::executor::streaming::helpers::runtime_bridge::wait(
392                                        "Fulltext drop",
393                                        manager.drop_index(
394                                            metadata.space_id,
395                                            &metadata.tag_name,
396                                            &metadata.field_name,
397                                        ),
398                                    )?;
399                                }
400                                Some(make_manage_result(
401                                    Arc::clone(&self.output_layout),
402                                    "drop_fulltext_index",
403                                    Some(index_name.as_str()),
404                                    "dropped",
405                                ))
406                            } else {
407                                return Err(QueryError::execution(
408                                    "DROP FULLTEXT INDEX cannot execute: no fulltext manager is configured",
409                                ));
410                            }
411                        }
412                        Describe { index_name } => {
413                            if let Some(manager) = fulltext_manager {
414                                if let Some(meta) = manager
415                                    .list_indexes()
416                                    .into_iter()
417                                    .find(|metadata| metadata.index_name == *index_name)
418                                {
419                                    let schema = Arc::new(Schema::new(vec![
420                                        ColumnInfo {
421                                            name: "index_id".to_string(),
422                                            data_type: "string".to_string(),
423                                        },
424                                        ColumnInfo {
425                                            name: "tag_name".to_string(),
426                                            data_type: "string".to_string(),
427                                        },
428                                        ColumnInfo {
429                                            name: "field_name".to_string(),
430                                            data_type: "string".to_string(),
431                                        },
432                                        ColumnInfo {
433                                            name: "status".to_string(),
434                                            data_type: "string".to_string(),
435                                        },
436                                    ]));
437                                    Some(make_single_row(
438                                        schema,
439                                        vec![
440                                            Value::string(meta.index_name),
441                                            Value::string(meta.tag_name),
442                                            Value::string(meta.field_name),
443                                            Value::string(format!("{:?}", meta.status)),
444                                        ],
445                                    ))
446                                } else {
447                                    Some(make_manage_result(
448                                        Arc::clone(&self.output_layout),
449                                        "describe_fulltext_index",
450                                        Some(index_name.as_str()),
451                                        "not-found",
452                                    ))
453                                }
454                            } else {
455                                return Err(QueryError::execution(
456                                    "DESCRIBE FULLTEXT INDEX cannot execute: no fulltext manager is configured",
457                                ));
458                            }
459                        }
460                        Show {
461                            pattern,
462                            from_schema,
463                        } => {
464                            if let Some(manager) = fulltext_manager {
465                                let indexes: Vec<_> = manager
466                                    .list_indexes()
467                                    .into_iter()
468                                    .filter(|metadata| {
469                                        pattern.as_ref().is_none_or(|pattern| {
470                                            metadata.index_name.contains(pattern)
471                                        }) && from_schema
472                                            .as_ref()
473                                            .is_none_or(|schema| &metadata.tag_name == schema)
474                                    })
475                                    .collect();
476                                let _schema = Arc::new(Schema::new(vec![
477                                    ColumnInfo {
478                                        name: "index_id".to_string(),
479                                        data_type: "string".to_string(),
480                                    },
481                                    ColumnInfo {
482                                        name: "space_id".to_string(),
483                                        data_type: "bigint".to_string(),
484                                    },
485                                    ColumnInfo {
486                                        name: "tag_name".to_string(),
487                                        data_type: "string".to_string(),
488                                    },
489                                    ColumnInfo {
490                                        name: "field_name".to_string(),
491                                        data_type: "string".to_string(),
492                                    },
493                                    ColumnInfo {
494                                        name: "status".to_string(),
495                                        data_type: "string".to_string(),
496                                    },
497                                ]));
498                                let rows: Vec<Vec<Value>> = indexes
499                                    .into_iter()
500                                    .map(|m| {
501                                        vec![
502                                            Value::string(m.index_name),
503                                            Value::BigInt(m.space_id as i64),
504                                            Value::string(m.tag_name),
505                                            Value::string(m.field_name),
506                                            Value::string(format!("{:?}", m.status)),
507                                        ]
508                                    })
509                                    .collect();
510                                Some(DataChunk::new_with_layout(
511                                    rows,
512                                    Arc::clone(&self.output_layout),
513                                ))
514                            } else {
515                                return Err(QueryError::execution(
516                                    "SHOW FULLTEXT INDEXES cannot execute: no fulltext manager is configured",
517                                ));
518                            }
519                        }
520                        Alter {
521                            index_name,
522                            actions,
523                        } => {
524                            use crate::parser::ast::fulltext::AlterIndexAction;
525                            if let Some(manager) = fulltext_manager {
526                                let matching: Vec<_> = manager
527                                    .list_indexes()
528                                    .into_iter()
529                                    .filter(|metadata| metadata.index_name == *index_name)
530                                    .collect();
531                                if matching.is_empty() {
532                                    return Err(QueryError::execution(format!(
533                                        "Fulltext index not found: {}",
534                                        index_name
535                                    )));
536                                }
537                                for metadata in matching {
538                                    for action in actions.iter() {
539                                        match action {
540                                            AlterIndexAction::Rebuild => {
541                                                return Err(QueryError::execution(
542                                                    crate::planning::fulltext_planner::FulltextSearchPlanner::FULLTEXT_REBUILD_UNAVAILABLE,
543                                                ));
544                                            }
545                                            AlterIndexAction::Optimize => {
546                                                crate::executor::streaming::helpers::runtime_bridge::wait(
547                                                    "Fulltext optimize",
548                                                    manager.optimize_index(
549                                                        metadata.space_id,
550                                                        &metadata.tag_name,
551                                                        &metadata.field_name,
552                                                    ),
553                                                )?;
554                                            }
555                                            AlterIndexAction::AddField(_)
556                                            | AlterIndexAction::DropField(_)
557                                            | AlterIndexAction::SetOption(_, _) => {
558                                                return Err(QueryError::execution(
559                                                    "ALTER FULLTEXT INDEX only supports REBUILD and OPTIMIZE; use CREATE/DROP for field or option changes",
560                                                ));
561                                            }
562                                        }
563                                    }
564                                }
565                                Some(make_manage_result(
566                                    Arc::clone(&self.output_layout),
567                                    "alter_fulltext_index",
568                                    Some(index_name.as_str()),
569                                    "altered",
570                                ))
571                            } else {
572                                return Err(QueryError::execution(
573                                    "ALTER FULLTEXT INDEX cannot execute: no fulltext manager is configured",
574                                ));
575                            }
576                        }
577                    };
578                    Ok(result)
579                }
580
581                #[cfg(not(feature = "fulltext"))]
582                {
583                    let (operation, _) = fulltext_command_info(command);
584                    Err(QueryError::feature_disabled(
585                        "fulltext",
586                        &operation.to_uppercase(),
587                    ))
588                }
589            }
590
591            FulltextOperatorKind::FulltextSearch {
592                structured_query,
593                space_id,
594                tag_name,
595                field_name,
596                limit,
597                #[cfg(feature = "fulltext")]
598                fulltext_manager,
599                ..
600            } => {
601                #[cfg(feature = "fulltext")]
602                {
603                    if let Some(manager) = fulltext_manager {
604                        let search_limit = limit.unwrap_or(100);
605                        let search_results =
606                            crate::executor::streaming::helpers::runtime_bridge::wait(
607                                "Fulltext search",
608                                manager.search_structured(
609                                    *space_id,
610                                    tag_name,
611                                    field_name,
612                                    structured_query,
613                                    search_limit,
614                                ),
615                            )?;
616                        let mut rows = Vec::new();
617                        for result in search_results {
618                            rows.push(vec![result.doc_id, Value::Double(result.score as f64)]);
619                        }
620                        if !rows.is_empty() {
621                            return Ok(Some(DataChunk::new_with_layout(
622                                rows,
623                                self.output_layout.clone(),
624                            )));
625                        }
626                        return Ok(None);
627                    }
628
629                    let _ = input;
630                    Err(QueryError::execution(
631                        "FULLTEXT SEARCH cannot execute: no fulltext manager is configured",
632                    ))
633                }
634
635                #[cfg(not(feature = "fulltext"))]
636                {
637                    let _ = (
638                        &structured_query,
639                        &space_id,
640                        &tag_name,
641                        &field_name,
642                        limit,
643                        input,
644                    );
645                    Err(QueryError::feature_disabled("fulltext", "FULLTEXT SEARCH"))
646                }
647            }
648
649            FulltextOperatorKind::FulltextLookup {
650                structured_query,
651                space_id,
652                tag_name,
653                field_name,
654                limit,
655                #[cfg(feature = "fulltext")]
656                fulltext_manager,
657                ..
658            } => {
659                #[cfg(feature = "fulltext")]
660                {
661                    if let Some(manager) = fulltext_manager {
662                        let search_limit = limit.unwrap_or(100);
663                        let search_results =
664                            crate::executor::streaming::helpers::runtime_bridge::wait(
665                                "Fulltext lookup",
666                                manager.search_structured(
667                                    *space_id,
668                                    tag_name,
669                                    field_name,
670                                    structured_query,
671                                    search_limit,
672                                ),
673                            )?;
674                        let mut rows = Vec::new();
675                        for result in search_results {
676                            rows.push(vec![result.doc_id, Value::Double(result.score as f64)]);
677                        }
678                        return if rows.is_empty() {
679                            Ok(None)
680                        } else {
681                            Ok(Some(DataChunk::new_with_layout(
682                                rows,
683                                self.output_layout.clone(),
684                            )))
685                        };
686                    }
687
688                    let _ = input;
689                    Err(QueryError::execution(
690                        "FULLTEXT LOOKUP cannot execute: no fulltext manager is configured",
691                    ))
692                }
693
694                #[cfg(not(feature = "fulltext"))]
695                {
696                    let _ = (
697                        &structured_query,
698                        &space_id,
699                        &tag_name,
700                        &field_name,
701                        limit,
702                        input,
703                    );
704                    Err(QueryError::feature_disabled("fulltext", "FULLTEXT LOOKUP"))
705                }
706            }
707
708            FulltextOperatorKind::MatchFulltext {
709                match_expr: _match_expr,
710                space_id,
711                structured_query,
712                tag_name,
713                field_name,
714                limit,
715                #[cfg(feature = "fulltext")]
716                fulltext_manager,
717                ..
718            } => {
719                #[cfg(feature = "fulltext")]
720                {
721                    if let Some(manager) = fulltext_manager {
722                        let search_limit = limit.unwrap_or(100);
723                        let search_results =
724                            crate::executor::streaming::helpers::runtime_bridge::wait(
725                                "Fulltext match",
726                                manager.search_structured(
727                                    *space_id,
728                                    tag_name,
729                                    field_name,
730                                    structured_query,
731                                    search_limit,
732                                ),
733                            )?;
734                        let mut rows = Vec::new();
735                        for result in search_results {
736                            rows.push(vec![result.doc_id, Value::Double(result.score as f64)]);
737                        }
738                        return if rows.is_empty() {
739                            Ok(None)
740                        } else {
741                            Ok(Some(DataChunk::new_with_layout(
742                                rows,
743                                self.output_layout.clone(),
744                            )))
745                        };
746                    }
747
748                    let _ = input;
749                    Err(QueryError::execution(
750                        "FULLTEXT MATCH cannot execute: no fulltext manager is configured",
751                    ))
752                }
753
754                #[cfg(not(feature = "fulltext"))]
755                {
756                    let _ = (
757                        &_match_expr,
758                        &space_id,
759                        &structured_query,
760                        &tag_name,
761                        &field_name,
762                        limit,
763                        input,
764                    );
765                    Err(QueryError::feature_disabled("fulltext", "FULLTEXT MATCH"))
766                }
767            }
768        }
769    }
770
771    pub fn stop(&mut self) -> Result<(), QueryError> {
772        Ok(())
773    }
774
775    pub fn close(&mut self) -> Result<(), QueryError> {
776        Ok(())
777    }
778}
779
780#[cfg(all(test, feature = "fulltext"))]
781mod tests {
782    use super::*;
783    use crate::parser::ast::fulltext::{BM25Options, IndexOptions};
784    use std::collections::HashMap;
785
786    #[test]
787    fn engine_config_is_none_without_options() {
788        let options = IndexOptions {
789            bm25_config: None,
790            common_options: HashMap::new(),
791        };
792        assert_eq!(fulltext_engine_config(&options), None);
793    }
794
795    #[test]
796    fn engine_config_carries_bm25_and_common_options() {
797        let mut common = HashMap::new();
798        common.insert("custom".to_string(), linkrs_core::Value::string("x"));
799        let options = IndexOptions {
800            bm25_config: Some(BM25Options {
801                k1: Some(1.2),
802                b: Some(0.8),
803                field_weights: HashMap::new(),
804                analyzer: None,
805                store_original: Some(true),
806            }),
807            common_options: common,
808        };
809        let config = fulltext_engine_config(&options).expect("config rendered");
810        assert_eq!(config["k1"], serde_json::json!(1.2f32));
811        assert_eq!(config["b"], serde_json::json!(0.8f32));
812        assert_eq!(config["store_original"], serde_json::json!(true));
813        assert_eq!(config["custom"], serde_json::json!("x"));
814    }
815
816    #[test]
817    fn engine_type_maps_bm25() {
818        assert_eq!(
819            map_engine_type(linkrs_core::types::FulltextEngineType::Bm25),
820            linkrs_fulltext::EngineType::Bm25
821        );
822    }
823}
linkrs_query/executor/streaming/operators/fulltext_operator.rs

linkrs_query/executor/streaming/operators/
fulltext_operator.rs
