feat(elasticsearch_api): pushdown start/end timestamp on _mapping endpoint

quickwit/Cargo.toml

@@ -235,6 +235,7 @@ serde = { version = "1.0.228", features = ["derive", "rc"] }
 serde_json = "1.0"
 serde_json_borrow = "0.9"
 serde_qs = { version = "0.15" }
+serde_urlencoded = "0.7"
 serde_with = "3.16"
 serde_yaml = "0.9"
 serial_test = { version = "3.2", features = ["file_locks"] }

quickwit/quickwit-serve/Cargo.toml

@@ -89,6 +89,7 @@ itertools = { workspace = true }
 mockall = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry_sdk = { workspace = true }
+serde_urlencoded = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }

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

@@ -20,7 +20,7 @@ use warp::{Filter, Rejection};
 
 use super::model::{
     CatIndexQueryParams, DeleteQueryParams, FieldCapabilityQueryParams, FieldCapabilityRequestBody,
-    MultiSearchQueryParams, SearchQueryParamsCount,
+    IndexMappingQueryParams, MultiSearchQueryParams, SearchQueryParamsCount,
 };
 use crate::Body;
 use crate::decompression::get_body_bytes;
@@ -285,9 +285,10 @@ pub(crate) fn elastic_aliases_filter() -> impl Filter<Extract = (), Error = Reje
 }
 
 pub(crate) fn elastic_index_mapping_filter()
--> impl Filter<Extract = (String,), Error = Rejection> + Clone {
+-> impl Filter<Extract = (String, IndexMappingQueryParams), Error = Rejection> + Clone {
     warp::path!("_elastic" / String / "_mapping")
         .or(warp::path!("_elastic" / String / "_mappings"))
         .unify()
         .and(warp::get())
+        .and(warp::query())
 }

quickwit/quickwit-serve/src/elasticsearch_api/model/index_mapping_query_params.rs

@@ -0,0 +1,77 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use serde::{Deserialize, Serialize};
+
+/// Query parameters accepted by `GET /_elastic/{index}/_mapping` and
+/// `/_mappings`.
+///
+/// Both fields are optional and absent by default. When present, they are
+/// forwarded into [`quickwit_proto::search::ListFieldsRequest`] verbatim,
+/// where they prune the set of splits considered for field discovery. Unit
+/// is **epoch seconds**, interval is half-open `[start_timestamp,
+/// end_timestamp)` — matching the `ListFieldsRequest` proto contract
+/// exactly.
+#[serde_with::skip_serializing_none]
+#[derive(Default, Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct IndexMappingQueryParams {
+    #[serde(default)]
+    pub start_timestamp: Option<i64>,
+    #[serde(default)]
+    pub end_timestamp: Option<i64>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::IndexMappingQueryParams;
+
+    #[test]
+    fn empty_query_string_yields_none() {
+        let params: IndexMappingQueryParams = serde_urlencoded::from_str("").unwrap();
+        assert!(params.start_timestamp.is_none());
+        assert!(params.end_timestamp.is_none());
+    }
+
+    #[test]
+    fn both_params_present_yield_some() {
+        let qs = "start_timestamp=1712160204&end_timestamp=1712764984";
+        let params: IndexMappingQueryParams = serde_urlencoded::from_str(qs).unwrap();
+        assert_eq!(params.start_timestamp, Some(1712160204));
+        assert_eq!(params.end_timestamp, Some(1712764984));
+    }
+
+    #[test]
+    fn only_start_timestamp_present() {
+        let qs = "start_timestamp=1712160204";
+        let params: IndexMappingQueryParams = serde_urlencoded::from_str(qs).unwrap();
+        assert_eq!(params.start_timestamp, Some(1712160204));
+        assert!(params.end_timestamp.is_none());
+    }
+
+    #[test]
+    fn only_end_timestamp_present() {
+        let qs = "end_timestamp=1712764984";
+        let params: IndexMappingQueryParams = serde_urlencoded::from_str(qs).unwrap();
+        assert!(params.start_timestamp.is_none());
+        assert_eq!(params.end_timestamp, Some(1712764984));
+    }
+
+    #[test]
+    fn unknown_field_is_rejected() {
+        let qs = "start_timestamp=1&unexpected=foo";
+        let result: Result<IndexMappingQueryParams, _> = serde_urlencoded::from_str(qs);
+        assert!(result.is_err());
+    }
+}

quickwit/quickwit-serve/src/elasticsearch_api/model/mod.rs

@@ -17,6 +17,7 @@ mod bulk_query_params;
 mod cat_indices;
 mod error;
 mod field_capability;
+mod index_mapping_query_params;
 mod mappings;
 mod multi_search;
 mod scroll;
@@ -36,6 +37,7 @@ pub use field_capability::{
     FieldCapabilityQueryParams, FieldCapabilityRequestBody, FieldCapabilityResponse,
     build_list_field_request_for_es_api, convert_to_es_field_capabilities_response,
 };
+pub use index_mapping_query_params::IndexMappingQueryParams;
 pub(crate) use mappings::ElasticsearchMappingsResponse;
 pub use multi_search::{
     MultiSearchHeader, MultiSearchQueryParams, MultiSearchResponse, MultiSearchSingleResponse,

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

@@ -58,9 +58,9 @@ use super::model::{
     CatIndexQueryParams, DeleteQueryParams, ElasticsearchCatIndexResponse, ElasticsearchError,
     ElasticsearchResolveIndexEntryResponse, ElasticsearchResolveIndexResponse,
     ElasticsearchResponse, ElasticsearchStatsResponse, FieldCapabilityQueryParams,
-    FieldCapabilityRequestBody, FieldCapabilityResponse, MultiSearchHeader, MultiSearchQueryParams,
-    MultiSearchResponse, MultiSearchSingleResponse, ScrollQueryParams, SearchBody,
-    SearchQueryParams, SearchQueryParamsCount, StatsResponseEntry,
+    FieldCapabilityRequestBody, FieldCapabilityResponse, IndexMappingQueryParams,
+    MultiSearchHeader, MultiSearchQueryParams, MultiSearchResponse, MultiSearchSingleResponse,
+    ScrollQueryParams, SearchBody, SearchQueryParams, SearchQueryParamsCount, StatsResponseEntry,
     build_list_field_request_for_es_api, convert_to_es_field_capabilities_response,
 };
 use super::{TrackTotalHits, make_elastic_api_response};
@@ -201,6 +201,7 @@ async fn get_index_metadata(
 
 pub(crate) async fn es_compat_index_mapping(
     index_id: String,
+    params: IndexMappingQueryParams,
     mut metastore: MetastoreServiceClient,
     search_service: Arc<dyn SearchService>,
 ) -> Result<ElasticsearchMappingsResponse, ElasticsearchError> {
@@ -217,8 +218,8 @@ pub(crate) async fn es_compat_index_mapping(
     let list_fields_request = quickwit_proto::search::ListFieldsRequest {
         index_id_patterns,
         fields: Vec::new(),
-        start_timestamp: None,
-        end_timestamp: None,
+        start_timestamp: params.start_timestamp,
+        end_timestamp: params.end_timestamp,
         query_ast: None,
     };
     let list_fields_response = search_service