diff --git a/CHANGELOG.md b/CHANGELOG.md index 5468dc7a4b3..636a7276b0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,11 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Azure Blob Storage: support custom endpoints via `endpoint` and `endpoint_suffix` configuration options for sovereign clouds (#6624) +- Search API: cursor paging with the `search_after` parameter and the per-hit `cursors` of the response ### Fixed - (Jaeger) Query resource attributes when Jaeger request carries tags +- Return a 400 for `search_after` sort values that do not match the sort fields (was a 500) ### Changed +- (Elasticsearch API) `search_after` on a datetime sort field with a sub-millisecond `fast_precision` requires the `epoch_nanos_int` format ### Deprecated diff --git a/docs/reference/es_compatible_api.md b/docs/reference/es_compatible_api.md index b1a0ba3702b..50a9f33f69a 100644 --- a/docs/reference/es_compatible_api.md +++ b/docs/reference/es_compatible_api.md @@ -237,6 +237,8 @@ You can pass the `sort` value of the last hit in a subsequent request where othe This allows you to paginate your results. +If a datetime sort field has a `fast_precision` finer than milliseconds, `search_after` requires the `epoch_nanos_int` format for that field. Quickwit rejects these requests without it: with millisecond sort values, Quickwit would skip the hits within the same millisecond as the last hit. + ### `_msearch`   Multi search API ``` diff --git a/docs/reference/rest-api.md b/docs/reference/rest-api.md index 54ac05c040b..2953492e8dd 100644 --- a/docs/reference/rest-api.md +++ b/docs/reference/rest-api.md @@ -64,6 +64,7 @@ POST api/v1//search | `start_timestamp` | `i64` | If set, restrict search to documents with a `timestamp >= start_timestamp`, taking advantage of potential time pruning opportunities. The value must be in seconds. | | | `end_timestamp` | `i64` | If set, restrict search to documents with a `timestamp < end_timestamp`, taking advantage of potential time pruning opportunities. The value must be in seconds. | | | `start_offset` | `Integer` | Number of documents to skip | `0` | +| `search_after` | `String` | Cursor from the `cursors` of a previous response. The search returns the hits that come after that hit in the `sort_by` order, i.e. the next page. See [paging with cursors](#paging-with-cursors). | | | `max_hits` | `Integer` | Maximum number of hits to return (by default 20) | `20` | | `search_field` | `[String]` | Fields to search on if no field name is specified in the query. Comma-separated list, e.g. "field1,field2" | index_config.search_settings.default_search_fields | | `snippet_fields` | `[String]` | Fields to extract snippet on. Comma-separated list, e.g. "field1,field2" | | @@ -82,9 +83,18 @@ The response is a JSON object, and the content type is `application/json; charse | Field | Description | Type | | -------------------- | ------------------------------ | :--------: | | `hits` | Results of the query | `[hit]` | +| `cursors` | Opaque `search_after` cursor of each hit, in the same order as `hits` | `[string]` | | `num_hits` | Total number of matches | `number` | | `elapsed_time_micros` | Processing time of the query | `number` | +#### Paging with cursors + +To page past the 10,000 `start_offset` limit, pass the last entry of `cursors` as `search_after` in the next request, with the same `query` and `sort_by`. To page backward, pass a hit's cursor and invert every `sort_by` order: `ts` and `-ts` swap, and `-_doc` inverts the default order. + +- Quickwit breaks ties on equal sort values by hit address and changes addresses when it merges splits. A merge between two requests can skip or repeat the tied hits, or any hit without `sort_by`. For a snapshot, use the Elasticsearch-compatible [scroll API](es_compatible_api.md#_searchscroll--scroll-api). +- Quickwit sorts hits without a value for a sort field last in both directions and rejects their cursors. +- Quickwit rejects paging on a datetime field whose `fast_precision` is finer than milliseconds, because cursors carry milliseconds. + ### Search multiple indices Search APIs that accept `index id` requests path parameter also support multi-target syntax. diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index 7553ed3d7ce..0dbf33a1051 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -565,6 +565,7 @@ pub async fn local_search_cli(args: LocalSearchArgs) -> anyhow::Result<()> { sort_by, count_all: CountHits::CountAll, allow_failed_splits: false, + search_after: None, }; let search_request = search_request_from_api_request(vec![args.index_id], search_request_query_string)?; diff --git a/quickwit/quickwit-rest-client/src/models.rs b/quickwit/quickwit-rest-client/src/models.rs index 2857495803f..cf7594e7311 100644 --- a/quickwit/quickwit-rest-client/src/models.rs +++ b/quickwit/quickwit-rest-client/src/models.rs @@ -81,6 +81,8 @@ impl ApiResponse { pub struct SearchResponseRestClient { pub num_hits: u64, pub hits: Vec, + #[serde(default)] + pub cursors: Vec, pub snippets: Option>, pub elapsed_time_micros: u64, pub errors: Vec, diff --git a/quickwit/quickwit-rest-client/src/rest_client.rs b/quickwit/quickwit-rest-client/src/rest_client.rs index ea5eda5a391..676fd4e6630 100644 --- a/quickwit/quickwit-rest-client/src/rest_client.rs +++ b/quickwit/quickwit-rest-client/src/rest_client.rs @@ -852,6 +852,7 @@ mod test { let expected_search_response = SearchResponseRestClient { num_hits: 0, hits: Vec::new(), + cursors: Vec::new(), snippets: None, aggregations: None, elapsed_time_micros: 100, @@ -874,6 +875,44 @@ mod test { ); } + #[tokio::test] + async fn test_search_endpoint_with_cursors() { + let mock_server = MockServer::start().await; + let server_url = Url::parse(&mock_server.uri()).unwrap(); + let qw_client = QuickwitClientBuilder::new(server_url).build(); + let search_query_params = SearchRequestQueryString { + ..Default::default() + }; + let expected_search_response = SearchResponseRestClient { + num_hits: 1, + hits: vec![json!({"body": "foo"})], + cursors: vec!["EgFhGAEgAlICEAU".to_string()], + snippets: None, + aggregations: None, + elapsed_time_micros: 100, + errors: Vec::new(), + }; + Mock::given(method("POST")) + .and(path("/api/v1/my-index/search")) + .respond_with(ResponseTemplate::new(StatusCode::OK).set_body_json(json!({ + "num_hits": 1, + "hits": [{"body": "foo"}], + "cursors": ["EgFhGAEgAlICEAU"], + "elapsed_time_micros": 100, + "errors": [] + }))) + .up_to_n_times(1) + .mount(&mock_server) + .await; + assert_eq!( + qw_client + .search("my-index", search_query_params) + .await + .unwrap(), + expected_search_response + ); + } + fn get_ndjson_filepath(ndjson_dataset_filename: &str) -> String { format!( "{}/resources/tests/{}", diff --git a/quickwit/quickwit-search/src/lib.rs b/quickwit/quickwit-search/src/lib.rs index e70f0360ec4..25e7b065452 100644 --- a/quickwit/quickwit-search/src/lib.rs +++ b/quickwit/quickwit-search/src/lib.rs @@ -91,7 +91,7 @@ pub use crate::root::{ }; pub use crate::search_job_placer::{Job, SearchJobPlacer}; pub use crate::search_response_rest::{ - AggregationResults, SearchPlanResponseRest, SearchResponseRest, + AggregationResults, SearchAfterCursor, SearchPlanResponseRest, SearchResponseRest, }; pub use crate::service::{MockSearchService, SearchService, SearchServiceImpl}; diff --git a/quickwit/quickwit-search/src/root.rs b/quickwit/quickwit-search/src/root.rs index 94da26a6b09..f92a3a862b1 100644 --- a/quickwit/quickwit-search/src/root.rs +++ b/quickwit/quickwit-search/src/root.rs @@ -46,7 +46,7 @@ use tantivy::TantivyError; use tantivy::aggregation::agg_result::AggregationResults; use tantivy::aggregation::intermediate_agg_result::IntermediateAggregationResults; use tantivy::collector::Collector; -use tantivy::schema::{Field, FieldEntry, FieldType, Schema}; +use tantivy::schema::{DateTimePrecision, Field, FieldEntry, FieldType, Schema}; use tracing::{Span, debug, error, info, info_span, instrument}; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -248,6 +248,7 @@ fn validate_request_and_build_metadata( validate_sort_field_types( &schema, &search_request.sort_fields, + search_request.search_after.is_some(), &mut sort_fields_is_datetime, )?; @@ -289,6 +290,7 @@ fn validate_request_and_build_metadata( fn validate_sort_field_types( schema: &Schema, sort_fields: &[SortField], + has_search_after: bool, sort_field_is_datetime: &mut HashMap, ) -> crate::Result<()> { for sort_field in sort_fields.iter() { @@ -297,6 +299,9 @@ fn validate_sort_field_types( sort_field_entry, sort_field.sort_datetime_format.is_some(), )?; + if has_search_after { + validate_search_after_datetime_precision(sort_field, sort_field_entry)?; + } // If sort field type is a date, ensure it's true for all indexes. if let Some(is_datetime) = sort_field_is_datetime.get(&sort_field.field_name) { if *is_datetime != sort_field_entry.field_type().is_date() { @@ -318,6 +323,36 @@ fn validate_sort_field_types( Ok(()) } +/// Rejects `search_after` on a datetime sort field with a sub-millisecond precision, unless its +/// sort values are in nanoseconds. +/// +/// Datetime sort values are returned in milliseconds by default and converted back into +/// nanoseconds when parsing `search_after`. For such fields, that round trip truncates the cursor +/// and silently skips the documents falling within the same millisecond. +fn validate_search_after_datetime_precision( + sort_field: &SortField, + sort_field_entry: &FieldEntry, +) -> crate::Result<()> { + let FieldType::Date(date_options) = sort_field_entry.field_type() else { + return Ok(()); + }; + let is_millisecond_or_coarser = matches!( + date_options.get_precision(), + DateTimePrecision::Seconds | DateTimePrecision::Milliseconds + ); + let has_nanosecond_sort_values = + sort_field.sort_datetime_format() == SortDatetimeFormat::UnixTimestampNanos; + if is_millisecond_or_coarser || has_nanosecond_sort_values { + return Ok(()); + } + Err(SearchError::InvalidArgument(format!( + "search_after is not supported on the datetime sort field `{}` because its precision is \ + finer than milliseconds, unless sort values are in nanoseconds (Elasticsearch API: \ + `\"format\": \"epoch_nanos_int\"`)", + sort_field.field_name + ))) +} + fn validate_requested_snippet_fields( schema: &Schema, snippet_fields: &[String], @@ -382,14 +417,13 @@ fn simplify_search_request_for_scroll_api(req: &SearchRequest) -> crate::Result< /// Validates sort fields and search after values. /// - validate sort fields length. -/// - search after values must be set for all sort fields. +/// - search after values are positional: the leaf search compares the n-th one with the n-th sort +/// field. They must be set for all sort fields except `_doc`, which has no sort value, and only +/// for them. Without sort fields, hits are sorted by `_doc`. fn validate_sort_by_fields_and_search_after( sort_fields: &[SortField], search_after: &Option, ) -> crate::Result<()> { - if sort_fields.is_empty() { - return Ok(()); - } if sort_fields.len() > 2 { return Err(SearchError::InvalidArgument(format!( "sort by field must be up to 2 fields, got {}", @@ -400,11 +434,15 @@ fn validate_sort_by_fields_and_search_after( return Ok(()); }; - let sort_fields_without_doc_count = sort_fields + let sort_field_names: Vec<&str> = sort_fields .iter() - .filter(|sort_field| !SORT_DOC_FIELD_NAMES.contains(&sort_field.field_name.as_str())) - .count(); - let has_doc_sort_field = sort_fields_without_doc_count != sort_fields.len(); + .map(|sort_field| sort_field.field_name.as_str()) + .collect(); + let is_doc_field = |field_name: &str| SORT_DOC_FIELD_NAMES.contains(&field_name); + let has_doc_sort_field = sort_field_names.is_empty() + || sort_field_names + .iter() + .any(|field_name| is_doc_field(field_name)); if has_doc_sort_field && search_after_partial_hit.split_id.is_empty() { return Err(SearchError::InvalidArgument( "search_after with a sort field `_doc` must define a split ID, segment ID and doc ID \ @@ -413,27 +451,42 @@ fn validate_sort_by_fields_and_search_after( )); } - let mut search_after_sort_value_count = 0; - // TODO: we could validate if the search after sort value types of consistent with the sort - // field types. - if let Some(sort_by_value) = search_after_partial_hit.sort_value.as_ref() { - sort_by_value.sort_value.context("sort value must be set")?; - search_after_sort_value_count += 1; - } - if let Some(sort_by_value_2) = search_after_partial_hit.sort_value2.as_ref() { - sort_by_value_2 - .sort_value - .context("sort value must be set")?; - search_after_sort_value_count += 1; - } - if search_after_sort_value_count != sort_fields_without_doc_count { - return Err(SearchError::InvalidArgument(format!( - "`search_after` must have the same number of sort values as sort by fields {:?}", - sort_fields - .iter() - .map(|sort_field| &sort_field.field_name) - .collect_vec() - ))); + let number_of_sort_values_error = || { + SearchError::InvalidArgument(format!( + "`search_after` must have the same number of sort values as sort by fields \ + {sort_field_names:?}" + )) + }; + // Like the leaf search, a `SortByValue` without a value counts as no sort value. + let search_after_sort_values = [ + &search_after_partial_hit.sort_value, + &search_after_partial_hit.sort_value2, + ] + .map(|sort_by_value_opt| { + sort_by_value_opt + .as_ref() + .and_then(|sort_by_value| sort_by_value.sort_value.as_ref()) + }); + for (position, sort_value_opt) in search_after_sort_values.into_iter().enumerate() { + match (sort_field_names.get(position).copied(), sort_value_opt) { + (Some(field_name), Some(_)) if is_doc_field(field_name) => { + return Err(SearchError::InvalidArgument(format!( + "`search_after` cannot have a sort value for the sort field `{field_name}`" + ))); + } + // The leaf search panics on scores that are not floats, while it converts the other + // fast field values leniently. + (Some("_score"), Some(sort_value)) if !matches!(sort_value, SortValue::F64(_)) => { + return Err(SearchError::InvalidArgument(format!( + "`search_after` sort value for `_score` must be a float, got `{sort_value:?}`" + ))); + } + (Some(field_name), None) if !is_doc_field(field_name) => { + return Err(number_of_sort_values_error()); + } + (None, Some(_)) => return Err(number_of_sort_values_error()), + _ => {} + } } Ok(()) } @@ -1534,8 +1587,9 @@ fn convert_search_after_datetime_values( Ok(()) } -/// Convert sort values from input datetime format into nanoseconds. +/// Convert `search_after` sort values from input datetime format into nanoseconds. /// The conversion is done only for U64 and I64 sort values, an error is returned for other types. +/// Errors are invalid arguments since the values come from the request. fn convert_sort_datetime_value_into_nanos( sort_value: &mut SortValue, input_format: SortDatetimeFormat, @@ -1544,7 +1598,7 @@ fn convert_sort_datetime_value_into_nanos( SortValue::U64(value) => match input_format { SortDatetimeFormat::UnixTimestampMillis => { *value = value.checked_mul(1_000_000).ok_or_else(|| { - SearchError::Internal(format!( + SearchError::InvalidArgument(format!( "sort value defined in milliseconds is too large and cannot be converted \ into nanoseconds: {value}" )) @@ -1557,7 +1611,7 @@ fn convert_sort_datetime_value_into_nanos( SortValue::I64(value) => match input_format { SortDatetimeFormat::UnixTimestampMillis => { *value = value.checked_mul(1_000_000).ok_or_else(|| { - SearchError::Internal(format!( + SearchError::InvalidArgument(format!( "sort value defined in milliseconds is too large and cannot be converted \ into nanoseconds: {value}" )) @@ -1568,7 +1622,7 @@ fn convert_sort_datetime_value_into_nanos( } }, _ => { - return Err(SearchError::Internal(format!( + return Err(SearchError::InvalidArgument(format!( "datetime conversion are only support for u64 and i64 sort values, not \ `{sort_value:?}`" ))); @@ -1923,7 +1977,7 @@ mod tests { ScrollRequest, SortByValue, SortOrder, SortValue, SplitSearchError, }; use quickwit_query::query_ast::{qast_helper, qast_json_helper, query_ast_from_user_text}; - use tantivy::schema::{FAST, STORED, TEXT}; + use tantivy::schema::{DateOptions, FAST, STORED, TEXT}; use super::*; use crate::{MockSearchService, searcher_pool_for_test}; @@ -2276,8 +2330,8 @@ mod tests { .unwrap_err(); assert_eq!( error.to_string(), - "internal error: `sort value defined in milliseconds is too large and cannot be \ - converted into nanoseconds: 1617000000000000`" + "Invalid argument: sort value defined in milliseconds is too large and cannot be \ + converted into nanoseconds: 1617000000000000" ); // conversion with float values should fail. let mut sort_value = SortValue::F64(1617000000000000.0); @@ -2288,8 +2342,8 @@ mod tests { .unwrap_err(); assert_eq!( error.to_string(), - "internal error: `datetime conversion are only support for u64 and i64 sort values, \ - not `F64(1617000000000000.0)``" + "Invalid argument: datetime conversion are only support for u64 and i64 sort values, \ + not `F64(1617000000000000.0)`" ); } @@ -2312,7 +2366,8 @@ mod tests { schema_builder.add_u64_field("id", FAST); let schema = schema_builder.build(); let mut sort_field_are_datetime = HashMap::new(); - validate_sort_field_types(&schema, &sort_fields, &mut sort_field_are_datetime).unwrap(); + validate_sort_field_types(&schema, &sort_fields, false, &mut sort_field_are_datetime) + .unwrap(); assert_eq!(sort_field_are_datetime.get("_doc"), Some(&false)); assert_eq!(sort_field_are_datetime.get("_shard_doc"), Some(&false)); } @@ -2336,11 +2391,73 @@ mod tests { schema_builder.add_u64_field("id", FAST); let schema = schema_builder.build(); let mut sort_field_are_datetime = HashMap::new(); - validate_sort_field_types(&schema, &sort_fields, &mut sort_field_are_datetime).unwrap(); + validate_sort_field_types(&schema, &sort_fields, false, &mut sort_field_are_datetime) + .unwrap(); assert_eq!(sort_field_are_datetime.get("timestamp"), Some(&true)); assert_eq!(sort_field_are_datetime.get("id"), Some(&false)); } + fn sort_field(field_name: &str, sort_datetime_format: Option) -> SortField { + SortField { + field_name: field_name.to_string(), + sort_order: 0, + sort_datetime_format: sort_datetime_format.map(|format| format as i32), + } + } + + #[test] + fn test_validate_sort_field_types_search_after_datetime_precision() { + let mut schema_builder = Schema::builder(); + for (field_name, precision) in [ + ("ts_seconds", DateTimePrecision::Seconds), + ("ts_millis", DateTimePrecision::Milliseconds), + ("ts_micros", DateTimePrecision::Microseconds), + ] { + schema_builder.add_date_field( + field_name, + DateOptions::default().set_fast().set_precision(precision), + ); + } + schema_builder.add_u64_field("id", FAST); + let schema = schema_builder.build(); + + for (sort_field, has_search_after) in [ + (sort_field("ts_seconds", None), true), + (sort_field("ts_millis", None), true), + (sort_field("id", None), true), + (sort_field("_shard_doc", None), true), + (sort_field("ts_micros", None), false), + ( + sort_field("ts_micros", Some(SortDatetimeFormat::UnixTimestampNanos)), + true, + ), + ] { + validate_sort_field_types( + &schema, + std::slice::from_ref(&sort_field), + has_search_after, + &mut HashMap::new(), + ) + .unwrap_or_else(|error| panic!("{sort_field:?}: {error}")); + } + for sort_field in [ + sort_field("ts_micros", None), + sort_field("ts_micros", Some(SortDatetimeFormat::UnixTimestampMillis)), + ] { + let error = validate_sort_field_types( + &schema, + std::slice::from_ref(&sort_field), + true, + &mut HashMap::new(), + ) + .unwrap_err(); + assert!( + matches!(error, SearchError::InvalidArgument(_)), + "{sort_field:?}: {error:?}" + ); + } + } + #[test] fn test_validate_sort_field_types_with_inconsistent_datetime_type() { let sort_fields = vec![ @@ -2363,9 +2480,13 @@ mod tests { let mut sort_field_are_datetime = HashMap::new(); sort_field_are_datetime.insert("timestamp".to_string(), false); sort_field_are_datetime.insert("id".to_string(), false); - let error = - validate_sort_field_types(&schema, &sort_fields, &mut sort_field_are_datetime) - .unwrap_err(); + let error = validate_sort_field_types( + &schema, + &sort_fields, + false, + &mut sort_field_are_datetime, + ) + .unwrap_err(); assert_eq!( error.to_string(), "sort datetime field `timestamp` must be of type datetime on all indexes" @@ -2374,9 +2495,13 @@ mod tests { { let mut sort_field_are_datetime = HashMap::new(); sort_field_are_datetime.insert("id".to_string(), true); - let error = - validate_sort_field_types(&schema, &sort_fields, &mut sort_field_are_datetime) - .unwrap_err(); + let error = validate_sort_field_types( + &schema, + &sort_fields, + false, + &mut sort_field_are_datetime, + ) + .unwrap_err(); assert_eq!( error.to_string(), "sort datetime field `id` must be of type datetime on all indexes" @@ -2611,6 +2736,71 @@ mod tests { ); } + #[test] + fn test_validate_sort_by_fields_and_search_after_checks_each_sort_value_position() { + let search_after = |sort_value: Option, sort_value2: Option| { + Some(PartialHit { + sort_value: sort_value.map(SortByValue::from), + sort_value2: sort_value2.map(SortByValue::from), + split_id: "split".to_string(), + segment_ord: 0, + doc_id: 0, + }) + }; + let valid_cases = [ + (vec![], search_after(None, None)), + ( + vec![sort_field("_score", None)], + search_after(Some(SortValue::F64(1.5)), None), + ), + // `_doc` has no sort value, so the value of `ts` is the second one. + ( + vec![sort_field("_doc", None), sort_field("ts", None)], + search_after(None, Some(SortValue::I64(5))), + ), + ]; + for (sort_fields, search_after) in valid_cases { + validate_sort_by_fields_and_search_after(&sort_fields, &search_after) + .unwrap_or_else(|error| panic!("{sort_fields:?}: {error}")); + } + let invalid_cases = [ + // Without sort fields, hits are sorted by address, which a split ID is needed for. + (vec![], Some(PartialHit::default())), + // A cursor of `sort_by=ts` reused without `sort_by`. + (vec![], search_after(Some(SortValue::I64(5)), None)), + // A second sort value without a second sort field. + ( + vec![sort_field("ts", None)], + search_after(None, Some(SortValue::I64(5))), + ), + ( + vec![sort_field("_doc", None), sort_field("ts", None)], + search_after(Some(SortValue::U64(5)), None), + ), + // A cursor of `sort_by=ts` reused with `sort_by=_score`. + ( + vec![sort_field("_score", None)], + search_after(Some(SortValue::I64(5)), None), + ), + ( + vec![sort_field("ts", None)], + Some(PartialHit { + sort_value: Some(SortByValue { sort_value: None }), + split_id: "split".to_string(), + ..Default::default() + }), + ), + ]; + for (sort_fields, search_after) in invalid_cases { + let error = + validate_sort_by_fields_and_search_after(&sort_fields, &search_after).unwrap_err(); + assert!( + matches!(error, SearchError::InvalidArgument(_)), + "{sort_fields:?}: {error:?}" + ); + } + } + #[test] fn test_validate_sort_by_fields_and_search_after_invalid_3() { // 3 sort fields is not possible. diff --git a/quickwit/quickwit-search/src/search_response_rest.rs b/quickwit/quickwit-search/src/search_response_rest.rs index 58eddc7b927..586b56e9b57 100644 --- a/quickwit/quickwit-search/src/search_response_rest.rs +++ b/quickwit/quickwit-search/src/search_response_rest.rs @@ -13,9 +13,14 @@ // limitations under the License. use std::convert::TryFrom; +use std::fmt; +use std::str::FromStr; +use base64::Engine; +use base64::prelude::BASE64_URL_SAFE_NO_PAD; +use prost::Message; use quickwit_common::truncate_str; -use quickwit_proto::search::SearchResponse; +use quickwit_proto::search::{PartialHit, SearchResponse}; use quickwit_query::aggregations::AggregationResults as AggregationResultsProxy; use quickwit_query::query_ast::QueryAst; use serde::{Deserialize, Serialize}; @@ -46,6 +51,8 @@ pub struct SearchResponseRest { #[schema(value_type = Vec)] /// List of hits returned. pub hits: Vec, + /// [`SearchAfterCursor`] of each hit, in the same order as `hits`. + pub cursors: Vec, /// List of snippets #[schema(value_type = Vec)] #[serde(skip_serializing_if = "Option::is_none")] @@ -65,8 +72,13 @@ impl TryFrom for SearchResponseRest { fn try_from(search_response: SearchResponse) -> Result { let mut documents = Vec::with_capacity(search_response.hits.len()); + let mut cursors = Vec::with_capacity(search_response.hits.len()); let mut snippets = Vec::new(); for hit in search_response.hits { + let partial_hit = hit.partial_hit.ok_or_else(|| { + SearchError::Internal("search response hit is missing its partial hit".to_string()) + })?; + cursors.push(SearchAfterCursor(partial_hit).to_string()); let document: JsonValue = serde_json::from_str(&hit.json).map_err(|err| { SearchError::Internal(format!( "failed to serialize document `{}` to JSON: `{}`", @@ -105,6 +117,7 @@ impl TryFrom for SearchResponseRest { Ok(SearchResponseRest { num_hits: search_response.num_hits, hits: documents, + cursors, snippets: snippet_opt, elapsed_time_micros: search_response.elapsed_time_micros, errors: search_response.errors, @@ -113,6 +126,37 @@ impl TryFrom for SearchResponseRest { } } +/// Opaque cursor of a hit, i.e. its sort values and address, passed as `search_after` to get the +/// hits sorted after it. +/// +/// It is written as the URL-safe base64 of the `PartialHit` protobuf. Datetime sort values are +/// kept as returned by the root search, i.e. in the sort datetime format of the request, which is +/// also how the root parses `search_after`. A cursor is only meaningful for the indexes, query, and +/// sort fields of the search that returned it: the root search rejects the sort values that cannot +/// match the sort fields, but cannot tell whether the cursor comes from another sort. +#[derive(Clone, Debug, PartialEq)] +pub struct SearchAfterCursor(pub PartialHit); + +impl fmt::Display for SearchAfterCursor { + fn fmt(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + let b64_payload = BASE64_URL_SAFE_NO_PAD.encode(self.0.encode_to_vec()); + write!(formatter, "{b64_payload}") + } +} + +impl FromStr for SearchAfterCursor { + type Err = &'static str; + + fn from_str(cursor_str: &str) -> Result { + let base64_decoded: Vec = BASE64_URL_SAFE_NO_PAD + .decode(cursor_str) + .map_err(|_| "search_after cursor is invalid base64")?; + let partial_hit = PartialHit::decode(base64_decoded.as_slice()) + .map_err(|_| "search_after cursor is malformed")?; + Ok(SearchAfterCursor(partial_hit)) + } +} + /// Details on how a query would be executed. #[derive(Serialize, Deserialize, PartialEq, Debug, utoipa::ToSchema)] pub struct SearchPlanResponseRest { @@ -148,3 +192,123 @@ pub struct StorageRequestCount { /// Number of position list downloaded pub position: usize, } + +#[cfg(test)] +mod tests { + use quickwit_proto::search::{Hit, SortByValue, SortValue}; + + use super::*; + + fn partial_hit(sort_value: Option, sort_value2: Option) -> PartialHit { + PartialHit { + sort_value: sort_value.map(SortByValue::from), + sort_value2: sort_value2.map(SortByValue::from), + split_id: "01HZ7Q0K9Y5X2M3N4P5Q6R7S8T".to_string(), + segment_ord: 1, + doc_id: 42, + } + } + + #[test] + fn test_search_after_cursor_round_trip() { + let partial_hits = [ + partial_hit(None, None), + partial_hit(Some(SortValue::I64(1_695_890_000_123)), None), + partial_hit(Some(SortValue::I64(-5)), Some(SortValue::U64(u64::MAX))), + partial_hit(Some(SortValue::F64(0.1)), Some(SortValue::Boolean(true))), + ]; + for partial_hit in partial_hits { + let cursor = SearchAfterCursor(partial_hit.clone()).to_string(); + assert!( + cursor + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'), + "cursor `{cursor}` is not URL-safe" + ); + assert_eq!( + SearchAfterCursor::from_str(&cursor).unwrap(), + SearchAfterCursor(partial_hit) + ); + } + } + + #[test] + fn test_search_after_cursor_encoding_is_stable() { + // Cursors outlive the node that issued them (rolling upgrades, clients holding them), so + // the encoding is a compatibility contract. + let partial_hit = PartialHit { + sort_value: Some(SortValue::I64(5).into()), + sort_value2: None, + split_id: "a".to_string(), + segment_ord: 1, + doc_id: 2, + }; + assert_eq!( + SearchAfterCursor(partial_hit).to_string(), + "EgFhGAEgAlICEAU" + ); + } + + #[test] + fn test_search_after_cursor_rejects_invalid_cursors() { + assert_eq!( + SearchAfterCursor::from_str("not a cursor!").unwrap_err(), + "search_after cursor is invalid base64" + ); + // `_w` is valid base64 for the truncated varint `0xff`. + assert_eq!( + SearchAfterCursor::from_str("_w").unwrap_err(), + "search_after cursor is malformed" + ); + } + + #[test] + fn test_search_response_rest_has_one_cursor_per_hit() { + let first_partial_hit = partial_hit(Some(SortValue::I64(2)), None); + let second_partial_hit = PartialHit { + doc_id: 43, + ..partial_hit(Some(SortValue::I64(1)), None) + }; + let search_response = SearchResponse { + num_hits: 2, + hits: vec![ + Hit { + json: r#"{"body": "first"}"#.to_string(), + partial_hit: Some(first_partial_hit.clone()), + ..Default::default() + }, + Hit { + json: r#"{"body": "second"}"#.to_string(), + partial_hit: Some(second_partial_hit.clone()), + ..Default::default() + }, + ], + ..Default::default() + }; + let search_response_rest = SearchResponseRest::try_from(search_response).unwrap(); + let decoded_partial_hits: Vec = search_response_rest + .cursors + .iter() + .map(|cursor| SearchAfterCursor::from_str(cursor).unwrap().0) + .collect(); + assert_eq!( + decoded_partial_hits, + vec![first_partial_hit, second_partial_hit] + ); + } + + #[test] + fn test_search_response_rest_rejects_hit_without_partial_hit() { + let search_response = SearchResponse { + num_hits: 1, + hits: vec![Hit { + json: "{}".to_string(), + partial_hit: None, + ..Default::default() + }], + ..Default::default() + }; + let error = SearchResponseRest::try_from(search_response).unwrap_err(); + assert!(matches!(error, SearchError::Internal(_)), "{error:?}"); + } +} diff --git a/quickwit/quickwit-serve/src/search_api/rest_handler.rs b/quickwit/quickwit-serve/src/search_api/rest_handler.rs index 01ad87d78e9..326acfcec2a 100644 --- a/quickwit/quickwit-serve/src/search_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/search_api/rest_handler.rs @@ -13,13 +13,16 @@ // limitations under the License. use std::convert::TryFrom; +use std::str::FromStr; use std::sync::Arc; use percent_encoding::percent_decode_str; use quickwit_config::validate_index_id_pattern; use quickwit_proto::search::{CountHits, SortField, SortOrder}; use quickwit_query::query_ast::query_ast_from_user_text; -use quickwit_search::{SearchError, SearchPlanResponseRest, SearchResponseRest, SearchService}; +use quickwit_search::{ + SearchAfterCursor, SearchError, SearchPlanResponseRest, SearchResponseRest, SearchService, +}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value as JsonValue; use tracing::info; @@ -191,6 +194,11 @@ pub struct SearchRequestQueryString { /// The results with rank [start_offset..start_offset + max_hits) are returned #[serde(default)] // Default to 0. (We are 0-indexed) pub start_offset: u64, + /// Opaque cursor from the `cursors` of a previous response. The search returns the hits that + /// come after that hit in the `sort_by` order, i.e. the next page. Use it with the same query + /// and `sort_by`. + #[serde(skip_serializing_if = "Option::is_none")] + pub search_after: Option, /// The output format. #[serde(default)] pub format: BodyFormat, @@ -244,6 +252,14 @@ pub fn search_request_from_api_request( index_id_patterns: Vec, search_request: SearchRequestQueryString, ) -> Result { + let search_after = match &search_request.search_after { + Some(cursor_str) => { + let cursor = SearchAfterCursor::from_str(cursor_str) + .map_err(|msg| SearchError::InvalidArgument(msg.to_string()))?; + Some(cursor.0) + } + None => None, + }; // The query ast below may still contain user input query. The actual // parsing of the user query will happen in the root service, and might require // the user of the docmapper default fields (which we do not have at this point). @@ -262,7 +278,7 @@ pub fn search_request_from_api_request( .map(|agg| serde_json::to_string(&agg).expect("could not serialize JsonValue")), sort_fields: search_request.sort_by.sort_fields, scroll_ttl_secs: None, - search_after: None, + search_after, count_hits: search_request.count_all.into(), ignore_missing_indexes: false, skip_aggregation_finalization: false, @@ -460,6 +476,17 @@ mod tests { use super::*; use crate::recover_fn; + /// The hit encoded by the cursor `EgFhGAEgAlICEAU`. + fn cursor_partial_hit() -> quickwit_proto::search::PartialHit { + quickwit_proto::search::PartialHit { + sort_value: Some(quickwit_proto::search::SortValue::I64(5).into()), + sort_value2: None, + split_id: "a".to_string(), + segment_ord: 1, + doc_id: 2, + } + } + fn search_handler( mock_search_service: MockSearchService, ) -> impl Filter + Clone { @@ -499,6 +526,7 @@ mod tests { let search_response = SearchResponseRest { num_hits: 55, hits: Vec::new(), + cursors: Vec::new(), snippets: None, elapsed_time_micros: 0u64, errors: Vec::new(), @@ -508,6 +536,7 @@ mod tests { let expected_search_response_json: JsonValue = json!({ "num_hits": 55, "hits": [], + "cursors": [], "elapsed_time_micros": 0, }); assert_json_include!( @@ -938,6 +967,51 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_rest_search_api_search_after_parameter() { + let expected_search_after = cursor_partial_hit(); + let mut mock_search_service = MockSearchService::new(); + mock_search_service + .expect_root_search() + .with(predicate::function( + move |search_request: &quickwit_proto::search::SearchRequest| { + search_request.search_after.as_ref() == Some(&expected_search_after) + }, + )) + .times(2) + .returning(|_| Ok(Default::default())); + let rest_search_api_handler = search_handler(mock_search_service); + let get_response = warp::test::request() + .path("/quickwit-demo-index/search?query=*&sort_by=-ts&search_after=EgFhGAEgAlICEAU") + .reply(&rest_search_api_handler) + .await; + assert_eq!(get_response.status(), 200); + let post_response = warp::test::request() + .method("POST") + .path("/quickwit-demo-index/search") + .json(&json!({"query": "*", "sort_by": "-ts", "search_after": "EgFhGAEgAlICEAU"})) + .reply(&rest_search_api_handler) + .await; + assert_eq!(post_response.status(), 200); + } + + #[tokio::test] + async fn test_rest_search_api_invalid_search_after_cursor() { + let mut mock_search_service = MockSearchService::new(); + mock_search_service.expect_root_search().never(); + let rest_search_api_handler = search_handler(mock_search_service); + let response = warp::test::request() + .path("/quickwit-demo-index/search?query=*&search_after=_w") + .reply(&rest_search_api_handler) + .await; + assert_eq!(response.status(), 400); + let response_json: JsonValue = serde_json::from_slice(response.body()).unwrap(); + assert_eq!( + response_json["message"], + "Invalid argument: search_after cursor is malformed" + ); + } + #[tokio::test] async fn test_rest_search_api_with_index_does_not_exist() -> anyhow::Result<()> { let mut mock_search_service = MockSearchService::new(); @@ -1000,7 +1074,7 @@ mod tests { Ok(quickwit_proto::search::SearchResponse { hits: vec![quickwit_proto::search::Hit { json: r#"{"title": "foo", "body": "foo bar baz"}"#.to_string(), - partial_hit: None, + partial_hit: Some(cursor_partial_hit()), snippet: Some(r#"{"title": [], "body": ["foo bar baz"]}"#.to_string()), index_id: "quickwit-demo-index".to_string(), }], @@ -1024,6 +1098,7 @@ mod tests { let expected_response_json = serde_json::json!({ "num_hits": 1, "hits": [{"title": "foo", "body": "foo bar baz"}], + "cursors": ["EgFhGAEgAlICEAU"], "snippets": [{"title": [], "body": ["foo bar baz"]}], "elapsed_time_micros": 16, "errors": [], diff --git a/quickwit/rest-api-tests/scenarii/qw_search_api/0006_search_after.yaml b/quickwit/rest-api-tests/scenarii/qw_search_api/0006_search_after.yaml new file mode 100644 index 00000000000..ae2dd6fa378 --- /dev/null +++ b/quickwit/rest-api-tests/scenarii/qw_search_api/0006_search_after.yaml @@ -0,0 +1,125 @@ +# `-ts` sorts ascending. The cursor of doc 2 falls in the middle of the 2, 3, 4 tie. +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts" + max_hits: 2 +expected: + num_hits: 5 + hits: [{"id": 1}, {"id": 2}] + cursors: + $expect: "len(val) == 2" +--- +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts" + max_hits: 2 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [{"id": 3}, {"id": 4}] +--- +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts" + max_hits: 2 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [{"id": 5}] +--- +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts" + max_hits: 2 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [] + cursors: [] +--- +# Inverting the sort order pages backward from the same cursor. +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts" + max_hits: 4 +expected: + hits: [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}] +--- +endpoint: cursor/search +params: + query: "*" + sort_by: "ts" + max_hits: 10 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [{"id": 3}, {"id": 2}, {"id": 1}] +--- +# A cursor with a sort value cannot be used without `sort_by`. +endpoint: cursor/search +params: + query: "*" + search_after: + $previous: "val['cursors'][-1]" +status_code: 400 +--- +# Without `sort_by`, hits are sorted by descending address, i.e. in reverse ingestion order. +endpoint: cursor/search +params: + query: "*" + max_hits: 2 +expected: + hits: [{"id": 4}, {"id": 1}] +--- +endpoint: cursor/search +params: + query: "*" + max_hits: 2 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [{"id": 3}, {"id": 2}] +--- +# `-_doc` inverts the default order. +endpoint: cursor/search +params: + query: "*" + sort_by: "-_doc" + max_hits: 10 + search_after: + $previous: "val['cursors'][-1]" +expected: + hits: [{"id": 3}, {"id": 1}, {"id": 4}] +--- +# Sub-millisecond precision fields cannot be paged with millisecond sort values. +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts_micros" + max_hits: 1 +--- +endpoint: cursor/search +params: + query: "*" + sort_by: "-ts_micros" + search_after: + $previous: "val['cursors'][-1]" +status_code: 400 +--- +endpoint: cursor/search +params: + query: "*" + search_after: "not a cursor!" +status_code: 400 +--- +# Decodes into a cursor without a split ID. +endpoint: cursor/search +params: + query: "*" + search_after: "GAA" +status_code: 400 diff --git a/quickwit/rest-api-tests/scenarii/qw_search_api/_setup.quickwit.yaml b/quickwit/rest-api-tests/scenarii/qw_search_api/_setup.quickwit.yaml index e410ecd96c0..39e71f89aad 100644 --- a/quickwit/rest-api-tests/scenarii/qw_search_api/_setup.quickwit.yaml +++ b/quickwit/rest-api-tests/scenarii/qw_search_api/_setup.quickwit.yaml @@ -125,4 +125,43 @@ params: ndjson: - {"ts": "2022-12-16T10:00:56.297Z"} - {"ts": "2022-12-16T10:00:57.000Z"} - - {"ts": "2022-12-16T10:00:57.297Z"} \ No newline at end of file + - {"ts": "2022-12-16T10:00:57.297Z"} +--- +method: DELETE +endpoint: indexes/cursor +status_code: null +--- +method: POST +endpoint: indexes/ +json: + version: "0.7" + index_id: cursor + doc_mapping: + timestamp_field: ts + mode: strict + field_mappings: + - name: ts + type: datetime + fast: true + input_formats: ["rfc3339"] + fast_precision: milliseconds + - name: ts_micros + type: datetime + fast: true + input_formats: ["rfc3339"] + fast_precision: microseconds + - name: id + type: u64 +--- +# A single batch lands in a single split, so docs are addressed in ingestion order: the tied docs +# 2, 3, and 4 keep that order, and the address order differs from the timestamp order. +method: POST +endpoint: cursor/ingest +params: + commit: force +ndjson: + - {"ts": "2022-12-16T10:00:00.003Z", "ts_micros": "2022-12-16T10:00:00.003Z", "id": 5} + - {"ts": "2022-12-16T10:00:00.002Z", "ts_micros": "2022-12-16T10:00:00.002Z", "id": 2} + - {"ts": "2022-12-16T10:00:00.002Z", "ts_micros": "2022-12-16T10:00:00.002Z", "id": 3} + - {"ts": "2022-12-16T10:00:00.001Z", "ts_micros": "2022-12-16T10:00:00.001Z", "id": 1} + - {"ts": "2022-12-16T10:00:00.002Z", "ts_micros": "2022-12-16T10:00:00.002Z", "id": 4} diff --git a/quickwit/rest-api-tests/scenarii/qw_search_api/_teardown.quickwit.yaml b/quickwit/rest-api-tests/scenarii/qw_search_api/_teardown.quickwit.yaml index ebfa1c4931b..99159d028d0 100644 --- a/quickwit/rest-api-tests/scenarii/qw_search_api/_teardown.quickwit.yaml +++ b/quickwit/rest-api-tests/scenarii/qw_search_api/_teardown.quickwit.yaml @@ -5,4 +5,7 @@ method: DELETE endpoint: indexes/nested --- method: DELETE -endpoint: indexes/millisec \ No newline at end of file +endpoint: indexes/millisec +--- +method: DELETE +endpoint: indexes/cursor