Delete term set

quickwit/Cargo.toml

@@ -192,14 +192,14 @@ quickwit-serve = { version = "0.3.1", path = "./quickwit-serve" }
 quickwit-storage = { version = "0.3.1", path = "./quickwit-storage" }
 quickwit-telemetry = { version = "0.3.1", path = "./quickwit-telemetry" }
 
-fastfield_codecs = { git = "https://github.com/quickwit-oss/tantivy/", rev = "55a9d80" }
-tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "55a9d80", default-features = false, features = [
+fastfield_codecs = { git = "https://github.com/quickwit-oss/tantivy/", rev = "9a090ed" }
+tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "9a090ed", default-features = false, features = [
   "mmap",
   "lz4-compression",
   "zstd-compression",
   "quickwit",
 ] }
-tantivy-query-grammar = { git = "https://github.com/quickwit-oss/tantivy/", rev = "55a9d80" }
+tantivy-query-grammar = { git = "https://github.com/quickwit-oss/tantivy/", rev = "9a090ed" }
 
 # This is actually not used directly the goal is to fix the version
 # used by reqwest. 0.8.30 has an unclear license.

quickwit/quickwit-cli/src/index.rs

@@ -995,7 +995,7 @@ pub async fn search_index(args: SearchIndexArgs) -> anyhow::Result<SearchRespons
         .await?;
     let search_request = SearchRequest {
         index_id: args.index_id,
-        query: args.query.clone(),
+        query: Some(args.query.clone().into()),
         search_fields: args.search_fields.unwrap_or_default(),
         snippet_fields: args.snippet_fields.unwrap_or_default(),
         start_timestamp: args.start_timestamp,

quickwit/quickwit-directories/src/caching_directory.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 use std::{fmt, io};
 
 use async_trait::async_trait;
-use quickwit_storage::MemorySizedCache;
+use quickwit_storage::ByteRangeCache;
 use tantivy::directory::error::OpenReadError;
 use tantivy::directory::{FileHandle, OwnedBytes};
 use tantivy::{AsyncIoResult, Directory, HasLen};
@@ -33,43 +33,18 @@ use tantivy::{AsyncIoResult, Directory, HasLen};
 pub struct CachingDirectory {
     underlying: Arc<dyn Directory>,
     // TODO fixme: that's a pretty ugly cache we have here.
-    cache: Arc<MemorySizedCache>,
+    cache: Arc<ByteRangeCache>,
 }
 
 impl CachingDirectory {
-    /// Creates a new CachingDirectory.
-    /// `capacity_in_bytes` acts as a memory budget for the directory.
-    ///
-    /// The implementation is voluntarily very naive as it was design solely to
-    /// address Quickwit's requirements.
-    /// Most notably, if two reads targetting the same read on the same path
-    /// happen concurrently, the read will be executed twice.
-    ///
-    /// The overall number of bytes held in memory may exceed the capacity at one point
-    /// if a read request is large than `capacity_in_bytes`.
-    /// In that case, the read payload will not be saved in the cache.
-    pub fn new_with_capacity_in_bytes(
-        underlying: Arc<dyn Directory>,
-        capacity_in_bytes: usize,
-    ) -> CachingDirectory {
-        CachingDirectory {
-            underlying,
-            cache: Arc::new(MemorySizedCache::with_capacity_in_bytes(
-                capacity_in_bytes,
-                &quickwit_storage::STORAGE_METRICS.shortlived_cache,
-            )),
-        }
-    }
-
     /// Creates a new CachingDirectory.
     ///
     /// Warming: The resulting CacheDirectory will cache all information without ever
     /// removing any item from the cache.
-    /// Prefer using `.new_with_capacity_in_bytes(...)` in most case.
     pub fn new_unbounded(underlying: Arc<dyn Directory>) -> CachingDirectory {
         CachingDirectory {
             underlying,
-            cache: Arc::new(MemorySizedCache::with_infinite_capacity(
+            cache: Arc::new(ByteRangeCache::with_infinite_capacity(
                 &quickwit_storage::STORAGE_METRICS.shortlived_cache,
             )),
         }
@@ -84,7 +59,7 @@ impl fmt::Debug for CachingDirectory {
 
 struct CachingFileHandle {
     path: PathBuf,
-    cache: Arc<MemorySizedCache>,
+    cache: Arc<ByteRangeCache>,
     underlying_filehandle: Arc<dyn FileHandle>,
 }
 
@@ -179,8 +154,7 @@ mod tests {
         let test_path = Path::new("test");
         ram_directory.atomic_write(test_path, &b"test"[..])?;
         let debug_proxy_directory = Arc::new(DebugProxyDirectory::wrap(ram_directory));
-        let caching_directory =
-            CachingDirectory::new_with_capacity_in_bytes(debug_proxy_directory.clone(), 10_000);
+        let caching_directory = CachingDirectory::new_unbounded(debug_proxy_directory.clone());
         caching_directory.atomic_read(test_path)?;
         caching_directory.atomic_read(test_path)?;
         assert_eq!(debug_proxy_directory.drain_read_operations().count(), 1);

quickwit/quickwit-doc-mapper/src/default_doc_mapper/default_mapper.rs

@@ -1191,7 +1191,7 @@ mod tests {
         query: &str,
     ) -> Result<String, String> {
         let search_request = SearchRequest {
-            query: query.to_string(),
+            query: Some(query.to_string().into()),
             ..Default::default()
         };
         let query = doc_mapper

quickwit/quickwit-doc-mapper/src/doc_mapper.rs

@@ -225,7 +225,7 @@ mod tests {
         let schema = doc_mapper.schema();
         let search_request = SearchRequest {
             index_id: "quickwit-index".to_string(),
-            query: "json_field.toto.titi:hello".to_string(),
+            query: Some("json_field.toto.titi:hello".to_string().into()),
             search_fields: vec![],
             snippet_fields: vec![],
             start_timestamp: None,
@@ -262,7 +262,7 @@ mod tests {
         let schema = doc_mapper.schema();
         let search_request = SearchRequest {
             index_id: "quickwit-index".to_string(),
-            query: "text_field:hello".to_string(),
+            query: Some("text_field:hello".to_string().into()),
             search_fields: vec![],
             snippet_fields: vec![],
             start_timestamp: None,
@@ -297,7 +297,7 @@ mod tests {
         let schema = doc_mapper.schema();
         let search_request = SearchRequest {
             index_id: "quickwit-index".to_string(),
-            query: "toto.titi:hello".to_string(),
+            query: Some("toto.titi:hello".to_string().into()),
             search_fields: vec![],
             snippet_fields: vec![],
             start_timestamp: None,
@@ -332,7 +332,7 @@ mod tests {
         let schema = doc_mapper.schema();
         let search_request = SearchRequest {
             index_id: "quickwit-index".to_string(),
-            query: "toto:5".to_string(),
+            query: Some("toto:5".to_string().into()),
             search_fields: vec![],
             snippet_fields: vec![],
             start_timestamp: None,

quickwit/quickwit-doc-mapper/src/query_builder.rs

@@ -19,9 +19,10 @@
 
 use std::collections::HashSet;
 
+use quickwit_proto::search_request::Query as SearchQuery;
 use quickwit_proto::SearchRequest;
 use tantivy::query::{Query, QueryParser, QueryParserError as TantivyQueryParserError};
-use tantivy::schema::{Field, FieldType, Schema};
+use tantivy::schema::{Field, FieldType, Schema, Term, Type, Value};
 use tantivy_query_grammar::{UserInputAst, UserInputLeaf, UserInputLiteral};
 
 use crate::sort_by::validate_sort_by_field_name;
@@ -33,38 +34,103 @@ pub(crate) fn build_query(
     request: &SearchRequest,
     default_field_names: &[String],
 ) -> Result<Box<dyn Query>, QueryParserError> {
-    let user_input_ast = tantivy_query_grammar::parse_query(&request.query)
-        .map_err(|_| TantivyQueryParserError::SyntaxError(request.query.to_string()))?;
+    let query = match &request.query {
+        Some(SearchQuery::Text(query)) => {
+            let user_input_ast = tantivy_query_grammar::parse_query(query)
+                .map_err(|_| TantivyQueryParserError::SyntaxError(query.clone()))?;
 
-    if has_range_clause(&user_input_ast) {
-        return Err(anyhow::anyhow!("Range queries are not currently allowed.").into());
-    }
+            if has_range_clause(&user_input_ast) {
+                return Err(anyhow::anyhow!("Range queries are not currently allowed.").into());
+            }
 
-    if needs_default_search_field(&user_input_ast)
-        && request.search_fields.is_empty()
-        && (default_field_names.is_empty() || default_field_names == [DYNAMIC_FIELD_NAME])
-    {
-        return Err(
-            anyhow::anyhow!("No default field declared and no field specified in query.").into(),
-        );
-    }
+            if needs_default_search_field(&user_input_ast)
+                && request.search_fields.is_empty()
+                && (default_field_names.is_empty() || default_field_names == [DYNAMIC_FIELD_NAME])
+            {
+                return Err(anyhow::anyhow!(
+                    "No default field declared and no field specified in query."
+                )
+                .into());
+            }
 
-    validate_requested_snippet_fields(&schema, request, &user_input_ast, default_field_names)?;
+            validate_requested_snippet_fields(
+                &schema,
+                request,
+                &user_input_ast,
+                default_field_names,
+            )?;
+
+            let search_fields = if request.search_fields.is_empty() {
+                resolve_fields(&schema, default_field_names)?
+            } else {
+                resolve_fields(&schema, &request.search_fields)?
+            };
+
+            if let Some(sort_by_field) = request.sort_by_field.as_ref() {
+                validate_sort_by_field_name(sort_by_field, &schema, Some(&search_fields))?;
+            }
 
-    let search_fields = if request.search_fields.is_empty() {
-        resolve_fields(&schema, default_field_names)?
-    } else {
-        resolve_fields(&schema, &request.search_fields)?
-    };
+            let mut query_parser =
+                QueryParser::new(schema, search_fields, QUICKWIT_TOKENIZER_MANAGER.clone());
+            query_parser.set_conjunction_by_default();
+            query_parser.parse_query(query)?
+        }
+        Some(SearchQuery::SetQuery(set_query)) => {
+            let _ = (&set_query.terms, &set_query.field_name, &set_query.tags);
+            let field_name = &set_query.field_name;
+            let field = schema
+                .get_field(field_name)
+                .ok_or_else(|| TantivyQueryParserError::FieldDoesNotExist(field_name.clone()))?;
+
+            let field_type = schema.get_field_entry(field).field_type();
+            if !field_type.is_indexed() {
+                return Err(anyhow::anyhow!("Attempted to search on not indexed field.").into());
+            }
 
-    if let Some(sort_by_field) = request.sort_by_field.as_ref() {
-        validate_sort_by_field_name(sort_by_field, &schema, Some(&search_fields))?;
-    }
+            // TODO maybe Facet could be allowed?
+            if matches!(field_type.value_type(), Type::Json | Type::Facet) {
+                return Err(
+                    anyhow::anyhow!("Attempted to search on unsuported field type.").into(),
+                );
+            }
+
+            // TODO error on any parse error instead??
+            let terms: Vec<_> = set_query
+                .terms
+                .iter()
+                // TODO not correct
+                .flat_map(|term| field_type.value_from_json(term.clone().into()).ok())
+                .map(|term| match term {
+                    Value::Str(text) => Term::from_field_text(field, &text),
+                    Value::PreTokStr(pre_tokenized_string) => {
+                        Term::from_field_text(field, &pre_tokenized_string.text)
+                    }
+                    Value::U64(int) => Term::from_field_u64(field, int),
+                    Value::I64(int) => Term::from_field_i64(field, int),
+                    Value::F64(int) => Term::from_field_f64(field, int),
+                    Value::Bool(b) => Term::from_field_bool(field, b),
+                    Value::Date(date) => Term::from_field_date(field, date),
+                    Value::Bytes(buf) => Term::from_field_bytes(field, &buf),
+                    Value::IpAddr(ip) => Term::from_field_ip_addr(field, ip),
+                    Value::JsonObject(_) | Value::Facet(_) => unreachable!(),
+                })
+                .collect();
+
+            if terms.is_empty() {
+                return Err(anyhow::anyhow!("No valid term to search for").into());
+            }
+
+            // TODO BooleanQuery(Must(Set(terms)), Must(Set(tag))) would probably be more coherent
+            Box::new(tantivy::query::TermSetQuery::new(terms))
+        }
+        None => {
+            return Err(anyhow::anyhow!(
+                "Invalid request state: neither a text query nor a set query"
+            )
+            .into())
+        }
+    };
 
-    let mut query_parser =
-        QueryParser::new(schema, search_fields, QUICKWIT_TOKENIZER_MANAGER.clone());
-    query_parser.set_conjunction_by_default();
-    let query = query_parser.parse_query(&request.query)?;
     Ok(query)
 }
 
@@ -205,7 +271,7 @@ mod test {
         let request = SearchRequest {
             aggregation_request: None,
             index_id: "test_index".to_string(),
-            query: query_str.to_string(),
+            query: Some(query_str.to_string().into()),
             search_fields,
             snippet_fields: vec![],
             start_timestamp: None,
@@ -358,7 +424,7 @@ mod test {
         let request = SearchRequest {
             aggregation_request: None,
             index_id: "test_index".to_string(),
-            query: query_str.to_string(),
+            query: Some(query_str.to_string().into()),
             search_fields,
             snippet_fields,
             start_timestamp: None,
@@ -368,8 +434,8 @@ mod test {
             sort_order: None,
             sort_by_field: None,
         };
-        let user_input_ast = tantivy_query_grammar::parse_query(&request.query)
-            .map_err(|_| QueryParserError::SyntaxError(request.query.clone()))
+        let user_input_ast = tantivy_query_grammar::parse_query(query_str)
+            .map_err(|_| QueryParserError::SyntaxError(query_str.to_owned()))
             .unwrap();
         let default_field_names =
             default_search_fields.unwrap_or_else(|| vec!["title".to_string(), "desc".to_string()]);

quickwit/quickwit-indexing/src/actors/merge_executor.rs

@@ -459,7 +459,7 @@ impl MergeExecutor {
                     .expect("A delete task must have a delete query.");
                 let search_request = SearchRequest {
                     index_id: delete_query.index_id,
-                    query: delete_query.query,
+                    query: delete_query.query.map(Into::into),
                     start_timestamp: delete_query.start_timestamp,
                     end_timestamp: delete_query.end_timestamp,
                     search_fields: delete_query.search_fields,
@@ -675,7 +675,7 @@ mod tests {
                 index_id: index_id.to_string(),
                 start_timestamp: None,
                 end_timestamp: None,
-                query: "body:delete".to_string(),
+                query: Some("body:delete".to_string().into()),
                 search_fields: Vec::new(),
             })
             .await?;