diff --git a/docs/configuration/index-config.md b/docs/configuration/index-config.md index 067f16c7fc8..caae1e19378 100644 --- a/docs/configuration/index-config.md +++ b/docs/configuration/index-config.md @@ -594,6 +594,7 @@ This section describes indexing settings for a given index. | ------------- | ------------- | ------------- | | `commit_timeout_secs` | Maximum number of seconds before committing a split since its creation. | `60` | | `split_num_docs_target` | Target number of docs per split. | `10000000` | +| `sort_fields` | Physical document order within splits. An empty list disables sorting; currently at most one raw text fast field is supported. | `[]` | | `merge_policy` | Describes the strategy used to trigger split merge operations for logs/traces (see [Merge policies](#merge-policies) section below). | | `parquet_merge_policy` | Describes the merge policy for Parquet (metrics/sketches) splits (see [Parquet merge policy](#parquet-merge-policy) section below). | | `parquet_indexing` | Parquet-specific indexing settings: sort schema, window duration (see [Parquet indexing settings](#parquet-indexing-settings) section below). | @@ -609,6 +610,33 @@ When decommissioning definitively an indexer node that received data through the ::: +### Physical document sorting + +To sort newly written splits by service, configure a single-valued raw text fast field: + +```yaml +doc_mapping: + field_mappings: + - name: service + type: text + tokenizer: raw + fast: true + tag_fields: [service] + +indexing_settings: + sort_fields: [service] +``` + +`sort_fields` accepts a list of mapped field names but currently rejects more than one entry. +A bare name sorts ascending; prefix it with `-` for descending order, for example +`sort_fields: [-service]`. An explicit `+` also means ascending. Nested paths and escaped dots +use the same syntax as other field references. Empty names and repeated order prefixes are +rejected. The field must be indexed with the `raw` tokenizer and stored as a fast field. +Arrays, numeric fields, and unmapped JSON paths are not supported. Strings are compared by their +UTF-8 bytes. Missing and null values sort +first for `asc` and last for `desc`; an empty string is a distinct value. Equal values retain all +documents. Physical ordering does not change query result ordering. + ### Merge policies Quickwit makes it possible to define the strategy used to decide which splits should be merged together and when. diff --git a/docs/internals/sorting.md b/docs/internals/sorting.md index 1dc2eca1805..f2179a21f0f 100644 --- a/docs/internals/sorting.md +++ b/docs/internals/sorting.md @@ -12,6 +12,13 @@ to use for sorting. Sorting is Descending by default. The sorting order can be r a field name with a hyphen `-`. The special value `_score` means sorting by score, it is also Descending by default. +The prefix parser is shared with `indexing_settings.sort_fields` through +`quickwit_common::parse_sort_fields`, which returns field names and orders. +Each caller supplies its default and reverse orders: descending by default for search, +ascending by default for indexing. +A `+` keeps that default and a `-` reverses it. Each caller handles its own list format +and field validation. + In case of equality between two documents, the GlobalDocId, composed of (SplitId, SegmentId, DocId) is used as a tie breaker. It is used to sort in the same order as the first field being sorted by. This means it is in Descending order by default. diff --git a/docs/internals/split-format.md b/docs/internals/split-format.md index bd85ae8dbe9..fc2809f3e04 100644 --- a/docs/internals/split-format.md +++ b/docs/internals/split-format.md @@ -12,6 +12,24 @@ It contains the field name, type and capabilities. - a versioned protobuf entry named `split_recovery_metadata` containing the immutable split metadata and direct lineage needed to reconstruct its metastore record. +Splits may declare `sort_fields` in their metastore metadata and recovery entry. Each entry +records a field name, order, and type, for example +`{"field": "service", "order": "asc", "type": "text"}`. The type is resolved from the writer's +schema, even for an all-null field; it is independent of the physical column encoding. A nonempty +declaration with a missing or unknown type is rejected, including during recovery. + +In Rust, `IndexingSortSchema` represents the ordered list used for merge compatibility, with +`SortFieldMetadata` describing each field. It serializes directly as the `sort_fields` array; +the wrapper does not add a level to the stored JSON format. + +The list is empty for legacy and unsorted splits; currently writers support one scalar raw text +field, with ascending or descending byte ordering. Missing values sort first ascending and last +descending. The declaration must agree with both the sort setting and schema in Tantivy's +`meta.json`. Merge planners group by the complete sort schema, including type, and merge execution +rejects mismatches rather than adopting the current index configuration as the order of older +data. Splits must also have the same document-mapping UID to merge; matching sort declarations +do not remove that constraint. + The split file data layout looks like this: - concatenation all of the files in the split - a footer diff --git a/quickwit/quickwit-common/src/lib.rs b/quickwit/quickwit-common/src/lib.rs index 2ed260ebcee..18edf6579a1 100644 --- a/quickwit/quickwit-common/src/lib.rs +++ b/quickwit/quickwit-common/src/lib.rs @@ -40,6 +40,7 @@ pub mod retry; pub mod ring_buffer; pub mod runtimes; pub mod shared_consts; +mod sort_fields; pub mod sorted_iter; pub mod stream_utils; pub mod temp_dir; @@ -66,6 +67,7 @@ pub use kill_switch::KillSwitch; pub use metrics_specific::*; pub use path_hasher::PathHasher; pub use progress::{Progress, ProtectedZoneGuard}; +pub use sort_fields::parse_sort_fields; pub use stream_utils::{BoxStream, ServiceStream}; /// Returns true at compile time. This function is mostly used with serde to initialize boolean diff --git a/quickwit/quickwit-common/src/sort_fields.rs b/quickwit/quickwit-common/src/sort_fields.rs new file mode 100644 index 00000000000..5af3295bb55 --- /dev/null +++ b/quickwit/quickwit-common/src/sort_fields.rs @@ -0,0 +1,41 @@ +// 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. + +/// Parses field names with an optional direction prefix. Bare names and `+` use +/// `default_order`; `-` uses `reverse_order`. Indexing uses ascending by default, while the +/// REST search API historically uses descending. +/// +/// Callers handle list delimiters, whitespace, empty entries, and field validation. +/// Only one leading sign is consumed; the rest of each field name is preserved verbatim. +pub fn parse_sort_fields( + fields: impl IntoIterator>, + default_order: Order, + reverse_order: Order, +) -> Vec<(String, Order)> { + fields + .into_iter() + .map(|expression| { + let expression = expression.as_ref(); + let (field, order) = if let Some(field) = expression.strip_prefix('-') { + (field, reverse_order) + } else { + ( + expression.strip_prefix('+').unwrap_or(expression), + default_order, + ) + }; + (field.to_string(), order) + }) + .collect() +} diff --git a/quickwit/quickwit-compaction/src/planner/compaction_state.rs b/quickwit/quickwit-compaction/src/planner/compaction_state.rs index 79cfad6ac34..1900e5331c2 100644 --- a/quickwit/quickwit-compaction/src/planner/compaction_state.rs +++ b/quickwit/quickwit-compaction/src/planner/compaction_state.rs @@ -36,6 +36,7 @@ pub struct CompactionPartitionKey { pub source_id: SourceId, pub partition_id: u64, pub doc_mapping_uid: DocMappingUid, + pub sort_fields: quickwit_metastore::IndexingSortSchema, } impl CompactionPartitionKey { @@ -45,6 +46,7 @@ impl CompactionPartitionKey { source_id: split.source_id.clone(), partition_id: split.partition_id, doc_mapping_uid: split.doc_mapping_uid, + sort_fields: split.sort_fields.clone(), } } } @@ -319,6 +321,56 @@ mod tests { assert_eq!(state.partition_keys().len(), 1); } + #[test] + fn test_sort_fields_separate_compaction_groups() { + use quickwit_metastore::SortFieldMetadata; + use quickwit_proto::search::{SortFieldType, SortOrder}; + + let index_uid = IndexUid::for_test("test-index", 0); + let mut state = CompactionState::default(); + + for (group, sort_fields) in [ + Vec::new(), + vec![SortFieldMetadata { + field: "service".to_string(), + order: SortOrder::Asc, + field_type: SortFieldType::Text, + }], + vec![SortFieldMetadata { + field: "service".to_string(), + order: SortOrder::Desc, + field_type: SortFieldType::Text, + }], + vec![SortFieldMetadata { + field: "service".to_string(), + order: SortOrder::Asc, + field_type: SortFieldType::I64, + }], + ] + .into_iter() + .enumerate() + { + for i in 0..2 { + let mut split = test_split(&format!("split-{group}-{i}"), &index_uid); + split.sort_fields.fields = sort_fields.clone(); + state.track_split(split); + } + } + let keys = state.partition_keys(); + assert_eq!(keys.len(), 4); + + for key in keys { + state.plan_partition(&key, &*test_merge_policy()); + } + assert_eq!(state.pending_operations.len(), 4); + + for pending in state.pending_operations.iter() { + let splits = &pending.operation.splits; + assert_eq!(splits.len(), 2); + assert_eq!(splits[0].sort_fields, splits[1].sort_fields); + } + } + #[test] fn test_plan_partition_moves_splits_to_in_flight() { let index_uid = IndexUid::for_test("test-index", 0); diff --git a/quickwit/quickwit-config/src/index_config/mod.rs b/quickwit/quickwit-config/src/index_config/mod.rs index 0e7fb242166..5995771a0c6 100644 --- a/quickwit/quickwit-config/src/index_config/mod.rs +++ b/quickwit/quickwit-config/src/index_config/mod.rs @@ -13,7 +13,7 @@ // limitations under the License. pub(crate) mod serialize; - +mod sort_fields; use std::collections::HashSet; use std::hash::{Hash, Hasher}; use std::num::NonZeroUsize; @@ -34,6 +34,7 @@ use rand::{RngExt, distr, rng}; use serde::{Deserialize, Serialize}; pub use serialize::{load_index_config_from_user_config, load_index_config_update}; use siphasher::sip::SipHasher; +pub use sort_fields::IndexingSortField; use tracing::warn; use crate::index_config::serialize::VersionedIndexConfig; @@ -104,6 +105,15 @@ impl Default for IndexingResources { #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Hash, utoipa::ToSchema)] #[serde(deny_unknown_fields)] pub struct IndexingSettings { + /// Physical document ordering within splits, independent of query result ordering. + /// Currently at most one scalar raw text fast field is supported. + #[serde( + default, + skip_serializing_if = "Vec::is_empty", + with = "sort_fields::config" + )] + #[schema(value_type = Vec)] + pub sort_fields: Vec, #[schema(default = 60)] #[serde(default = "IndexingSettings::default_commit_timeout_secs")] pub commit_timeout_secs: usize, @@ -206,6 +216,17 @@ impl Default for ParquetIndexingConfig { } impl IndexingSettings { + pub fn validate_sort_fields(&self, doc_mapper: &DocMapper) -> anyhow::Result<()> { + ensure!( + self.sort_fields.len() <= 1, + "indexing_settings.sort_fields supports at most one field" + ); + for sort_field in &self.sort_fields { + doc_mapper.validate_sort_field(&sort_field.field)?; + } + Ok(()) + } + pub fn commit_timeout(&self) -> Duration { Duration::from_secs(self.commit_timeout_secs as u64) } @@ -252,6 +273,7 @@ impl IndexingSettings { impl Default for IndexingSettings { fn default() -> Self { Self { + sort_fields: Vec::new(), commit_timeout_secs: Self::default_commit_timeout_secs(), docstore_blocksize: Self::default_docstore_blocksize(), docstore_compression_level: Self::default_docstore_compression_level(), @@ -681,7 +703,8 @@ pub(super) fn validate_index_config( // Note: this needs a deep refactoring to separate the doc mapping configuration, // and doc mapper implementations. // TODO see if we should store the byproducton the IndexConfig. - build_doc_mapper(doc_mapping, search_settings)?; + let doc_mapper = build_doc_mapper(doc_mapping, search_settings)?; + indexing_settings.validate_sort_fields(&doc_mapper)?; indexing_settings.merge_policy.validate()?; indexing_settings.resources.validate()?; @@ -1392,3 +1415,149 @@ mod tests { assert_eq!(updated_doc_mapping.mode, Mode::Strict); } } + +#[cfg(test)] +mod sort_field_tests { + use quickwit_proto::search::SortOrder; + use serde_json::json; + + use super::{ + DocMapping, IndexConfig, IndexingSettings, IndexingSortField, SearchSettings, + validate_index_config, + }; + + #[test] + fn test_sort_field_validation() { + let mapping: DocMapping = serde_json::from_value(json!({"field_mappings": [ + {"name": "service", "type": "text", "tokenizer": "raw"} + ]})) + .unwrap(); + let mut settings = IndexingSettings::default(); + validate_index_config(&mapping, &settings, &SearchSettings::default(), &None).unwrap(); + settings.sort_fields.push(IndexingSortField { + field: "service".to_string(), + order: SortOrder::Asc, + }); + let error = validate_index_config(&mapping, &settings, &SearchSettings::default(), &None) + .unwrap_err(); + assert_eq!( + error.to_string(), + "sort field `service` must be a fast field; set `fast: true`" + ); + } + + #[test] + fn test_sort_fields_shorthand() { + for (expression, field, order, canonical) in [ + ("service", "service", SortOrder::Asc, "service"), + ("+service", "service", SortOrder::Asc, "service"), + ("-service", "service", SortOrder::Desc, "-service"), + ( + "-resource.service", + "resource.service", + SortOrder::Desc, + "-resource.service", + ), + ( + r"service\.name", + r"service\.name", + SortOrder::Asc, + r"service\.name", + ), + ] { + let settings: IndexingSettings = + serde_yaml::from_str(&format!("sort_fields: [{expression}]\n")).unwrap(); + assert_eq!( + settings.sort_fields, + vec![IndexingSortField { + field: field.to_string(), + order, + }] + ); + let serialized = serde_json::to_value(&settings).unwrap(); + assert_eq!(serialized["sort_fields"], json!([canonical])); + assert_eq!( + serde_json::from_value::(serialized).unwrap(), + settings + ); + } + } + + #[test] + fn test_sort_fields_invalid_shorthand() { + for expression in [ + "", + "+", + "-", + "--service", + "++service", + "-+service", + "+-service", + " service", + "service desc", + ] { + let error = serde_json::from_value::(json!({ + "sort_fields": [expression] + })) + .unwrap_err(); + assert!( + error.to_string().contains("invalid sort field"), + "{expression:?}: {error}" + ); + } + for value in [ + json!("service"), + json!([42]), + json!([null]), + json!([{"field": "service"}]), + ] { + serde_json::from_value::(json!({"sort_fields": value})).unwrap_err(); + } + } + + #[test] + fn test_sort_field_serialization_and_pipeline_fingerprint() { + let legacy: IndexingSettings = serde_json::from_value(json!({})).unwrap(); + assert!(legacy.sort_fields.is_empty()); + assert!( + serde_json::to_value(&legacy) + .unwrap() + .get("sort_fields") + .is_none() + ); + + let mut config = IndexConfig::for_test("test-index", "ram://indexes/test-index"); + let original_fingerprint = config.indexing_params_fingerprint(); + config.indexing_settings.sort_fields = vec![IndexingSortField { + field: "service".to_string(), + order: Default::default(), + }]; + assert_ne!(original_fingerprint, config.indexing_params_fingerprint()); + let serialized = serde_json::to_value(&config.indexing_settings).unwrap(); + let deserialized: IndexingSettings = serde_json::from_value(serialized).unwrap(); + assert_eq!(deserialized, config.indexing_settings); + + let ascending_fingerprint = config.indexing_params_fingerprint(); + config.indexing_settings.sort_fields = + serde_json::from_value::(json!({"sort_fields": ["-service"]})) + .unwrap() + .sort_fields; + assert_ne!(ascending_fingerprint, config.indexing_params_fingerprint()); + config.indexing_settings.sort_fields.clear(); + assert_eq!(original_fingerprint, config.indexing_params_fingerprint()); + } + + #[test] + fn test_sort_fields_rejects_multiple_fields() { + let settings: IndexingSettings = + serde_json::from_value(json!({"sort_fields": ["service", "-host"]})).unwrap(); + let mapping: DocMapping = serde_json::from_value(json!({"field_mappings": [ + {"name": "service", "type": "text", "tokenizer": "raw", "fast": true}, + {"name": "host", "type": "text", "tokenizer": "raw", "fast": true} + ]})) + .unwrap(); + let error = validate_index_config(&mapping, &settings, &SearchSettings::default(), &None) + .unwrap_err(); + assert!(error.to_string().contains("at most one field")); + } +} diff --git a/quickwit/quickwit-config/src/index_config/sort_fields.rs b/quickwit/quickwit-config/src/index_config/sort_fields.rs new file mode 100644 index 00000000000..17d07e90aef --- /dev/null +++ b/quickwit/quickwit-config/src/index_config/sort_fields.rs @@ -0,0 +1,72 @@ +// 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 quickwit_proto::search::SortOrder; +use serde::{Deserialize, Serialize}; + +/// A configured physical sort field. Its logical type is resolved from the split's schema. +/// Currently only raw strings are supported. Missing values sort first ascending, last descending. +#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct IndexingSortField { + pub field: String, + #[serde(default)] + pub order: SortOrder, +} + +/// Index configuration uses field names prefixed with `-` for descending order. +/// The configured declaration does not include a type; split metadata records the resolved type. +pub(super) mod config { + use quickwit_common::parse_sort_fields; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + + use super::{IndexingSortField, SortOrder}; + + pub fn serialize( + sort_fields: &[IndexingSortField], + serializer: S, + ) -> Result + where + S: Serializer, + { + let fields: Vec = sort_fields + .iter() + .map(|sort| match sort.order { + SortOrder::Asc => sort.field.clone(), + SortOrder::Desc => format!("-{}", sort.field), + }) + .collect(); + fields.serialize(serializer) + } + + pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> + where D: Deserializer<'de> { + let fields = Vec::::deserialize(deserializer)?; + let mut sort_fields = Vec::with_capacity(fields.len()); + let parsed_fields = parse_sort_fields(&fields, SortOrder::Asc, SortOrder::Desc); + for (expression, (field, order)) in fields.iter().zip(parsed_fields) { + if field.is_empty() + || field.starts_with(['+', '-']) + || field.chars().any(char::is_whitespace) + { + return Err(serde::de::Error::custom(format!( + "invalid sort field `{expression}`: expected a field name optionally prefixed \ + with `+` or `-`" + ))); + } + sort_fields.push(IndexingSortField { field, order }); + } + Ok(sort_fields) + } +} diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs index 904e65f8180..eb1e286ef1d 100644 --- a/quickwit/quickwit-config/src/lib.rs +++ b/quickwit/quickwit-config/src/lib.rs @@ -48,9 +48,9 @@ pub use docs_clustering::{ // See #2048 use index_config::serialize::{IndexConfigV0_8, VersionedIndexConfig}; pub use index_config::{ - IndexConfig, IndexingResources, IndexingSettings, IngestSettings, ParquetIndexingConfig, - RetentionPolicy, SearchSettings, build_doc_mapper, load_index_config_from_user_config, - load_index_config_update, prepare_doc_mapping_update, + IndexConfig, IndexingResources, IndexingSettings, IndexingSortField, IngestSettings, + ParquetIndexingConfig, RetentionPolicy, SearchSettings, build_doc_mapper, + load_index_config_from_user_config, load_index_config_update, prepare_doc_mapping_update, }; pub use quickwit_doc_mapper::DocMapping; use serde::Serialize; diff --git a/quickwit/quickwit-doc-mapper/src/doc_mapper/doc_mapper_impl.rs b/quickwit/quickwit-doc-mapper/src/doc_mapper/doc_mapper_impl.rs index 087fe2839ab..573b4e96fd1 100644 --- a/quickwit/quickwit-doc-mapper/src/doc_mapper/doc_mapper_impl.rs +++ b/quickwit/quickwit-doc-mapper/src/doc_mapper/doc_mapper_impl.rs @@ -425,6 +425,33 @@ fn extract_single_obj( } impl DocMapper { + /// Validates the physical index sort contract: a scalar text field with raw + /// indexing and raw fast-field values. Missing values follow the sort direction. + pub fn validate_sort_field(&self, field_name: &str) -> anyhow::Result<()> { + let Some(FieldMappingType::Text(options, Cardinality::SingleValued)) = + self.field_mappings.find_field_mapping_type(field_name) + else { + bail!( + "sort field `{field_name}` must be an explicitly mapped single-valued text field" + ); + }; + let text_options: tantivy::schema::TextOptions = options.into(); + if !text_options.is_fast() { + bail!("sort field `{field_name}` must be a fast field; set `fast: true`"); + } + if let Some(normalizer) = text_options.get_fast_field_tokenizer_name() { + if normalizer != "raw" { + bail!( + "sort field `{field_name}` must use the `raw` fast-field normalizer, got \ + `{normalizer}`" + ); + } + } + validate_tag(field_name, &self.schema) + .with_context(|| format!("invalid sort field `{field_name}`"))?; + Ok(()) + } + /// Returns the unique identifier of the doc mapping. pub fn doc_mapping_uid(&self) -> DocMappingUid { self.doc_mapping_uid @@ -728,6 +755,73 @@ mod tests { FIELD_PRESENCE_FIELD_NAME, SOURCE_FIELD_NAME, }; + fn validate_sort_field(field: JsonValue, field_name: &str) -> anyhow::Result<()> { + let builder: DocMapperBuilder = serde_json::from_value(json!({"field_mappings": [field]}))?; + builder.try_build()?.validate_sort_field(field_name) + } + + #[test] + fn test_sort_field_validation() { + let valid = json!({"name": "service", "type": "text", "tokenizer": "raw", "fast": true}); + validate_sort_field(valid.clone(), "service").unwrap(); + validate_sort_field(valid.clone(), "unknown").unwrap_err(); + validate_sort_field(valid.clone(), "").unwrap_err(); + + for (key, value) in [ + ("type", json!("array")), + ("tokenizer", json!("default")), + ] { + let mut invalid = valid.clone(); + invalid[key] = value; + validate_sort_field(invalid, "service").unwrap_err(); + } + for (fast, expected_error) in [ + ( + json!(false), + "sort field `service` must be a fast field; set `fast: true`", + ), + ( + json!({"normalizer": "lowercase"}), + "sort field `service` must use the `raw` fast-field normalizer, got `lowercase`", + ), + ] { + let mut invalid = valid.clone(); + invalid["fast"] = fast; + assert_eq!( + validate_sort_field(invalid, "service") + .unwrap_err() + .to_string(), + expected_error + ); + } + validate_sort_field( + json!({"name": "service", "type": "u64", "fast": true}), + "service", + ) + .unwrap_err(); + validate_sort_field( + json!({"name": "service", "type": "text", "indexed": false, "fast": true}), + "service", + ) + .unwrap_err(); + } + + #[test] + fn test_sort_field_nested_and_escaped_names() { + validate_sort_field( + json!({"name": "resource", "type": "object", "field_mappings": [ + {"name": "service", "type": "text", "tokenizer": "raw", "fast": true} + ]}), + "resource.service", + ) + .unwrap(); + validate_sort_field( + json!({"name": "service.name", "type": "text", "tokenizer": "raw", "fast": true}), + r"service\.name", + ) + .unwrap(); + } + fn example_json_doc_value() -> JsonValue { serde_json::json!({ "timestamp": 1586960586i64, diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 742c94c25c1..53dcc547171 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -33,11 +33,13 @@ use quickwit_common::temp_dir::TempDirectory; use quickwit_config::IndexingSettings; use quickwit_doc_mapper::DocMapper; use quickwit_metastore::checkpoint::{IndexCheckpointDelta, SourceCheckpointDelta}; +use quickwit_metastore::{IndexingSortSchema, SortFieldMetadata}; use quickwit_metrics::GaugeGuard; use quickwit_proto::indexing::{IndexingPipelineId, PipelineMetrics}; use quickwit_proto::metastore::{ LastDeleteOpstampRequest, MetastoreService, MetastoreServiceClient, }; +use quickwit_proto::search::{SortFieldType, SortOrder}; use quickwit_proto::types::DocMappingUid; use quickwit_query::get_quickwit_fastfield_normalizer_manager; use serde::Serialize; @@ -97,6 +99,7 @@ struct IndexerState { publish_lock: PublishLock, schema: Schema, doc_mapping_uid: DocMappingUid, + sort_fields: IndexingSortSchema, tokenizer_manager: TokenizerManager, max_num_partitions: NonZeroU32, index_settings: IndexSettings, @@ -137,6 +140,7 @@ impl IndexerState { partition_id, last_delete_opstamp, self.doc_mapping_uid, + self.sort_fields.clone(), self.indexing_directory.clone(), index_builder, io_controls, @@ -527,7 +531,7 @@ impl Handler for Indexer { impl Indexer { #[allow(clippy::too_many_arguments)] - pub fn new( + pub fn try_new( pipeline_id: IndexingPipelineId, doc_mapper: Arc, metastore: MetastoreServiceClient, @@ -537,14 +541,44 @@ impl Indexer { index_serializer_mailbox: Mailbox, fingerprinter_opt: Option, indexing_io_throughput_limiter_opt: Option, - ) -> Self { + ) -> anyhow::Result { + indexing_settings.validate_sort_fields(&doc_mapper)?; + // Validation guarantees a single raw text field. Resolve the schema once for this + // pipeline and pass it to every split, including splits with only missing values. + let sort_fields = IndexingSortSchema { + fields: indexing_settings + .sort_fields + .iter() + .map(|sort| SortFieldMetadata { + field: sort.field.clone(), + order: sort.order, + field_type: SortFieldType::Text, + }) + .collect(), + }; + if !indexing_settings.sort_fields.is_empty() { + anyhow::ensure!( + fingerprinter_opt.is_none(), + "indexing_settings.sort_fields cannot be combined with document fingerprint \ + clustering" + ); + } let schema = doc_mapper.schema(); let tokenizer_manager = doc_mapper.tokenizer_manager().clone(); let docstore_compression = Compressor::Zstd(ZstdCompressor { compression_level: Some(indexing_settings.docstore_compression_level), }); let index_settings = IndexSettings { - sort_by_field: None, + sort_by_field: sort_fields + .fields + .first() + .map(|sort| tantivy::IndexSortByField { + field: sort.field.clone(), + order: match sort.order { + SortOrder::Asc => tantivy::Order::Asc, + SortOrder::Desc => tantivy::Order::Desc, + }, + }), docstore_blocksize: indexing_settings.docstore_blocksize, docstore_compression, docstore_compress_dedicated_thread: true, @@ -559,7 +593,7 @@ impl Indexer { cooperative_indexing_permits, ) }); - Self { + Ok(Self { indexer_state: IndexerState { pipeline_id, metastore: metastore.clone(), @@ -569,6 +603,7 @@ impl Indexer { publish_lock: PublishLock::default(), schema, doc_mapping_uid: doc_mapper.doc_mapping_uid(), + sort_fields, tokenizer_manager: tokenizer_manager.tantivy_manager().clone(), index_settings, max_num_partitions: doc_mapper.max_num_partitions(), @@ -578,7 +613,7 @@ impl Indexer { index_serializer_mailbox, indexing_workbench_opt: None, counters: IndexerCounters::default(), - } + }) } fn memory_usage(&self) -> ByteSize { @@ -728,7 +763,7 @@ mod tests { index_serializer_mailbox: Mailbox, fingerprinter_opt: Option, ) -> Indexer { - Indexer::new( + Indexer::try_new( pipeline_id, doc_mapper, metastore, @@ -739,6 +774,50 @@ mod tests { fingerprinter_opt, None, ) + .unwrap() + } + + #[tokio::test] + async fn test_sort_fields_reject_fingerprint_clustering() { + let universe = Universe::with_accelerated_time(); + let (mailbox, _inbox) = universe.create_test_mailbox(); + let doc_mapper: DocMapper = serde_json::from_value(serde_json::json!({ + "field_mappings": [{"name": "service", "type": "text", "tokenizer": "raw", "fast": true}] + })).unwrap(); + let clustering: DocsClusteringConfig = serde_json::from_value(serde_json::json!([ + {"fingerprint": [{"kind": "structure"}]} + ])) + .unwrap(); + let result = Indexer::try_new( + IndexingPipelineId { + index_uid: IndexUid::for_test("test-index", 0), + source_id: "test-source".to_string(), + node_id: NodeId::from_str("test-node"), + pipeline_uid: PipelineUid::default(), + }, + Arc::new(doc_mapper), + MetastoreServiceClient::from_mock(MockMetastoreService::new()), + TempDirectory::for_test(), + IndexingSettings { + sort_fields: vec![quickwit_config::IndexingSortField { + field: "service".to_string(), + order: Default::default(), + }], + ..IndexingSettings::for_test() + }, + None, + mailbox, + Some(Fingerprinter::new(&clustering)), + None, + ); + assert!( + result + .err() + .unwrap() + .to_string() + .contains("cannot be combined") + ); + universe.assert_quit().await; } #[test] diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs index 466bbb5dd58..91e4f12dd60 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs @@ -367,7 +367,7 @@ impl IndexingPipeline { .spawn(index_serializer); // Indexer - let indexer = Indexer::new( + let indexer = Indexer::try_new( self.params.pipeline_id.clone(), self.params.doc_mapper.clone(), self.params.metastore.clone(), @@ -377,7 +377,7 @@ impl IndexingPipeline { index_serializer_mailbox, self.params.fingerprinter_opt.clone(), self.params.indexing_io_throughput_limiter_opt.clone(), - ); + )?; let (indexer_mailbox, indexer_handle) = ctx .spawn_actor() .set_backpressure_micros_counter(counter!(parent: BACKPRESSURE_MICROS, labels: [label_values!(ACTOR_NAME => "indexer")])) diff --git a/quickwit/quickwit-indexing/src/actors/merge_executor.rs b/quickwit/quickwit-indexing/src/actors/merge_executor.rs index 32a9111cbdf..329a2f8a459 100644 --- a/quickwit/quickwit-indexing/src/actors/merge_executor.rs +++ b/quickwit/quickwit-indexing/src/actors/merge_executor.rs @@ -18,7 +18,7 @@ use std::path::Path; use std::sync::Arc; use std::time::Instant; -use anyhow::{Context, anyhow}; +use anyhow::{Context, anyhow, ensure}; use async_trait::async_trait; use fail::fail_point; use itertools::Itertools; @@ -190,6 +190,28 @@ impl Handler for MergeExecutor { fn combine_index_meta(mut index_metas: Vec) -> anyhow::Result { let mut union_index_meta = index_metas.pop().with_context(|| "only one IndexMeta")?; for index_meta in index_metas { + ensure!( + index_meta.index_settings.sort_by_field + == union_index_meta.index_settings.sort_by_field, + "cannot merge indexes with different physical sort fields" + ); + if let Some(sort_field) = &index_meta.index_settings.sort_by_field { + let field = index_meta.schema.get_field(&sort_field.field)?; + let union_field = union_index_meta.schema.get_field(&sort_field.field)?; + ensure!( + index_meta + .schema + .get_field_entry(field) + .field_type() + .value_type() + == union_index_meta + .schema + .get_field_entry(union_field) + .field_type() + .value_type(), + "cannot merge indexes with different physical sort fields" + ); + } union_index_meta.segments.extend(index_meta.segments); } Ok(union_index_meta) @@ -198,14 +220,57 @@ fn combine_index_meta(mut index_metas: Vec) -> anyhow::Result], + splits: &[SplitMetadata], tokenizer_manager: &TokenizerManager, ) -> anyhow::Result<(IndexMeta, Vec>)> { + ensure!( + tantivy_dirs.len() == splits.len(), + "merge input metadata count mismatch" + ); let mut directories: Vec> = Vec::new(); let mut index_metas = Vec::new(); - for tantivy_dir in tantivy_dirs { + for (tantivy_dir, split) in tantivy_dirs.iter().zip(splits) { directories.push(tantivy_dir.clone()); let index_meta = open_index(tantivy_dir.clone(), tokenizer_manager)?.load_metas()?; + // Verify the persisted declaration against the actual input, not the current mapping. + let sort_schema_matches = match ( + &index_meta.index_settings.sort_by_field, + split.sort_fields.fields.as_slice(), + ) { + (None, []) => true, + (Some(actual_sort), [sort_field]) => { + use quickwit_proto::search::{SortFieldType, SortOrder}; + use tantivy::schema::Type; + + let field = index_meta.schema.get_field(&actual_sort.field)?; + let field_entry = index_meta.schema.get_field_entry(field); + let field_type = match field_entry.field_type().value_type() { + Type::Str => SortFieldType::Text, + Type::I64 => SortFieldType::I64, + Type::U64 => SortFieldType::U64, + Type::F64 => SortFieldType::F64, + Type::Date => SortFieldType::Datetime, + Type::Bytes => SortFieldType::Bytes, + Type::Bool => SortFieldType::Bool, + other => anyhow::bail!("unsupported physical sort field type {other:?}"), + }; + let order = match actual_sort.order { + tantivy::Order::Asc => SortOrder::Asc, + tantivy::Order::Desc => SortOrder::Desc, + }; + field_entry.is_fast() + && actual_sort.field == sort_field.field + && order == sort_field.order + && field_type == sort_field.field_type + } + _ => false, + }; + ensure!( + sort_schema_matches, + "split {} physical sort schema disagrees with metastore metadata", + split.split_id + ); index_metas.push(index_meta); } let union_index_meta = combine_index_meta(index_metas)?; @@ -292,6 +357,11 @@ pub fn merge_split_attrs( .first() .ok_or_else(|| anyhow::anyhow!("attempted to merge zero splits"))? .doc_mapping_uid; + let sort_fields = splits[0].sort_fields.clone(); + ensure!( + splits.iter().all(|split| split.sort_fields == sort_fields), + "cannot merge splits with different physical sort fields" + ); if splits .iter() .any(|split| split.doc_mapping_uid != doc_mapping_uid) @@ -299,11 +369,12 @@ pub fn merge_split_attrs( anyhow::bail!("attempted to merge splits with different doc mapping uid"); } Ok(SplitAttrs { + split_id: merge_split_id, node_id: pipeline_id.node_id.clone(), index_uid: pipeline_id.index_uid.clone(), source_id: pipeline_id.source_id.clone(), doc_mapping_uid, - split_id: merge_split_id, + sort_fields, partition_id, replaced_split_ids, time_range, @@ -351,6 +422,7 @@ impl MergeExecutor { ) -> anyhow::Result { let (union_index_meta, split_directories) = open_split_directories( &tantivy_dirs, + &splits, self.doc_mapper.tokenizer_manager().tantivy_manager(), )?; // TODO it would be nice if tantivy could let us run the merge in the current thread. @@ -419,6 +491,7 @@ impl MergeExecutor { let (union_index_meta, split_directories) = open_split_directories( &tantivy_dirs, + std::slice::from_ref(&split), self.doc_mapper.tokenizer_manager().tantivy_manager(), )?; let controlled_directory = self @@ -484,11 +557,12 @@ impl MergeExecutor { }; let indexed_split = IndexedSplit { split_attrs: SplitAttrs { + split_id: merge_split_id, node_id: NodeId::from_str(&split.node_id), index_uid: split.index_uid, source_id: split.source_id, doc_mapping_uid: split.doc_mapping_uid, - split_id: merge_split_id, + sort_fields: split.sort_fields, partition_id: split.partition_id, replaced_split_ids: vec![split.split_id.clone()], time_range, @@ -952,3 +1026,346 @@ mod tests { .await } } + +#[cfg(test)] +mod sort_tests { + use quickwit_common::split_file; + use quickwit_metastore::{ + IndexingSortSchema, ListSplitsRequestExt, MetastoreServiceStreamSplitsExt, + SortFieldMetadata, + }; + use quickwit_proto::metastore::{DeleteQuery, ListSplitsRequest}; + use quickwit_proto::search::{SortFieldType, SortOrder}; + use serde_json::json; + use tantivy::collector::{Count, DocSetCollector}; + use tantivy::query::{PhraseQuery, TermQuery}; + use tantivy::schema::{IndexRecordOption, Value}; + use tantivy::{DocAddress, TantivyDocument, Term}; + + use super::*; + use crate::merge_policy::MergeOperation; + use crate::{TestSandbox, get_tantivy_directory_from_split_bundle}; + + const MAPPING: &str = r#" + field_mappings: + - name: service + type: text + tokenizer: raw + fast: true + - name: id + type: u64 + fast: true + - name: body + type: text + record: position + tag_fields: [service] + "#; + + fn sort_fields(order: SortOrder) -> IndexingSortSchema { + IndexingSortSchema { + fields: vec![SortFieldMetadata { + field: "service".to_string(), + order, + field_type: SortFieldType::Text, + }], + } + } + + #[test] + fn test_sort_fields_reject_incompatible_schemas() -> anyhow::Result<()> { + let mut text_schema = tantivy::schema::Schema::builder(); + text_schema.add_text_field("service", tantivy::schema::FAST); + let mut int_schema = tantivy::schema::Schema::builder(); + int_schema.add_i64_field("service", tantivy::schema::FAST); + let settings = tantivy::IndexSettings { + sort_by_field: Some(tantivy::IndexSortByField { + field: "service".to_string(), + order: tantivy::Order::Asc, + }), + ..Default::default() + }; + let mut index_metas = Vec::new(); + for schema in [text_schema.build(), int_schema.build()] { + let index = Index::builder() + .schema(schema) + .settings(settings.clone()) + .create_in_ram()?; + index_metas.push(index.load_metas()?); + } + let error = combine_index_meta(index_metas).unwrap_err(); + assert!(error.to_string().contains("different physical sort fields")); + Ok(()) + } + + /// Check doc-ID order and verify stored fields, fast fields, postings and positions + /// still refer to the same documents after the permutation. + fn check_index(index: &Index, order: SortOrder) -> anyhow::Result> { + let reader = index.reader()?; + let searcher = reader.searcher(); + assert_eq!(searcher.segment_readers().len(), 1); + let segment = searcher.segment_reader(0); + let service_field = index.schema().get_field("service")?; + let id_field = index.schema().get_field("id")?; + let body_field = index.schema().get_field("body")?; + let service_column = segment.fast_fields().str("service")?.unwrap(); + let id_column = segment.fast_fields().u64("id")?; + let mut keys = Vec::new(); + let mut ids = Vec::new(); + for doc_id in 0..segment.max_doc() { + let document: TantivyDocument = searcher.doc(DocAddress::new(0, doc_id))?; + let service = document + .get_first(service_field) + .and_then(|value| value.as_str()); + let id = document.get_first(id_field).unwrap().as_u64().unwrap(); + assert_eq!(id_column.first(doc_id), Some(id)); + let fast_service = match service_column.term_ords(doc_id).next() { + Some(ordinal) => { + let mut value = String::new(); + service_column.ord_to_str(ordinal, &mut value)?; + Some(value) + } + None => None, + }; + assert_eq!(fast_service.as_deref(), service); + let expected_service = match id { + 0 => Some("zulu"), + 2 | 5 => Some("api"), + 3 => Some(""), + 4 => Some("équipe"), + 1 | 6 | 7 => None, + _ => panic!("unexpected id {id}"), + }; + assert_eq!(service, expected_service); + let phrase = PhraseQuery::new(vec![ + Term::from_field_text(body_field, "payload"), + Term::from_field_text(body_field, &format!("item{id}")), + ]); + let matches = searcher.search(&phrase, &DocSetCollector)?; + assert_eq!(matches.len(), 1); + assert!(matches.contains(&DocAddress::new(0, doc_id))); + keys.push(fast_service); + ids.push(id); + } + assert!(keys.windows(2).all(|pair| match order { + SortOrder::Asc => pair[0] <= pair[1], + SortOrder::Desc => pair[0] >= pair[1], + })); + let query = TermQuery::new( + Term::from_field_text(service_field, "api"), + IndexRecordOption::Basic, + ); + assert_eq!( + searcher.search(&query, &Count)?, + ids.iter().filter(|&&id| id == 2 || id == 5).count() + ); + Ok(ids) + } + + async fn execute( + sandbox: &TestSandbox, + splits: Vec, + delete: bool, + ) -> anyhow::Result { + let scratch = TempDirectory::for_test(); + let downloads = scratch.named_temp_child("downloads-")?; + let mut directories = Vec::new(); + for split in &splits { + let filename = split_file(split.split_id()); + let path = downloads.path().join(&filename); + sandbox + .storage() + .copy_to_file(Path::new(&filename), &path) + .await?; + directories.push(get_tantivy_directory_from_split_bundle(&path)?); + } + for (directory, split) in directories.iter().zip(&splits) { + let index = open_index( + directory.clone(), + sandbox.doc_mapper().tokenizer_manager().tantivy_manager(), + )?; + check_index(&index, split.sort_fields.fields[0].order)?; + assert_eq!(split.num_merge_ops, 0); + } + // A stale or missing declaration must not silently change merge semantics. + let mut wrong_type = splits[0].sort_fields.clone(); + wrong_type.fields[0].field_type = SortFieldType::I64; + for invalid_sort_fields in [IndexingSortSchema::default(), wrong_type] { + let mut stale_metadata = splits.clone(); + stale_metadata[0].sort_fields = invalid_sort_fields; + let error = open_split_directories( + &directories, + &stale_metadata, + sandbox.doc_mapper().tokenizer_manager().tantivy_manager(), + ) + .err() + .unwrap(); + assert!(error.to_string().contains("disagrees with metastore")); + } + + let operation = if delete { + MergeOperation::new_delete_and_merge_operation(splits[0].clone()) + } else { + MergeOperation::new_merge_operation(splits) + }; + let (mailbox, inbox) = sandbox.universe().create_test_mailbox(); + let executor = MergeExecutor::new( + MergePipelineId { + node_id: sandbox.node_id(), + index_uid: sandbox.index_uid(), + source_id: sandbox.source_id(), + }, + sandbox.metastore(), + sandbox.doc_mapper(), + IoControls::default(), + mailbox, + None, + ); + let (executor_mailbox, handle) = sandbox.universe().spawn_builder().spawn(executor); + executor_mailbox + .send_message(MergeScratch { + merge_source: MergeSource::Operation(operation), + tantivy_dirs: directories, + merge_scratch_directory: scratch, + downloaded_splits_directory: downloads, + }) + .await?; + handle.process_pending_and_observe().await; + let mut batches = inbox.drain_for_test_typed::(); + assert_eq!(batches.len(), 1); + Ok(batches.pop().unwrap().splits.pop().unwrap()) + } + + #[tokio::test] + async fn test_sort_fields_generation_zero_and_merge() -> anyhow::Result<()> { + for order in [SortOrder::Asc, SortOrder::Desc] { + let sort_field = match order { + SortOrder::Asc => "service", + SortOrder::Desc => "-service", + }; + let settings = serde_json::to_string(&json!({"sort_fields": [sort_field]}))?; + let sandbox = + TestSandbox::create("service-sort", MAPPING, &settings, &["body"]).await?; + for batch in [ + vec![ + (0, Some("zulu")), + (1, None), + (2, Some("api")), + (3, Some("")), + ], + vec![(4, Some("équipe")), (5, Some("api")), (6, None)], + vec![(7, None)], + ] { + sandbox + .add_documents(batch.into_iter().map(|(id, service)| { + json!({ + "id": id, "service": service, "body": format!("payload item{id}") + }) + })) + .await?; + } + let splits = sandbox + .metastore() + .list_splits(ListSplitsRequest::try_from_index_uid(sandbox.index_uid())?) + .await? + .collect_splits_metadata() + .await?; + assert_eq!(splits.len(), 3); + assert!( + splits + .iter() + .all(|split| split.sort_fields == sort_fields(order)) + ); + let merged = execute(&sandbox, splits, false).await?; + assert_eq!(merged.split_attrs.sort_fields, sort_fields(order)); + assert_eq!(merged.split_attrs.num_docs, 8); + let mut ids = check_index(&merged.index, order)?; + ids.sort_unstable(); + assert_eq!(ids, (0..8).collect::>()); + drop(merged); + sandbox.assert_quit().await; + } + Ok(()) + } + + #[tokio::test] + async fn test_sort_fields_preserved_after_delete() -> anyhow::Result<()> { + let sandbox = TestSandbox::create( + "service-sort-delete", + MAPPING, + "sort_fields: [service]", + &["body"], + ) + .await?; + sandbox + .add_documents(vec![ + json!({"id": 0, "service": "zulu", "body": "payload item0"}), + json!({"id": 1, "body": "payload item1"}), + json!({"id": 2, "service": "api", "body": "payload item2"}), + ]) + .await?; + let splits = sandbox + .metastore() + .list_splits(ListSplitsRequest::try_from_index_uid(sandbox.index_uid())?) + .await? + .collect_splits_metadata() + .await?; + sandbox + .metastore() + .create_delete_task(DeleteQuery { + index_uid: Some(sandbox.index_uid()), + start_timestamp: None, + end_timestamp: None, + query_ast: quickwit_query::query_ast::qast_json_helper("service:api", &["body"]), + }) + .await?; + let rewritten = execute(&sandbox, splits, true).await?; + assert_eq!( + rewritten.split_attrs.sort_fields, + sort_fields(SortOrder::Asc) + ); + assert_eq!(rewritten.split_attrs.num_docs, 2); + assert_eq!(rewritten.split_attrs.delete_opstamp, 1); + assert_eq!(check_index(&rewritten.index, SortOrder::Asc)?, vec![1, 0]); + drop(rewritten); + sandbox.assert_quit().await; + Ok(()) + } + + #[test] + fn test_sort_fields_reject_incompatible_merge_inputs() { + let mut first = SplitMetadata { + sort_fields: sort_fields(SortOrder::Asc), + ..Default::default() + }; + let mut second = first.clone(); + second.sort_fields = IndexingSortSchema::default(); + let pipeline_id = MergePipelineId { + node_id: NodeId::from_str("test"), + index_uid: first.index_uid.clone(), + source_id: "test".to_string(), + }; + merge_split_attrs( + pipeline_id.clone(), + SplitId::new(), + &[first.clone(), second.clone()], + ) + .unwrap_err(); + second.sort_fields = sort_fields(SortOrder::Desc); + merge_split_attrs( + pipeline_id.clone(), + SplitId::new(), + &[first.clone(), second.clone()], + ) + .unwrap_err(); + second.sort_fields = first.sort_fields.clone(); + second.sort_fields.fields[0].field_type = SortFieldType::I64; + merge_split_attrs( + pipeline_id.clone(), + SplitId::new(), + &[first.clone(), second.clone()], + ) + .unwrap_err(); + first.sort_fields = second.sort_fields.clone(); + merge_split_attrs(pipeline_id, SplitId::new(), &[first, second]).unwrap(); + } +} diff --git a/quickwit/quickwit-indexing/src/actors/merge_planner.rs b/quickwit/quickwit-indexing/src/actors/merge_planner.rs index 7182545d8b0..88862370ebd 100644 --- a/quickwit/quickwit-indexing/src/actors/merge_planner.rs +++ b/quickwit/quickwit-indexing/src/actors/merge_planner.rs @@ -38,6 +38,7 @@ pub(crate) struct RunFinalizeMergePolicyAndQuit; struct MergePartition { partition_id: u64, doc_mapping_uid: DocMappingUid, + sort_fields: quickwit_metastore::IndexingSortSchema, } impl MergePartition { @@ -45,6 +46,7 @@ impl MergePartition { MergePartition { partition_id: split_meta.partition_id, doc_mapping_uid: split_meta.doc_mapping_uid, + sort_fields: split_meta.sort_fields.clone(), } } } @@ -405,6 +407,77 @@ mod tests { } } + #[tokio::test] + async fn test_sort_fields_separate_legacy_merge_groups() -> anyhow::Result<()> { + use quickwit_metastore::SortFieldMetadata; + use quickwit_proto::search::{SortFieldType, SortOrder}; + + let universe = Universe::with_accelerated_time(); + let index_uid = IndexUid::new_with_random_ulid("test-index"); + let pipeline_id = MergePipelineId { + node_id: NodeId::from_str("test-node"), + index_uid: index_uid.clone(), + source_id: "test-source".to_string(), + }; + let settings = IndexingSettings { + merge_policy: MergePolicyConfig::ConstWriteAmplification( + ConstWriteAmplificationMergePolicyConfig { + merge_factor: 2, + max_merge_factor: 2, + ..Default::default() + }, + ), + ..Default::default() + }; + let mut splits = Vec::new(); + for (group, sort) in [ + None, + Some((SortOrder::Asc, SortFieldType::Text)), + Some((SortOrder::Desc, SortFieldType::Text)), + Some((SortOrder::Asc, SortFieldType::I64)), + ] + .into_iter() + .enumerate() + { + for i in 0..2 { + let mut split = split_metadata_for_test( + &index_uid, + &format!("split-{group}-{i}"), + 0, + DocMappingUid::default(), + 1000, + 0, + ); + split.sort_fields.fields = sort + .map(|(order, field_type)| SortFieldMetadata { + field: "service".to_string(), + order, + field_type, + }) + .into_iter() + .collect(); + splits.push(split); + } + } + let (downloader_mailbox, downloader_inbox) = universe.create_test_mailbox(); + let planner = MergePlanner::new( + &pipeline_id, + splits, + merge_policy_from_settings(&settings), + downloader_mailbox, + universe.get_or_spawn_one(), + ); + let (_mailbox, _handle) = universe.spawn_builder().spawn(planner); + for _ in 0..4 { + let source = downloader_inbox.recv_typed_message::().await?; + let splits = &source.as_operation().splits; + assert_eq!(splits.len(), 2); + assert_eq!(splits[0].sort_fields, splits[1].sort_fields); + } + universe.assert_quit().await; + Ok(()) + } + #[tokio::test] async fn test_merge_planner_with_stable_custom_merge_policy() -> anyhow::Result<()> { let node_id = NodeId::from_str("test-node"); diff --git a/quickwit/quickwit-indexing/src/actors/packager.rs b/quickwit/quickwit-indexing/src/actors/packager.rs index 821c90a7968..dd901abf225 100644 --- a/quickwit/quickwit-indexing/src/actors/packager.rs +++ b/quickwit/quickwit-indexing/src/actors/packager.rs @@ -377,7 +377,7 @@ mod tests { use quickwit_common::io::IoControls; use quickwit_metastore::checkpoint::IndexCheckpointDelta; use quickwit_proto::search::{ListFieldsEntry, ListFieldsMetadata}; - use quickwit_proto::types::{DocMappingUid, IndexUid, NodeId}; + use quickwit_proto::types::{DocMappingUid, IndexUid, NodeId, SplitId}; use tantivy::directory::MmapDirectory; use tantivy::schema::{FAST, NumericOptions, STRING, Schema, TEXT, Type}; use tantivy::{DateTime, IndexBuilder, IndexSettings, doc}; @@ -514,11 +514,12 @@ mod tests { // but this will require work on tantivy. let indexed_split = IndexedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from("test-split"), node_id, index_uid, source_id, doc_mapping_uid: DocMappingUid::default(), - split_id: "test-split".into(), + sort_fields: Default::default(), partition_id: 17u64, num_docs, uncompressed_docs_size_in_bytes: num_docs * 15, diff --git a/quickwit/quickwit-indexing/src/actors/uploader.rs b/quickwit/quickwit-indexing/src/actors/uploader.rs index 9ee80f35610..a5f642e2ede 100644 --- a/quickwit/quickwit-indexing/src/actors/uploader.rs +++ b/quickwit/quickwit-indexing/src/actors/uploader.rs @@ -461,6 +461,12 @@ fn create_split_recovery_metadata( num_merge_ops: split_metadata.num_merge_ops as u64, parent_split_ids, maturation_period_millis, + sort_fields: split_metadata + .sort_fields + .fields + .iter() + .map(Into::into) + .collect(), } } @@ -624,6 +630,33 @@ mod tests { assert_eq!(recovered_mature_metadata.maturity, SplitMaturity::Mature); } + #[test] + fn test_split_recovery_metadata_preserves_sort_fields() { + let split_metadata = SplitMetadata { + sort_fields: quickwit_metastore::IndexingSortSchema { + fields: vec![quickwit_metastore::SortFieldMetadata { + field: "service".to_string(), + order: quickwit_proto::search::SortOrder::Desc, + field_type: quickwit_proto::search::SortFieldType::Text, + }], + }, + ..Default::default() + }; + let bytes = create_split_recovery_metadata(&split_metadata, &[]).serialize(); + let recovery_metadata = SplitRecoveryMetadata::deserialize(&bytes).unwrap(); + let (recovered, _) = + SplitMetadata::try_from_recovery_metadata(recovery_metadata, 1..2).unwrap(); + assert_eq!(recovered.sort_fields, split_metadata.sort_fields); + + for invalid_type in [0, 99] { + let mut recovery_metadata = create_split_recovery_metadata(&split_metadata, &[]); + recovery_metadata.sort_fields[0].field_type = invalid_type; + let bytes = recovery_metadata.serialize(); + let recovery_metadata = SplitRecoveryMetadata::deserialize(&bytes).unwrap(); + SplitMetadata::try_from_recovery_metadata(recovery_metadata, 1..2).unwrap_err(); + } + } + #[tokio::test] async fn test_uploader_with_sequencer() -> anyhow::Result<()> { quickwit_common::setup_logging_for_tests(); @@ -673,10 +706,12 @@ mod tests { .send_message(PackagedSplitBatch::new( vec![PackagedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from("test-split"), node_id, index_uid, source_id, doc_mapping_uid: DocMappingUid::default(), + sort_fields: Default::default(), partition_id: 3u64, time_range: Some( DateTime::from_timestamp_secs(1_628_203_589) @@ -685,7 +720,6 @@ mod tests { uncompressed_docs_size_in_bytes: 1_000, num_docs: 10, replaced_split_ids: Vec::new(), - split_id: "test-split".into(), delete_opstamp: 10, num_merge_ops: 0, }, @@ -783,11 +817,12 @@ mod tests { let split_scratch_directory_2 = TempDirectory::for_test(); let packaged_split_1 = PackagedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from("test-split-1"), node_id: node_id.clone(), index_uid: index_uid.clone(), source_id: source_id.clone(), doc_mapping_uid: DocMappingUid::default(), - split_id: "test-split-1".into(), + sort_fields: Default::default(), partition_id: 3u64, num_docs: 10, uncompressed_docs_size_in_bytes: 1_000, @@ -810,11 +845,12 @@ mod tests { }; let package_split_2 = PackagedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from("test-split-2"), node_id, index_uid, source_id, doc_mapping_uid: DocMappingUid::default(), - split_id: "test-split-2".into(), + sort_fields: Default::default(), partition_id: 3u64, num_docs: 10, uncompressed_docs_size_in_bytes: 1_000, @@ -935,11 +971,12 @@ mod tests { .send_message(PackagedSplitBatch::new( vec![PackagedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from("test-split"), node_id, index_uid, source_id, doc_mapping_uid: DocMappingUid::default(), - split_id: "test-split".into(), + sort_fields: Default::default(), partition_id: 3u64, time_range: None, uncompressed_docs_size_in_bytes: 1_000, @@ -1112,10 +1149,12 @@ mod tests { .send_message(PackagedSplitBatch::new( vec![PackagedSplit { split_attrs: SplitAttrs { + split_id: SplitId::from(SPLIT_ULID_STR), node_id, index_uid, source_id, doc_mapping_uid: DocMappingUid::default(), + sort_fields: Default::default(), partition_id: 3u64, time_range: Some( DateTime::from_timestamp_secs(1_628_203_589) @@ -1124,7 +1163,6 @@ mod tests { uncompressed_docs_size_in_bytes: 1_000, num_docs: 10, replaced_split_ids: Vec::new(), - split_id: SPLIT_ULID_STR.into(), delete_opstamp: 10, num_merge_ops: 0, }, diff --git a/quickwit/quickwit-indexing/src/models/indexed_split.rs b/quickwit/quickwit-indexing/src/models/indexed_split.rs index 3175ec90612..24dae9c6982 100644 --- a/quickwit/quickwit-indexing/src/models/indexed_split.rs +++ b/quickwit/quickwit-indexing/src/models/indexed_split.rs @@ -18,6 +18,7 @@ use std::path::Path; use quickwit_common::io::IoControls; use quickwit_common::metrics::index_label; use quickwit_common::temp_dir::TempDirectory; +use quickwit_metastore::IndexingSortSchema; use quickwit_metastore::checkpoint::IndexCheckpointDelta; use quickwit_metrics::{GaugeGuard, label_values}; use quickwit_proto::indexing::IndexingPipelineId; @@ -78,12 +79,15 @@ impl fmt::Debug for IndexedSplitBuilder { } impl IndexedSplitBuilder { + /// The caller must configure `index_builder` with the same physical ordering as `sort_fields`. + /// The indexer validates and constructs both together when the pipeline is created. #[allow(clippy::too_many_arguments)] pub fn new_in_dir( pipeline_id: IndexingPipelineId, partition_id: u64, last_delete_opstamp: u64, doc_mapping_uid: DocMappingUid, + sort_fields: IndexingSortSchema, scratch_directory: TempDirectory, index_builder: IndexBuilder, io_controls: IoControls, @@ -112,12 +116,13 @@ impl IndexedSplitBuilder { index_builder.single_segment_index_writer(indexing_directory, 15_000_000)?; Ok(Self { split_attrs: SplitAttrs { + split_id, node_id: pipeline_id.node_id, index_uid: pipeline_id.index_uid, source_id: pipeline_id.source_id, doc_mapping_uid, + sort_fields, partition_id, - split_id, num_docs: 0, replaced_split_ids: Vec::new(), uncompressed_docs_size_in_bytes: 0, diff --git a/quickwit/quickwit-indexing/src/models/split_attrs.rs b/quickwit/quickwit-indexing/src/models/split_attrs.rs index 9c016b5722e..1300d9f0138 100644 --- a/quickwit/quickwit-indexing/src/models/split_attrs.rs +++ b/quickwit/quickwit-indexing/src/models/split_attrs.rs @@ -18,7 +18,7 @@ use std::ops::{Range, RangeInclusive}; use std::sync::Arc; use std::time::Duration; -use quickwit_metastore::{SplitMaturity, SplitMetadata}; +use quickwit_metastore::{IndexingSortSchema, SplitMaturity, SplitMetadata}; use quickwit_proto::types::{DocMappingUid, IndexUid, NodeId, SourceId, SplitId}; use tantivy::DateTime; use time::OffsetDateTime; @@ -26,6 +26,12 @@ use time::OffsetDateTime; use crate::merge_policy::MergePolicy; pub struct SplitAttrs { + /// Split ID. Joined with the index URI (/), this ID + /// should be enough to uniquely identify a split. + /// In reality, some information may be implicitly configured + /// in the storage resolver: for instance, the Amazon S3 region. + pub split_id: SplitId, + /// ID of the node that produced the split. pub node_id: NodeId, // Index UID to which the split belongs. @@ -36,11 +42,8 @@ pub struct SplitAttrs { /// Doc mapping UID used to produce this split. pub doc_mapping_uid: DocMappingUid, - /// Split ID. Joined with the index URI (/), this ID - /// should be enough to uniquely identify a split. - /// In reality, some information may be implicitly configured - /// in the storage resolver: for instance, the Amazon S3 region. - pub split_id: SplitId, + /// Physical ordering and logical comparison types. Empty for unsorted splits. + pub sort_fields: IndexingSortSchema, /// Partition to which the split belongs. /// @@ -82,6 +85,7 @@ impl fmt::Debug for SplitAttrs { ) .field("num_docs", &self.num_docs) .field("num_merge_ops", &self.num_merge_ops) + .field("sort_fields", &self.sort_fields) .finish() } } @@ -114,6 +118,7 @@ pub fn create_split_metadata( index_uid: split_attrs.index_uid.clone(), source_id: split_attrs.source_id.clone(), doc_mapping_uid: split_attrs.doc_mapping_uid, + sort_fields: split_attrs.sort_fields.clone(), split_id: split_attrs.split_id.clone(), partition_id: split_attrs.partition_id, num_docs: split_attrs.num_docs as usize, diff --git a/quickwit/quickwit-metastore/src/indexing_sort_schema.rs b/quickwit/quickwit-metastore/src/indexing_sort_schema.rs new file mode 100644 index 00000000000..4c201bd4de2 --- /dev/null +++ b/quickwit/quickwit-metastore/src/indexing_sort_schema.rs @@ -0,0 +1,34 @@ +// 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}; + +use crate::SortFieldMetadata; + +/// Ordered physical sort fields resolved from the writer's schema. +/// Merge compatibility requires the same field names, orders, types, and list order. +/// An empty schema represents an unsorted split, including legacy splits. +#[derive(Clone, Debug, Default, Eq, PartialEq, Hash, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(transparent)] +pub struct IndexingSortSchema { + /// Fields in comparison order. Serialized directly as the `sort_fields` array. + pub fields: Vec, +} + +impl IndexingSortSchema { + /// Returns whether no physical ordering is guaranteed. + pub fn is_empty(&self) -> bool { + self.fields.is_empty() + } +} diff --git a/quickwit/quickwit-metastore/src/lib.rs b/quickwit/quickwit-metastore/src/lib.rs index 31bec51a292..273e7cf2393 100644 --- a/quickwit/quickwit-metastore/src/lib.rs +++ b/quickwit/quickwit-metastore/src/lib.rs @@ -26,9 +26,11 @@ #[allow(missing_docs)] pub mod checkpoint; mod error; +mod indexing_sort_schema; mod metastore; mod metastore_factory; mod metastore_resolver; +mod sort_field_metadata; mod split_metadata; mod split_metadata_version; #[cfg(test)] @@ -37,6 +39,7 @@ pub(crate) mod tests; use std::ops::Range; pub use error::MetastoreResolverError; +pub use indexing_sort_schema::IndexingSortSchema; pub use metastore::control_plane_metastore::ControlPlaneMetastore; pub use metastore::file_backed::FileBackedMetastore; pub(crate) use metastore::index_metadata::serialize::{IndexMetadataV0_8, VersionedIndexMetadata}; @@ -56,6 +59,7 @@ pub use metastore_factory::{MetastoreFactory, MetastoreFactoryOptions, Unsupport pub use metastore_resolver::MetastoreResolver; use quickwit_common::is_disjoint; use quickwit_doc_mapper::tag_pruning::TagFilterAst; +pub use sort_field_metadata::SortFieldMetadata; pub use split_metadata::{Split, SplitInfo, SplitMaturity, SplitMetadata, SplitState}; pub(crate) use split_metadata_version::{SplitMetadataV0_8, VersionedSplitMetadata}; diff --git a/quickwit/quickwit-metastore/src/sort_field_metadata.rs b/quickwit/quickwit-metastore/src/sort_field_metadata.rs new file mode 100644 index 00000000000..a80c9506b9d --- /dev/null +++ b/quickwit/quickwit-metastore/src/sort_field_metadata.rs @@ -0,0 +1,112 @@ +// 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 quickwit_proto::metastore::SplitSortField; +use quickwit_proto::search::{SortFieldType, SortOrder}; +use serde::{Deserialize, Serialize}; + +/// One resolved field in an [`IndexingSortSchema`](crate::IndexingSortSchema). +/// Missing values sort first ascending and last descending. The type is required even when +/// every document is missing the field. +#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct SortFieldMetadata { + /// Mapped field path. + pub field: String, + /// Physical sort direction. + pub order: SortOrder, + /// Logical comparison type, taken from the schema used to write this split. + #[serde(rename = "type", deserialize_with = "deserialize_sort_field_type")] + pub field_type: SortFieldType, +} + +fn deserialize_sort_field_type<'de, D>(deserializer: D) -> Result +where D: serde::Deserializer<'de> { + let field_type = SortFieldType::deserialize(deserializer)?; + if field_type == SortFieldType::Unspecified { + return Err(serde::de::Error::custom( + "sort field type must be specified", + )); + } + Ok(field_type) +} + +impl TryFrom for SortFieldMetadata { + type Error = anyhow::Error; + + fn try_from(sort: SplitSortField) -> anyhow::Result { + let field_type = SortFieldType::try_from(sort.field_type)?; + anyhow::ensure!( + field_type != SortFieldType::Unspecified, + "missing recovery sort field type for `{}`", + sort.field + ); + Ok(Self { + field: sort.field, + order: if sort.descending { + SortOrder::Desc + } else { + SortOrder::Asc + }, + field_type, + }) + } +} + +impl From<&SortFieldMetadata> for SplitSortField { + fn from(sort: &SortFieldMetadata) -> Self { + Self { + field: sort.field.clone(), + descending: sort.order == SortOrder::Desc, + field_type: sort.field_type as i32, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_sort_field_metadata_roundtrip() { + for (field_type, json_type) in [ + (SortFieldType::Text, "text"), + (SortFieldType::I64, "i64"), + (SortFieldType::U64, "u64"), + (SortFieldType::F64, "f64"), + (SortFieldType::Datetime, "datetime"), + (SortFieldType::Bytes, "bytes"), + (SortFieldType::Bool, "bool"), + ] { + for order in [SortOrder::Asc, SortOrder::Desc] { + let metadata = SortFieldMetadata { + field: "service".to_string(), + order, + field_type, + }; + let json = serde_json::to_value(&metadata).unwrap(); + assert_eq!(json["type"], json_type); + assert_eq!( + serde_json::from_value::(json).unwrap(), + metadata + ); + let recovery_field = SplitSortField::from(&metadata); + assert_eq!( + SortFieldMetadata::try_from(recovery_field).unwrap(), + metadata + ); + } + } + } +} diff --git a/quickwit/quickwit-metastore/src/split_metadata.rs b/quickwit/quickwit-metastore/src/split_metadata.rs index 918ad5a28f8..e59f285de06 100644 --- a/quickwit/quickwit-metastore/src/split_metadata.rs +++ b/quickwit/quickwit-metastore/src/split_metadata.rs @@ -134,6 +134,12 @@ pub struct SplitMetadata { /// Doc mapping UID used when creating this split. This split may only be merged with other /// splits using the same doc mapping UID. pub doc_mapping_uid: DocMappingUid, + + /// Physical ordering, including each field's logical comparison type. + /// Missing values sort first ascending, last descending. + /// An empty list means no ordering is guaranteed, including for legacy splits. Merge inputs + /// must have identical declarations; the index configuration cannot change this claim. + pub sort_fields: crate::IndexingSortSchema, } impl fmt::Debug for SplitMetadata { @@ -178,6 +184,9 @@ impl fmt::Debug for SplitMetadata { debug_struct.field("footer_offsets", &self.footer_offsets); debug_struct.field("delete_opstamp", &self.delete_opstamp); debug_struct.field("num_merge_ops", &self.num_merge_ops); + if !self.sort_fields.is_empty() { + debug_struct.field("sort_fields", &self.sort_fields); + } debug_struct.finish() } } @@ -206,6 +215,7 @@ impl SplitMetadata { num_merge_ops, parent_split_ids, maturation_period_millis, + sort_fields, } = recovery_metadata; let time_range = match (time_range_start_inclusive, time_range_end_inclusive) { (Some(start), Some(end)) if start <= end => Some(start..=end), @@ -242,6 +252,12 @@ impl SplitMetadata { num_merge_ops: num_merge_ops.try_into()?, doc_mapping_uid: doc_mapping_uid .ok_or_else(|| anyhow::anyhow!("missing recovery doc mapping UID"))?, + sort_fields: crate::IndexingSortSchema { + fields: sort_fields + .into_iter() + .map(TryInto::try_into) + .collect::>()?, + }, }; let parent_split_ids = parent_split_ids.into_iter().map(SplitId::from).collect(); Ok((split_metadata, parent_split_ids)) @@ -359,6 +375,7 @@ impl quickwit_config::TestableForRegression for SplitMetadata { footer_offsets: 1000..2000, num_merge_ops: 3, doc_mapping_uid: DocMappingUid::default(), + sort_fields: Default::default(), } } @@ -499,6 +516,7 @@ mod tests { delete_opstamp: 0, num_merge_ops: 0, doc_mapping_uid: DocMappingUid::default(), + sort_fields: Default::default(), }; let expected_output = "SplitMetadata { split_id: \"split-1\", index_uid: IndexUid { \ @@ -513,6 +531,50 @@ mod tests { assert_eq!(format!("{split_metadata:?}"), expected_output); } + #[test] + fn test_split_sort_fields_serde_compatibility() { + let legacy = SplitMetadata::default(); + let json = serde_json::to_value(&legacy).unwrap(); + assert!(json.get("sort_fields").is_none()); + assert!( + serde_json::from_value::(json) + .unwrap() + .sort_fields + .is_empty() + ); + + let sorted = SplitMetadata { + sort_fields: crate::IndexingSortSchema { + fields: vec![crate::SortFieldMetadata { + field: "service".to_string(), + order: quickwit_proto::search::SortOrder::Asc, + field_type: quickwit_proto::search::SortFieldType::Text, + }], + }, + ..Default::default() + }; + let json = serde_json::to_value(&sorted).unwrap(); + assert_eq!( + json["sort_fields"], + serde_json::json!([{"field": "service", "order": "asc", "type": "text"}]) + ); + assert_eq!( + serde_json::from_value::(json.clone()).unwrap(), + sorted + ); + let mut missing_type = json.clone(); + missing_type["sort_fields"][0] + .as_object_mut() + .unwrap() + .remove("type"); + serde_json::from_value::(missing_type).unwrap_err(); + for invalid_type in ["unknown", "unspecified"] { + let mut invalid_metadata = json.clone(); + invalid_metadata["sort_fields"][0]["type"] = serde_json::json!(invalid_type); + serde_json::from_value::(invalid_metadata).unwrap_err(); + } + } + #[test] fn test_spit_maturity_order() { assert!( diff --git a/quickwit/quickwit-metastore/src/split_metadata_version.rs b/quickwit/quickwit-metastore/src/split_metadata_version.rs index 8325290be92..7cf4afd8952 100644 --- a/quickwit/quickwit-metastore/src/split_metadata_version.rs +++ b/quickwit/quickwit-metastore/src/split_metadata_version.rs @@ -92,6 +92,9 @@ pub(crate) struct SplitMetadataV0_8 { // splits before when updates first appeared are compatible with each other. #[serde(default)] doc_mapping_uid: DocMappingUid, + + #[serde(default, skip_serializing_if = "crate::IndexingSortSchema::is_empty")] + sort_fields: crate::IndexingSortSchema, } impl From for SplitMetadata { @@ -128,6 +131,7 @@ impl From for SplitMetadata { footer_offsets: v8.footer_offsets, num_merge_ops: v8.num_merge_ops, doc_mapping_uid: v8.doc_mapping_uid, + sort_fields: v8.sort_fields, } } } @@ -150,6 +154,7 @@ impl From for SplitMetadataV0_8 { footer_offsets: split.footer_offsets, num_merge_ops: split.num_merge_ops, doc_mapping_uid: split.doc_mapping_uid, + sort_fields: split.sort_fields, } } } diff --git a/quickwit/quickwit-proto/protos/quickwit/metastore.proto b/quickwit/quickwit-proto/protos/quickwit/metastore.proto index 2a8c9b40347..05f2e686f70 100644 --- a/quickwit/quickwit-proto/protos/quickwit/metastore.proto +++ b/quickwit/quickwit-proto/protos/quickwit/metastore.proto @@ -18,6 +18,7 @@ package quickwit.metastore; import "quickwit/common.proto"; import "quickwit/ingest.proto"; +import "quickwit/search.proto"; enum SourceType { SOURCE_TYPE_UNSPECIFIED = 0; @@ -62,6 +63,17 @@ message SplitRecoveryMetadata { // None means the split is mature; otherwise, this is its maturation period in milliseconds. optional uint64 maturation_period_millis = 16; + + // Physical ordering. Empty for legacy splits and splits without a sort guarantee. + repeated SplitSortField sort_fields = 17; +} + +// Physical ordering. Missing values sort first ascending, last descending. +message SplitSortField { + string field = 1; + bool descending = 2; + // Required for every declared sort field. Unspecified and unknown types are invalid. + quickwit.search.SortFieldType field_type = 3; } // Metastore meant to manage Quickwit's indexes, their splits and delete tasks. diff --git a/quickwit/quickwit-proto/protos/quickwit/search.proto b/quickwit/quickwit-proto/protos/quickwit/search.proto index a1a10efbe09..b935961329e 100644 --- a/quickwit/quickwit-proto/protos/quickwit/search.proto +++ b/quickwit/quickwit-proto/protos/quickwit/search.proto @@ -311,6 +311,19 @@ enum SortOrder { DESC = 1; //< This will be the default value; } +// Value type shared by query sorting and physical split sort schemas. +// A split's logical type comes from its writer schema, not its column encoding. +enum SortFieldType { + SORT_FIELD_TYPE_UNSPECIFIED = 0; + SORT_FIELD_TYPE_TEXT = 1; + SORT_FIELD_TYPE_I64 = 2; + SORT_FIELD_TYPE_U64 = 3; + SORT_FIELD_TYPE_F64 = 4; + SORT_FIELD_TYPE_DATETIME = 5; + SORT_FIELD_TYPE_BYTES = 6; + SORT_FIELD_TYPE_BOOL = 7; +} + // Sort value format for datetime field. // We keep an enum with only one format // for future extension. diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs index 9a44e8a166a..4ef137ae988 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs @@ -2,7 +2,7 @@ /// Immutable metadata embedded directly in a split so that the metastore split metadata can be /// reconstructed if the metastore database is lost. #[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, PartialEq, ::prost::Message)] pub struct SplitRecoveryMetadata { #[prost(string, tag = "1")] pub split_id: ::prost::alloc::string::String, @@ -38,6 +38,21 @@ pub struct SplitRecoveryMetadata { /// None means the split is mature; otherwise, this is its maturation period in milliseconds. #[prost(uint64, optional, tag = "16")] pub maturation_period_millis: ::core::option::Option, + /// Physical ordering. Empty for legacy splits and splits without a sort guarantee. + #[prost(message, repeated, tag = "17")] + pub sort_fields: ::prost::alloc::vec::Vec, +} +/// Physical ordering. Missing values sort first ascending, last descending. +#[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SplitSortField { + #[prost(string, tag = "1")] + pub field: ::prost::alloc::string::String, + #[prost(bool, tag = "2")] + pub descending: bool, + /// Required for every declared sort field. Unspecified and unknown types are invalid. + #[prost(enumeration = "super::search::SortFieldType", tag = "3")] + pub field_type: i32, } #[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs index ad0c1099cf7..cd8b0a5ebd5 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs @@ -895,6 +895,54 @@ impl SortOrder { } } } +/// Value type shared by query sorting and physical split sort schemas. +/// A split's logical type comes from its writer schema, not its column encoding. +#[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum SortFieldType { + Unspecified = 0, + Text = 1, + I64 = 2, + U64 = 3, + F64 = 4, + Datetime = 5, + Bytes = 6, + Bool = 7, +} +impl SortFieldType { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Unspecified => "SORT_FIELD_TYPE_UNSPECIFIED", + Self::Text => "SORT_FIELD_TYPE_TEXT", + Self::I64 => "SORT_FIELD_TYPE_I64", + Self::U64 => "SORT_FIELD_TYPE_U64", + Self::F64 => "SORT_FIELD_TYPE_F64", + Self::Datetime => "SORT_FIELD_TYPE_DATETIME", + Self::Bytes => "SORT_FIELD_TYPE_BYTES", + Self::Bool => "SORT_FIELD_TYPE_BOOL", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "SORT_FIELD_TYPE_UNSPECIFIED" => Some(Self::Unspecified), + "SORT_FIELD_TYPE_TEXT" => Some(Self::Text), + "SORT_FIELD_TYPE_I64" => Some(Self::I64), + "SORT_FIELD_TYPE_U64" => Some(Self::U64), + "SORT_FIELD_TYPE_F64" => Some(Self::F64), + "SORT_FIELD_TYPE_DATETIME" => Some(Self::Datetime), + "SORT_FIELD_TYPE_BYTES" => Some(Self::Bytes), + "SORT_FIELD_TYPE_BOOL" => Some(Self::Bool), + _ => None, + } + } +} /// Sort value format for datetime field. /// We keep an enum with only one format /// for future extension. diff --git a/quickwit/quickwit-proto/src/metastore/mod.rs b/quickwit/quickwit-proto/src/metastore/mod.rs index 34a23537520..9796f29eb1f 100644 --- a/quickwit/quickwit-proto/src/metastore/mod.rs +++ b/quickwit/quickwit-proto/src/metastore/mod.rs @@ -100,6 +100,11 @@ mod split_recovery_metadata_tests { num_merge_ops: 7, parent_split_ids: vec!["parent-a".to_string(), "parent-b".to_string()], maturation_period_millis: Some(40_500), + sort_fields: vec![super::SplitSortField { + field: "service".to_string(), + descending: false, + field_type: crate::search::SortFieldType::Text as i32, + }], }; let mut serialized = metadata.serialize(); // An unknown protobuf varint field must be ignored by older readers. diff --git a/quickwit/quickwit-search/src/collector.rs b/quickwit/quickwit-search/src/collector.rs index 08ed20a8486..da4a9bd83b4 100644 --- a/quickwit/quickwit-search/src/collector.rs +++ b/quickwit/quickwit-search/src/collector.rs @@ -20,8 +20,8 @@ use itertools::Itertools; use quickwit_common::binary_heap::{SortKeyMapper, TopK}; use quickwit_doc_mapper::{FastFieldWarmupInfo, WarmupInfo}; use quickwit_proto::search::{ - LeafResourceStats, LeafSearchResponse, PartialHit, SearchRequest, SortByValue, SortOrder, - SortValue, SplitSearchError, + LeafResourceStats, LeafSearchResponse, PartialHit, SearchRequest, SortByValue, SortFieldType, + SortOrder, SortValue, SplitSearchError, }; use quickwit_proto::types::SplitId; use serde::Deserialize; @@ -91,7 +91,18 @@ impl SortByComponent { ColumnType::U64, ) }); - let sort_field_type = SortFieldType::try_from(column_type)?; + let sort_field_type = match column_type { + ColumnType::U64 => SortFieldType::U64, + ColumnType::I64 => SortFieldType::I64, + ColumnType::F64 => SortFieldType::F64, + ColumnType::DateTime => SortFieldType::Datetime, + ColumnType::Bool => SortFieldType::Bool, + _ => { + return Err(TantivyError::InvalidArgument(format!( + "Unsupported sort field type `{column_type:?}`." + ))); + } + }; Ok(SortingFieldExtractorComponent::FastField { sort_column, sort_field_type, @@ -125,15 +136,6 @@ impl SortByComponent { } } -#[derive(Copy, Clone, Debug, Eq, PartialEq)] -pub(crate) enum SortFieldType { - U64, - I64, - F64, - DateTime, - Bool, -} - /// The `SortingFieldExtractor` is used to extract a score, which can either be a true score, /// a value from a fast field, or nothing (sort by DocId). pub(crate) enum SortingFieldExtractorComponent { @@ -141,6 +143,8 @@ pub(crate) enum SortingFieldExtractorComponent { DocId, FastField { sort_column: Column, + // Extractor construction accepts only numeric and boolean types. The shared enum + // also describes physical text/bytes sorting, which query sorting does not support. sort_field_type: SortFieldType, }, Score, @@ -192,8 +196,11 @@ impl SortingFieldExtractorComponent { SortFieldType::U64 => SortValue::U64(fast_field_value), SortFieldType::I64 => SortValue::I64(i64::from_u64(fast_field_value)), SortFieldType::F64 => SortValue::F64(f64::from_u64(fast_field_value)), - SortFieldType::DateTime => SortValue::I64(i64::from_u64(fast_field_value)), + SortFieldType::Datetime => SortValue::I64(i64::from_u64(fast_field_value)), SortFieldType::Bool => SortValue::Boolean(fast_field_value != 0u64), + SortFieldType::Text | SortFieldType::Bytes | SortFieldType::Unspecified => { + unreachable!("search sort extractors only accept numeric and boolean columns") + } }; match self { SortingFieldExtractorComponent::DocId => SortValue::U64(sort_value), @@ -274,7 +281,7 @@ impl SortingFieldExtractorComponent { (val as i64).to_u64() } (SortValue::U64(val), SortFieldType::F64) => (val as f64).to_u64(), - (SortValue::U64(mut val), SortFieldType::DateTime) => { + (SortValue::U64(mut val), SortFieldType::Datetime) => { // Match everything if sort_order == SortOrder::Desc && val > i64::MAX as u64 { return None; @@ -294,7 +301,7 @@ impl SortingFieldExtractorComponent { } } (SortValue::I64(val), SortFieldType::F64) => (val as f64).to_u64(), - (SortValue::I64(val), SortFieldType::DateTime) => { + (SortValue::I64(val), SortFieldType::Datetime) => { DateTime::from_timestamp_nanos(val).to_u64() } (SortValue::F64(val), SortFieldType::U64) => { @@ -309,7 +316,7 @@ impl SortingFieldExtractorComponent { (val as u64).to_u64() } (SortValue::F64(val), SortFieldType::I64) - | (SortValue::F64(val), SortFieldType::DateTime) => { + | (SortValue::F64(val), SortFieldType::Datetime) => { let all_values_ahead1 = val < i64::MIN as f64 && sort_order == SortOrder::Asc; let all_values_ahead2 = @@ -320,7 +327,7 @@ impl SortingFieldExtractorComponent { // f64 cast already handles under/overflow and clamps the value let val_i64 = val as i64; - if *sort_field_type == SortFieldType::DateTime { + if *sort_field_type == SortFieldType::Datetime { DateTime::from_timestamp_nanos(val_i64).to_u64() } else { val_i64.to_u64() @@ -330,7 +337,7 @@ impl SortingFieldExtractorComponent { (SortValue::Boolean(val), SortFieldType::U64) => val as u64, (SortValue::Boolean(val), SortFieldType::F64) => (val as u64 as f64).to_u64(), (SortValue::Boolean(val), SortFieldType::I64) => (val as i64).to_u64(), - (SortValue::Boolean(val), SortFieldType::DateTime) => { + (SortValue::Boolean(val), SortFieldType::Datetime) => { DateTime::from_timestamp_nanos(val as i64).to_u64() } (SortValue::U64(mut val), SortFieldType::Bool) => { @@ -361,6 +368,14 @@ impl SortingFieldExtractorComponent { val = val.clamp(0.0, 1.0); (val >= 0.5).to_u64() // Is this correct? } + ( + _, + SortFieldType::Text | SortFieldType::Bytes | SortFieldType::Unspecified, + ) => { + unreachable!( + "search sort extractors only accept numeric and boolean columns" + ) + } }; Some(val) } @@ -431,23 +446,6 @@ impl SortingFieldExtractorPair { } } -impl TryFrom for SortFieldType { - type Error = tantivy::TantivyError; - - fn try_from(column_type: ColumnType) -> tantivy::Result { - match column_type { - ColumnType::U64 => Ok(SortFieldType::U64), - ColumnType::I64 => Ok(SortFieldType::I64), - ColumnType::F64 => Ok(SortFieldType::F64), - ColumnType::DateTime => Ok(SortFieldType::DateTime), - ColumnType::Bool => Ok(SortFieldType::Bool), - _ => Err(TantivyError::InvalidArgument(format!( - "Unsupported sort field type `{column_type:?}`." - ))), - } - } -} - /// Takes a user-defined sorting criteria and resolves it to a /// segment specific `SortingFieldExtractorPair`. fn get_score_extractor( diff --git a/quickwit/quickwit-serve/src/search_api/rest_handler.rs b/quickwit/quickwit-serve/src/search_api/rest_handler.rs index 01ad87d78e9..476d31bdfe6 100644 --- a/quickwit/quickwit-serve/src/search_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/search_api/rest_handler.rs @@ -16,6 +16,7 @@ use std::convert::TryFrom; use std::sync::Arc; use percent_encoding::percent_decode_str; +use quickwit_common::parse_sort_fields; use quickwit_config::validate_index_id_pattern; use quickwit_proto::search::{CountHits, SortField, SortOrder}; use quickwit_query::query_ast::query_ast_from_user_text; @@ -92,28 +93,15 @@ impl SortBy { impl From for SortBy { fn from(sort_by: String) -> Self { - let mut sort_fields = Vec::new(); - - for field_name in sort_by.split(',') { - if field_name.is_empty() { - continue; - } - let (field_name, sort_order) = if let Some(tail) = field_name.strip_prefix('+') { - (tail.trim().to_string(), SortOrder::Desc) - } else if let Some(tail) = field_name.strip_prefix('-') { - (tail.trim().to_string(), SortOrder::Asc) - } else { - let trimmed_field_name = field_name.trim().to_string(); - - (trimmed_field_name, SortOrder::Desc) - }; - let sort_field = SortField { - field_name, + let fields = sort_by.split(',').filter(|field| !field.is_empty()); + let sort_fields = parse_sort_fields(fields, SortOrder::Desc, SortOrder::Asc) + .into_iter() + .map(|(field_name, sort_order)| SortField { + field_name: field_name.trim().to_string(), sort_order: sort_order as i32, sort_datetime_format: None, - }; - sort_fields.push(sort_field); - } + }) + .collect(); Self { sort_fields } } } @@ -819,6 +807,21 @@ mod tests { }, ], ), + ( + "%2B%20field1%20,,,-%20field2%20,", + vec![ + SortField { + field_name: "field1".to_string(), + sort_order: SortOrder::Desc as i32, + sort_datetime_format: None, + }, + SortField { + field_name: "field2".to_string(), + sort_order: SortOrder::Asc as i32, + sort_datetime_format: None, + }, + ], + ), ] { let path = format!( "/quickwit-demo-index/search?query=*&format=json&sort_by={sort_by_query_param}"