add _elastic/_field_caps API

quickwit/quickwit-serve/src/elastic_search_api/filter.rs

@@ -29,7 +29,7 @@ use super::model::{
 use crate::elastic_search_api::model::{
     ElasticBulkOptions, ScrollQueryParams, SearchBody, SearchQueryParams,
 };
-use crate::search_api::extract_index_id_patterns;
+use crate::search_api::{extract_index_id_patterns, extract_index_id_patterns_default};
 
 const BODY_LENGTH_LIMIT: ByteSize = ByteSize::mib(1);
 const CONTENT_LENGTH_LIMIT: ByteSize = ByteSize::mib(10);
@@ -79,6 +79,31 @@ pub(crate) fn elastic_bulk_filter(
         .and(serde_qs::warp::query(serde_qs::Config::default()))
 }
 
+#[utoipa::path(
+    post,
+    tag = "Ingest",
+    path = "/{index}/_bulk",
+    request_body(content = String, description = "Elasticsearch compatible bulk request body limited to 10MB", content_type = "application/json"),
+    responses(
+        (status = 200, description = "Successfully ingested documents.", body = IngestResponse)
+    ),
+    params(
+        ("refresh" = Option<ElasticRefresh>, Query, description = "Force or wait for commit at the end of the indexing operation."),
+    )
+)]
+pub(crate) fn elastic_index_bulk_filter(
+) -> impl Filter<Extract = (String, Bytes, ElasticBulkOptions), Error = Rejection> + Clone {
+    warp::path!("_elastic" / String / "_bulk")
+        .and(warp::post())
+        .and(warp::body::content_length_limit(
+            CONTENT_LENGTH_LIMIT.as_u64(),
+        ))
+        .and(warp::body::bytes())
+        .and(serde_qs::warp::query::<ElasticBulkOptions>(
+            serde_qs::Config::default(),
+        ))
+}
+
 /// Like the warp json filter, but accepts an empty body and interprets it as `T::default`.
 fn json_or_empty<T: DeserializeOwned + Send + Default>(
 ) -> impl Filter<Extract = (T,), Error = Rejection> + Copy {
@@ -102,7 +127,7 @@ fn json_or_empty<T: DeserializeOwned + Send + Default>(
         .unify()
 }
 
-#[utoipa::path(get, tag = "metadata", path = "/{index}/_field_caps")]
+#[utoipa::path(get, tag = "Metadata", path = "/{index}/_field_caps")]
 pub(crate) fn elastic_index_field_capabilities_filter() -> impl Filter<
     Extract = (
         Vec<String>,
@@ -118,6 +143,22 @@ pub(crate) fn elastic_index_field_capabilities_filter() -> impl Filter<
         .and(json_or_empty())
 }
 
+#[utoipa::path(get, tag = "Metadata", path = "/_field_caps")]
+pub(crate) fn elastic_field_capabilities_filter() -> impl Filter<
+    Extract = (
+        Vec<String>,
+        FieldCapabilityQueryParams,
+        FieldCapabilityRequestBody,
+    ),
+    Error = Rejection,
+> + Clone {
+    warp::path!("_elastic" / "_field_caps")
+        .and_then(extract_index_id_patterns_default)
+        .and(warp::get().or(warp::post()).unify())
+        .and(serde_qs::warp::query(serde_qs::Config::default()))
+        .and(json_or_empty())
+}
+
 #[utoipa::path(get, tag = "Search", path = "/{index}/_search")]
 pub(crate) fn elastic_index_search_filter(
 ) -> impl Filter<Extract = (Vec<String>, SearchQueryParams, SearchBody), Error = Rejection> + Clone
@@ -129,31 +170,6 @@ pub(crate) fn elastic_index_search_filter(
         .and(json_or_empty())
 }
 
-#[utoipa::path(
-    post,
-    tag = "Ingest",
-    path = "/{index}/_bulk",
-    request_body(content = String, description = "Elasticsearch compatible bulk request body limited to 10MB", content_type = "application/json"),
-    responses(
-        (status = 200, description = "Successfully ingested documents.", body = IngestResponse)
-    ),
-    params(
-        ("refresh" = Option<ElasticRefresh>, Query, description = "Force or wait for commit at the end of the indexing operation."),
-    )
-)]
-pub(crate) fn elastic_index_bulk_filter(
-) -> impl Filter<Extract = (String, Bytes, ElasticBulkOptions), Error = Rejection> + Clone {
-    warp::path!("_elastic" / String / "_bulk")
-        .and(warp::post())
-        .and(warp::body::content_length_limit(
-            CONTENT_LENGTH_LIMIT.as_u64(),
-        ))
-        .and(warp::body::bytes())
-        .and(serde_qs::warp::query::<ElasticBulkOptions>(
-            serde_qs::Config::default(),
-        ))
-}
-
 #[utoipa::path(post, tag = "Search", path = "/_msearch")]
 pub(crate) fn elastic_multi_search_filter(
 ) -> impl Filter<Extract = (Bytes, MultiSearchQueryParams), Error = Rejection> + Clone {

quickwit/quickwit-serve/src/elastic_search_api/mod.rs

@@ -56,12 +56,12 @@ pub fn elastic_api_handlers(
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     es_compat_cluster_info_handler(node_config, BuildInfo::get())
         .or(es_compat_search_handler(search_service.clone()))
+        .or(es_compat_index_search_handler(search_service.clone()))
+        .or(es_compat_scroll_handler(search_service.clone()))
+        .or(es_compat_index_multi_search_handler(search_service.clone()))
         .or(es_compat_index_field_capabilities_handler(
             search_service.clone(),
         ))
-        .or(es_compat_index_search_handler(search_service.clone()))
-        .or(es_compat_scroll_handler(search_service.clone()))
-        .or(es_compat_index_multi_search_handler(search_service))
         .or(es_compat_bulk_handler(
             ingest_service.clone(),
             ingest_router.clone(),

quickwit/quickwit-serve/src/elastic_search_api/model/field_capability.rs

@@ -98,6 +98,7 @@ struct FieldCapabilityEntryResponse {
     // Option since it is filled later
     #[serde(rename = "type")]
     typ: Option<FieldCapabilityEntryType>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
     indices: Vec<String>, // [ "index1", "index2" ],
     #[serde(skip_serializing_if = "Vec::is_empty")]
     non_aggregatable_indices: Vec<String>, // [ "index1" ]
@@ -164,6 +165,13 @@ pub fn convert_to_es_field_capabilities_response(
         for field_type in types {
             let mut add_entry = add_entry.clone();
             add_entry.typ = Some(field_type.clone());
+
+            // If the field exists in all indices, we omit field.indices in the response.
+            let exists_in_all_indices = add_entry.indices.len() == indices.len();
+            if exists_in_all_indices {
+                add_entry.indices = Vec::new();
+            }
+
             entry.insert(field_type, add_entry);
         }
     }

quickwit/quickwit-serve/src/elastic_search_api/rest_handler.rs

@@ -42,9 +42,9 @@ use serde_json::json;
 use warp::{Filter, Rejection};
 
 use super::filter::{
-    elastic_cluster_info_filter, elastic_index_field_capabilities_filter,
-    elastic_index_search_filter, elastic_multi_search_filter, elastic_scroll_filter,
-    elastic_search_filter,
+    elastic_cluster_info_filter, elastic_field_capabilities_filter,
+    elastic_index_field_capabilities_filter, elastic_index_search_filter,
+    elastic_multi_search_filter, elastic_scroll_filter, elastic_search_filter,
 };
 use super::model::{
     build_list_field_request_for_es_api, convert_to_es_field_capabilities_response,
@@ -98,11 +98,12 @@ pub fn es_compat_search_handler(
 }
 
 /// GET or POST _elastic/{index}/_field_caps
-/// TODO: add route handling for _elastic/_field_caps
 pub fn es_compat_index_field_capabilities_handler(
     search_service: Arc<dyn SearchService>,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     elastic_index_field_capabilities_filter()
+        .or(elastic_field_capabilities_filter())
+        .unify()
         .and(with_arg(search_service))
         .then(es_compat_index_field_capabilities)
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
@@ -118,16 +119,6 @@ pub fn es_compat_index_search_handler(
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
 }
 
-/// GET or POST _elastic/_search/scroll
-pub fn es_compat_scroll_handler(
-    search_service: Arc<dyn SearchService>,
-) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    elastic_scroll_filter()
-        .and(with_arg(search_service))
-        .then(es_scroll)
-        .map(|result| make_elastic_api_response(result, BodyFormat::default()))
-}
-
 /// POST _elastic/_search
 pub fn es_compat_index_multi_search_handler(
     search_service: Arc<dyn SearchService>,
@@ -144,6 +135,16 @@ pub fn es_compat_index_multi_search_handler(
         })
 }
 
+/// GET or POST _elastic/_search/scroll
+pub fn es_compat_scroll_handler(
+    search_service: Arc<dyn SearchService>,
+) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
+    elastic_scroll_filter()
+        .and(with_arg(search_service))
+        .then(es_scroll)
+        .map(|result| make_elastic_api_response(result, BodyFormat::default()))
+}
+
 fn build_request_for_es_api(
     index_id_patterns: Vec<String>,
     search_params: SearchQueryParams,

quickwit/quickwit-serve/src/search_api/mod.rs

@@ -21,7 +21,7 @@ mod grpc_adapter;
 mod rest_handler;
 
 pub use self::grpc_adapter::GrpcSearchAdapter;
-pub(crate) use self::rest_handler::extract_index_id_patterns;
+pub(crate) use self::rest_handler::{extract_index_id_patterns, extract_index_id_patterns_default};
 pub use self::rest_handler::{
     search_get_handler, search_post_handler, search_request_from_api_request,
     search_stream_handler, SearchApi, SearchRequestQueryString, SortBy,

quickwit/quickwit-serve/src/search_api/rest_handler.rs

@@ -55,6 +55,11 @@ use crate::{with_arg, BodyFormat};
 )]
 pub struct SearchApi;
 
+pub(crate) async fn extract_index_id_patterns_default() -> Result<Vec<String>, Rejection> {
+    let index_id_patterns = Vec::new();
+    Ok(index_id_patterns)
+}
+
 pub(crate) async fn extract_index_id_patterns(
     comma_separated_index_id_patterns: String,
 ) -> Result<Vec<String>, Rejection> {