From 5f7e486da7905167fb7cd80b448bef71842171d9 Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Fri, 31 Jul 2026 14:40:39 +0200 Subject: [PATCH 1/6] Generalize document clustering fingerprints Support ordered fingerprint policies, recursive multi-level clustering, and JSON path exclusions so document clustering config is no longer limited to the fixed schema/grouping shape. Co-authored-by: Cursor --- .../quickwit-config/src/docs_clustering.rs | 240 ++++++------------ quickwit/quickwit-config/src/lib.rs | 2 +- quickwit/quickwit-indexing/Cargo.toml | 4 + .../benches/doc_id_clusterer_bench.rs | 168 ++++++++++++ .../src/actors/doc_processor.rs | 9 +- .../quickwit-indexing/src/actors/indexer.rs | 32 ++- .../src/docs_clustering/clusterer.rs | 177 ++++++++----- .../src/docs_clustering/fingerprinter.rs | 211 ++++++++------- .../src/docs_clustering/mod.rs | 13 +- 9 files changed, 508 insertions(+), 348 deletions(-) create mode 100644 quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs diff --git a/quickwit/quickwit-config/src/docs_clustering.rs b/quickwit/quickwit-config/src/docs_clustering.rs index cb4cdb2275f..6de0ede5fbf 100644 --- a/quickwit/quickwit-config/src/docs_clustering.rs +++ b/quickwit/quickwit-config/src/docs_clustering.rs @@ -13,9 +13,11 @@ // limitations under the License. use std::collections::HashMap; +use std::ops::Deref; use anyhow::ensure; -use serde::{Deserialize, Serialize}; +use serde::de::Error as SerdeError; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; use crate::config_value::ConfigValue; use crate::qw_env_vars::QW_DISABLE_DOCS_CLUSTERING; @@ -94,50 +96,6 @@ impl DocsClusteringConfig { for policy in &self.policies { policy.validate()?; } - // TODO: Remove this constraint once we support arbitrary levels of clustering in the - // runtime fingerprinter. - self.validate_fingerprinter_limitations()?; - Ok(()) - } - - // The runtime fingerprinter currently supports exactly two clustering levels. - // - // Implementation constraints are: - // - The first policy must contain exactly one structure field - // - The second policy may contain only raw and tokenized fields - // - Additional policies are not supported - // - // TODO: Remove this constraint once the runtime fingerprinter supports arbitrary clustering - // levels. - fn validate_fingerprinter_limitations(&self) -> anyhow::Result<()> { - ensure!( - self.policies.len() == 2, // one structure field and one or more other fields - "document clustering currently supports exactly two fingerprint policies" - ); - - // First policy must be a fingerprint policy with exactly one structure field - let ClusteringPolicy::Fingerprint { fingerprint } = &self.policies[0]; - ensure!( - fingerprint.fingerprint.len() == 1, // one fingerprinting policy - "first document clustering fingerprint policy must contain exactly one field" - ); - ensure!( - matches!( - fingerprint.fingerprint[0], - ClusteringField::Structure { .. } - ), - "first document clustering fingerprint policy must contain the structure field" - ); - - let ClusteringPolicy::Fingerprint { fingerprint } = &self.policies[1]; - let has_structure_field = fingerprint - .fingerprint - .iter() - .any(|field| matches!(field, ClusteringField::Structure { .. })); - ensure!( - !has_structure_field, - "document clustering fingerprint must contain exactly one structure field" - ); Ok(()) } } @@ -173,13 +131,61 @@ impl FingerprintPolicy { !self.fingerprint.is_empty(), "document clustering fingerprint policy must contain at least one field" ); - for field in &self.fingerprint { - field.validate()?; - } Ok(()) } } +#[repr(transparent)] +#[derive(Debug, Clone, PartialEq)] +pub struct JsonPath(Box<[String]>); + +impl Serialize for JsonPath { + fn serialize(&self, serializer: S) -> Result + where S: Serializer { + serializer.serialize_str(&self.0.join(".")) + } +} + +impl<'de> Deserialize<'de> for JsonPath { + fn deserialize(deserializer: D) -> Result + where D: Deserializer<'de> { + let path = String::deserialize(deserializer)?; + if path.is_empty() { + return Err(D::Error::custom( + "document clustering path must not be empty", + )); + } + + let json_path: Box<[String]> = path.split('.').map(ToString::to_string).collect(); + if json_path + .iter() + .any(|path_component| path_component.is_empty()) + { + return Err(D::Error::custom(format!( + "document clustering path `{path}` must not contain empty components" + ))); + } + if json_path + .iter() + .any(|path_component| path_component.trim() != path_component) + { + return Err(D::Error::custom(format!( + "document clustering path `{path}` must not contain leading or trailing whitespace" + ))); + } + + Ok(Self(json_path)) + } +} + +impl Deref for JsonPath { + type Target = [String]; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + /// A field used to partition documents. #[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] @@ -191,7 +197,7 @@ pub enum ClusteringField { Structure { /// Paths omitted from the structure fingerprint. #[serde(default, skip_serializing_if = "Vec::is_empty")] - exclude: Vec, + exclude: Vec, }, /// Groups documents by the exact string value at `path`. /// @@ -199,7 +205,7 @@ pub enum ClusteringField { /// when `path` is `service`. Raw { /// Dot-separated path to the string value. - path: String, + path: JsonPath, }, /// Groups documents by the pattern of the first 50 tokens in the string value at `path`. /// @@ -207,52 +213,15 @@ pub enum ClusteringField { /// group when `path` is `message`, because both values have the same token pattern. Tokenized { /// Dot-separated path to the string value. - path: String, + path: JsonPath, }, } -impl ClusteringField { - fn validate(&self) -> anyhow::Result<()> { - fn validate_json_path(path: &str) -> anyhow::Result<()> { - ensure!( - path.trim() == path, - "document clustering path `{path}` must not contain leading or trailing whitespace" - ); - ensure!( - !path.is_empty(), - "document clustering path must not be empty" - ); - ensure!( - !path.split('.').any(str::is_empty), - "document clustering path `{path}` must not contain empty components" - ); - Ok(()) - } - - match self { - Self::Structure { exclude } => { - for excluded_path in exclude { - validate_json_path(excluded_path)?; - } - } - Self::Raw { path } => { - validate_json_path(path)?; - } - Self::Tokenized { path } => { - validate_json_path(path)?; - } - } - Ok(()) - } -} - #[cfg(test)] mod tests { use std::collections::HashMap; - use super::{ - ClusteringField, ClusteringPolicy, DocsClusteringConfig, DocsClusteringConfigBuilder, - }; + use super::{DocsClusteringConfig, DocsClusteringConfigBuilder, JsonPath}; fn build_config(yaml: &str) -> anyhow::Result> { build_config_with_env(yaml, &HashMap::new()) @@ -266,45 +235,6 @@ mod tests { DocsClusteringConfigBuilder::build_optional(Some(config_builder), env_vars) } - #[test] - fn build_accepts_flattened_fingerprint_policy() { - let config = build_config( - r#" -- fingerprint: - - kind: structure - exclude: [custom] -- fingerprint: - - path: custom - kind: raw - - path: message - kind: tokenized -"#, - ) - .unwrap(); - - let config = config.unwrap(); - let ClusteringPolicy::Fingerprint { fingerprint } = &config.policies[0]; - let ClusteringField::Structure { - exclude: excluded_paths, - } = &fingerprint.fingerprint[0] - else { - panic!("expected structure field"); - }; - assert_eq!(excluded_paths, &["custom".to_string()]); - let ClusteringPolicy::Fingerprint { fingerprint } = &config.policies[1]; - assert_eq!( - fingerprint.fingerprint, - vec![ - ClusteringField::Raw { - path: "custom".to_string() - }, - ClusteringField::Tokenized { - path: "message".to_string() - }, - ] - ); - } - #[test] fn build_accepts_structure_without_exclusions() { let config = build_config( @@ -377,7 +307,7 @@ mod tests { } #[test] - fn config_validation_rejects_unsupported_fingerprinter_shapes() { + fn config_validation_rejects_invalid_fingerprint_policies() { let test_cases = [ ( serde_json::json!([ @@ -393,29 +323,6 @@ mod tests { ]), "fingerprint policy must contain at least one field", ), - ( - serde_json::json!([ - {"fingerprint": [{"path": "message", "kind": "raw"}]}, - {"fingerprint": [{"path": "service", "kind": "raw"}]} - ]), - "first document clustering fingerprint policy must contain the structure field", - ), - ( - serde_json::json!([ - {"fingerprint": [{"kind": "structure"}]}, - {"fingerprint": [{"kind": "structure"}]} - ]), - "exactly one structure field", - ), - ( - serde_json::json!([ - {"fingerprint": [{"kind": "structure"}]}, - {"fingerprint": [ - {"path": "message..template", "kind": "tokenized"} - ]} - ]), - "must not contain empty components", - ), ]; for (json_value, expected_error) in test_cases { @@ -429,19 +336,34 @@ mod tests { } #[test] - fn config_validation_rejects_wrong_policy_count() { - let config: DocsClusteringConfig = serde_json::from_value(serde_json::json!([ - {"fingerprint": [{"kind": "structure"}]} - ])) - .unwrap(); - let error = config.validate().unwrap_err(); + fn deserialization_rejects_invalid_json_path() { + let error = + serde_json::from_value::(serde_json::json!("message..template")).unwrap_err(); + assert!( error .to_string() - .contains("exactly two fingerprint policies") + .contains("must not contain empty components"), + "expected invalid json path failure, got: {error:?}" ); } + #[test] + fn config_validation_accepts_arbitrary_policy_shapes() { + let config: DocsClusteringConfig = serde_json::from_value(serde_json::json!([ + {"fingerprint": [{"path": "message", "kind": "raw"}]}, + {"fingerprint": [{"kind": "structure"}]}, + {"fingerprint": [{"kind": "structure"}]}, + {"fingerprint": [ + {"path": "service", "kind": "raw"}, + {"path": "message", "kind": "tokenized"} + ]} + ])) + .unwrap(); + + config.validate().unwrap(); + } + #[test] fn deserialization_rejects_malformed_yaml() { let error = serde_yaml::from_str::("- fingerprint: [") diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs index bc784c697fa..85195db4a44 100644 --- a/quickwit/quickwit-config/src/lib.rs +++ b/quickwit/quickwit-config/src/lib.rs @@ -42,7 +42,7 @@ mod templating; pub use cluster_config::ClusterConfig; pub use docs_clustering::{ - ClusteringField, ClusteringPolicy, DocsClusteringConfig, FingerprintPolicy, + ClusteringField, ClusteringPolicy, DocsClusteringConfig, FingerprintPolicy, JsonPath, }; // We export that one for backward compatibility. // See #2048 diff --git a/quickwit/quickwit-indexing/Cargo.toml b/quickwit/quickwit-indexing/Cargo.toml index 23d81a3e58c..2ed527afbed 100644 --- a/quickwit/quickwit-indexing/Cargo.toml +++ b/quickwit/quickwit-indexing/Cargo.toml @@ -147,6 +147,10 @@ required-features = ["fail/failpoints"] name = "doc_process_vrl_bench" harness = false +[[bench]] +name = "doc_id_clusterer_bench" +harness = false + [package.metadata.cargo-machete] # used to vendor/static build native dependencies ignored = ["libz-sys", "openssl"] diff --git a/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs b/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs new file mode 100644 index 00000000000..6b8a9ae207f --- /dev/null +++ b/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs @@ -0,0 +1,168 @@ +// 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 std::hint::black_box; + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use quickwit_config::DocsClusteringConfig; +use quickwit_indexing::docs_clustering::{DocIdClusterer, Fingerprint, Fingerprinter}; +use serde_json::{Value as JsonValue, json}; +use tantivy::DocId; + +const NUM_DOCS: usize = 100_000; + +fn docs_clustering_config(num_levels: usize) -> DocsClusteringConfig { + let mut policies = vec![json!({ + "fingerprint": [ + { + "kind": "structure", + "exclude": ["message", "timestamp"] + } + ] + })]; + if num_levels >= 2 { + policies.push(json!({ + "fingerprint": [ + { + "kind": "raw", + "path": "service" + } + ] + })); + } + if num_levels >= 3 { + policies.push(json!({ + "fingerprint": [ + { + "kind": "tokenized", + "path": "message" + } + ] + })); + } + if num_levels >= 4 { + policies.push(json!({ + "fingerprint": [ + { + "kind": "raw", + "path": "status" + } + ] + })); + } + + let config: DocsClusteringConfig = serde_json::from_value(JsonValue::Array(policies)).unwrap(); + config.validate().unwrap(); + config +} + +fn generate_fingerprints( + num_docs: usize, + num_levels: usize, + cardinality: usize, +) -> Vec { + let config = docs_clustering_config(num_levels); + let fingerprinter = Fingerprinter::new(&config); + (0..num_docs) + .map(|doc_idx| { + let service_id = doc_idx % cardinality; + let template_id = (doc_idx / cardinality) % 128; + let status = match doc_idx % 5 { + 0 => "debug", + 1 => "info", + 2 => "warn", + 3 => "error", + _ => "critical", + }; + let doc = json!({ + "timestamp": doc_idx, + "service": format!("service-{service_id}"), + "message": format!("template {template_id} request {} completed in {} ms", doc_idx, doc_idx % 1_000), + "status": status, + "host": format!("host-{}", doc_idx % 1_024), + }); + fingerprinter.fingerprint(&doc) + }) + .collect() +} + +fn is_unsorted_doc(doc_idx: usize, unsorted_doc_frequency_opt: Option) -> bool { + match unsorted_doc_frequency_opt { + Some(frequency) => doc_idx % frequency == 0, + None => false, + } +} + +fn bench_doc_id_mapping( + group: &mut criterion::BenchmarkGroup<'_, criterion::measurement::WallTime>, + name: &str, + fingerprints: &[Fingerprint], + unsorted_doc_frequency_opt: Option, +) { + group.throughput(criterion::Throughput::Elements(fingerprints.len() as u64)); + group.bench_function(BenchmarkId::new(name, fingerprints.len()), |b| { + b.iter(|| { + let mut clusterer = DocIdClusterer::default(); + for (doc_idx, fingerprint) in fingerprints.iter().enumerate() { + let fingerprint_opt = if is_unsorted_doc(doc_idx, unsorted_doc_frequency_opt) { + None + } else { + Some(fingerprint.clone()) + }; + clusterer.push(fingerprint_opt, doc_idx as DocId); + } + black_box( + clusterer + .into_doc_id_mapping(fingerprints.len() as u64) + .unwrap(), + ) + }) + }); +} + +fn bench_doc_id_clusterer(c: &mut Criterion) { + let low_cardinality_fingerprints = generate_fingerprints(NUM_DOCS, 2, 16); + let high_cardinality_fingerprints = generate_fingerprints(NUM_DOCS, 2, 16_384); + let deep_fingerprints = generate_fingerprints(NUM_DOCS, 4, 256); + + let mut group = c.benchmark_group("DocIdClusterer"); + bench_doc_id_mapping( + &mut group, + "two-level/low-cardinality", + &low_cardinality_fingerprints, + None, + ); + bench_doc_id_mapping( + &mut group, + "two-level/high-cardinality", + &high_cardinality_fingerprints, + None, + ); + bench_doc_id_mapping( + &mut group, + "four-level/mixed-cardinality", + &deep_fingerprints, + None, + ); + bench_doc_id_mapping( + &mut group, + "four-level/with-unsorted-docs", + &deep_fingerprints, + Some(10), + ); + group.finish(); +} + +criterion_group!(benches, bench_doc_id_clusterer); +criterion_main!(benches); diff --git a/quickwit/quickwit-indexing/src/actors/doc_processor.rs b/quickwit/quickwit-indexing/src/actors/doc_processor.rs index 41dec06c5c0..9c6294bea6d 100644 --- a/quickwit/quickwit-indexing/src/actors/doc_processor.rs +++ b/quickwit/quickwit-indexing/src/actors/doc_processor.rs @@ -654,6 +654,12 @@ mod tests { "path": "body", "kind": "raw" }] + }, + { + "fingerprint": [{ + "path": "body", + "kind": "tokenized" + }] } ])) .unwrap(); @@ -767,7 +773,8 @@ mod tests { let output_messages: Vec = indexer_inbox.drain_for_test_typed(); assert_eq!(output_messages.len(), 1); - assert!(output_messages[0].docs[0].fingerprint_opt.is_some()); + let fingerprint = output_messages[0].docs[0].fingerprint_opt.as_ref().unwrap(); + assert_eq!(fingerprint.len(), 3); universe.assert_quit().await; } diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index c8412ce9bdf..6ed248676ce 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -1251,6 +1251,12 @@ mod tests { "path": "body", "kind": "raw" }] + }, + { + "fingerprint": [{ + "path": "body", + "kind": "tokenized" + }] } ])) .unwrap(), @@ -1260,21 +1266,35 @@ mod tests { let docs = vec![ ProcessedDoc { doc: doc!(body_field=>"first"), - fingerprint_opt: Some(Fingerprint::for_test(1, 1)), + fingerprint_opt: Some(Fingerprint::for_test([1, 1, 1])), timestamp_opt: None, partition: 0, num_bytes: 5, }, ProcessedDoc { doc: doc!(body_field=>"second"), - fingerprint_opt: Some(Fingerprint::for_test(1, 2)), + fingerprint_opt: Some(Fingerprint::for_test([1, 2, 1])), timestamp_opt: None, partition: 0, num_bytes: 6, }, ProcessedDoc { doc: doc!(body_field=>"third"), - fingerprint_opt: Some(Fingerprint::for_test(1, 1)), + fingerprint_opt: Some(Fingerprint::for_test([1, 2, 1])), + timestamp_opt: None, + partition: 0, + num_bytes: 5, + }, + ProcessedDoc { + doc: doc!(body_field=>"fourth"), + fingerprint_opt: Some(Fingerprint::for_test([1, 1, 1])), + timestamp_opt: None, + partition: 0, + num_bytes: 6, + }, + ProcessedDoc { + doc: doc!(body_field=>"fifth"), + fingerprint_opt: Some(Fingerprint::for_test([1, 1, 2])), timestamp_opt: None, partition: 0, num_bytes: 5, @@ -1297,13 +1317,13 @@ mod tests { let clusterer = split_builder.doc_id_clusterer_opt.as_ref().unwrap(); let mut sort_group_sizes = clusterer.sort_group_sizes().collect_vec(); sort_group_sizes.sort_unstable(); - assert_eq!(sort_group_sizes, [1, 2]); + assert_eq!(sort_group_sizes, [1, 2, 2]); let indexed_split = split_builder.finalize()?; let reader = indexed_split.index.reader()?; let searcher = reader.searcher(); let mut bodies = Vec::new(); - for doc_id in 0..3 { + for doc_id in 0..5 { let doc: TantivyDocument = searcher.doc(DocAddress::new(0, doc_id))?; let body = doc .get_first(body_field) @@ -1311,7 +1331,7 @@ mod tests { .unwrap(); bodies.push(body.to_string()); } - assert_eq!(bodies, ["first", "third", "second"]); + assert_eq!(bodies, ["first", "fourth", "fifth", "second", "third"]); universe.assert_quit().await; Ok(()) } diff --git a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs index 82721e92b57..aed78e7ae65 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs @@ -21,11 +21,11 @@ //! //! ```text //! incoming docs in indexing order: -//! doc 0 -> schema A, grouping X -//! doc 1 -> schema A, grouping Y -//! doc 2 -> schema B, grouping X +//! doc 0 -> hashes [A, X] +//! doc 1 -> hashes [A, Y] +//! doc 2 -> hashes [B, X] //! doc 3 -> no fingerprint -//! doc 4 -> schema A, grouping X +//! doc 4 -> hashes [A, X] //! //! sort groups: //! A: @@ -39,9 +39,8 @@ //! [0, 4, 1, 2, 3] //! ``` //! -//! Schema groups are emitted largest first, and grouping sort groups within each schema are also -//! emitted largest first. Documents without a fingerprint are kept in insertion order after the -//! fingerprinted sort groups. +//! Groups are emitted largest first at every fingerprint level. Documents without a fingerprint are +//! kept in insertion order after the fingerprinted sort groups. use std::cmp::Reverse; use std::mem; @@ -64,40 +63,32 @@ const _: () = assert!(mem::size_of::() == mem::size_of::, - unsorted_docs: ClusterDocIds, + root: ClusterGroup, + unclustered_docs: ClusterDocIds, } +/// A node in the fingerprint prefix tree. +/// +/// Each level of the tree partitions documents by one fingerprint hash. Internal nodes hold their +/// children keyed by the hash at their level; leaf nodes (documents that share the exact same +/// fingerprint) hold the corresponding doc IDs in insertion order. #[derive(Default)] -struct SchemaCluster { +struct ClusterGroup { num_docs: usize, - docs_by_grouping_fingerprint: FnvHashMap, + doc_ids: ClusterDocIds, + children: FnvHashMap, } impl DocIdClusterer { pub fn push(&mut self, fingerprint_opt: Option, doc_id: DocId) { match fingerprint_opt { - Some(fingerprint) => { - let schema_group = self - .docs_by_schema_fingerprint - .entry(fingerprint.schema) - .or_default(); - schema_group - .docs_by_grouping_fingerprint - .entry(fingerprint.grouping) - .or_default() - .push(doc_id); - schema_group.num_docs += 1; - } - None => self.unsorted_docs.push(doc_id), + Some(fingerprint) => self.root.push(&fingerprint, doc_id), + None => self.unclustered_docs.push(doc_id), } } pub fn sort_group_sizes(&self) -> impl Iterator + '_ { - self.docs_by_schema_fingerprint - .values() - .flat_map(|schema_group| schema_group.docs_by_grouping_fingerprint.values()) - .map(ClusterDocIds::len) + self.root.leaf_sizes() } pub fn into_doc_id_mapping(self, num_docs: u64) -> anyhow::Result { @@ -108,22 +99,47 @@ impl DocIdClusterer { } fn into_sorted_doc_ids(self) -> Vec { - let mut schema_groups: Vec = - self.docs_by_schema_fingerprint.into_values().collect(); - schema_groups.sort_unstable_by_key(|schema_group| Reverse(schema_group.num_docs)); - - schema_groups - .into_iter() - .flat_map(|schema_group| { - let mut sort_groups: Vec = schema_group - .docs_by_grouping_fingerprint - .into_values() - .collect(); - sort_groups.sort_unstable_by_key(|sort_group| Reverse(sort_group.len())); - sort_groups.into_iter().flatten() - }) - .chain(self.unsorted_docs) - .collect() + let mut doc_ids = Vec::with_capacity(self.root.num_docs + self.unclustered_docs.len()); + self.root.append_sorted_doc_ids(&mut doc_ids); + doc_ids.extend(self.unclustered_docs); + doc_ids + } +} + +impl ClusterGroup { + fn push(&mut self, fingerprint: &[u64], doc_id: DocId) { + self.num_docs += 1; + match fingerprint.split_first() { + Some((fingerprint_hash, rest)) => { + self.children + .entry(*fingerprint_hash) + .or_default() + .push(rest, doc_id); + } + None => self.doc_ids.push(doc_id), + } + } + + fn append_sorted_doc_ids(self, doc_ids: &mut Vec) { + doc_ids.extend(self.doc_ids); + let mut children: Vec = self.children.into_values().collect(); + // Larger groups are emitted first. + children.sort_unstable_by_key(|child| Reverse(child.num_docs)); + for child in children { + child.append_sorted_doc_ids(doc_ids); + } + } + + /// Iterates over the leaf sort group sizes of this subtree. A leaf reports its document count, + /// while an empty tree reports nothing. + /// + /// The return type is boxed to break the otherwise self-referential (and thus infinitely sized) + /// recursive `impl Iterator` type. + fn leaf_sizes(&self) -> Box + '_> { + let leaf_size = (!self.doc_ids.is_empty()) + .then_some(self.doc_ids.len()) + .into_iter(); + Box::new(leaf_size.chain(self.children.values().flat_map(ClusterGroup::leaf_sizes))) } } @@ -131,21 +147,18 @@ impl DocIdClusterer { mod tests { use super::{DocIdClusterer, Fingerprint}; - fn fingerprint(schema_fingerprint: u64, grouping_fingerprint: u64) -> Fingerprint { - Fingerprint { - schema: schema_fingerprint, - grouping: grouping_fingerprint, - } + fn fingerprint(hashes: [u64; N]) -> Fingerprint { + Fingerprint::for_test(hashes) } #[test] fn emits_largest_schema_groups_first() { let mut clusterer = DocIdClusterer::default(); - clusterer.push(Some(fingerprint(1, 1)), 0); - clusterer.push(Some(fingerprint(2, 1)), 1); - clusterer.push(Some(fingerprint(1, 2)), 2); - clusterer.push(Some(fingerprint(2, 1)), 3); - clusterer.push(Some(fingerprint(1, 1)), 4); + clusterer.push(Some(fingerprint([1, 1])), 0); + clusterer.push(Some(fingerprint([2, 1])), 1); + clusterer.push(Some(fingerprint([1, 2])), 2); + clusterer.push(Some(fingerprint([2, 1])), 3); + clusterer.push(Some(fingerprint([1, 1])), 4); assert_eq!(clusterer.into_sorted_doc_ids(), [0, 4, 2, 1, 3]); } @@ -153,22 +166,50 @@ mod tests { #[test] fn emits_largest_sort_groups_first_within_schema() { let mut clusterer = DocIdClusterer::default(); - clusterer.push(Some(fingerprint(1, 1)), 0); - clusterer.push(Some(fingerprint(1, 2)), 1); - clusterer.push(Some(fingerprint(1, 2)), 2); - clusterer.push(Some(fingerprint(1, 1)), 3); - clusterer.push(Some(fingerprint(1, 2)), 4); + clusterer.push(Some(fingerprint([1, 1])), 0); + clusterer.push(Some(fingerprint([1, 2])), 1); + clusterer.push(Some(fingerprint([1, 2])), 2); + clusterer.push(Some(fingerprint([1, 1])), 3); + clusterer.push(Some(fingerprint([1, 2])), 4); assert_eq!(clusterer.into_sorted_doc_ids(), [1, 2, 4, 0, 3]); } + #[test] + fn emits_largest_groups_first_with_one_fingerprint_level() { + let mut clusterer = DocIdClusterer::default(); + clusterer.push(Some(fingerprint([1])), 0); + clusterer.push(Some(fingerprint([2])), 1); + clusterer.push(Some(fingerprint([1])), 2); + clusterer.push(Some(fingerprint([3])), 3); + clusterer.push(Some(fingerprint([3])), 4); + clusterer.push(Some(fingerprint([3])), 5); + + assert_eq!(clusterer.into_sorted_doc_ids(), [3, 4, 5, 0, 2, 1]); + } + + #[test] + fn emits_largest_groups_first_at_each_fingerprint_level() { + let mut clusterer = DocIdClusterer::default(); + clusterer.push(Some(fingerprint([1, 1, 1])), 0); + clusterer.push(Some(fingerprint([1, 1, 2])), 1); + clusterer.push(Some(fingerprint([1, 2, 1])), 2); + clusterer.push(Some(fingerprint([1, 2, 1])), 3); + clusterer.push(Some(fingerprint([2, 1, 1])), 4); + clusterer.push(Some(fingerprint([1, 2, 2])), 5); + clusterer.push(Some(fingerprint([1, 2, 1])), 6); + clusterer.push(Some(fingerprint([1, 1, 1])), 7); + + assert_eq!(clusterer.into_sorted_doc_ids(), [2, 3, 6, 5, 0, 7, 1, 4]); + } + #[test] fn appends_unsorted_docs_in_insertion_order() { let mut clusterer = DocIdClusterer::default(); clusterer.push(None, 0); - clusterer.push(Some(fingerprint(1, 1)), 1); + clusterer.push(Some(fingerprint([1, 1])), 1); clusterer.push(None, 2); - clusterer.push(Some(fingerprint(1, 1)), 3); + clusterer.push(Some(fingerprint([1, 1])), 3); clusterer.push(None, 4); assert_eq!(clusterer.into_sorted_doc_ids(), [1, 3, 0, 2, 4]); @@ -187,10 +228,10 @@ mod tests { #[test] fn preserves_split_doc_ids_across_batches() { let mut clusterer = DocIdClusterer::default(); - clusterer.push(Some(fingerprint(1, 1)), 0); - clusterer.push(Some(fingerprint(2, 1)), 1); + clusterer.push(Some(fingerprint([1, 1])), 0); + clusterer.push(Some(fingerprint([2, 1])), 1); // Simulate a second batch appended to the same split: doc IDs must keep increasing. - clusterer.push(Some(fingerprint(1, 1)), 2); + clusterer.push(Some(fingerprint([1, 1])), 2); clusterer.push(None, 3); assert_eq!(clusterer.into_sorted_doc_ids(), [0, 2, 1, 3]); @@ -199,10 +240,10 @@ mod tests { #[test] fn reports_leaf_grouping_sort_group_sizes() { let mut clusterer = DocIdClusterer::default(); - clusterer.push(Some(fingerprint(1, 1)), 0); - clusterer.push(Some(fingerprint(1, 2)), 1); - clusterer.push(Some(fingerprint(1, 2)), 2); - clusterer.push(Some(fingerprint(2, 1)), 3); + clusterer.push(Some(fingerprint([1, 1])), 0); + clusterer.push(Some(fingerprint([1, 2])), 1); + clusterer.push(Some(fingerprint([1, 2])), 2); + clusterer.push(Some(fingerprint([2, 1])), 3); let mut sort_group_sizes = clusterer.sort_group_sizes().collect::>(); sort_group_sizes.sort_unstable(); diff --git a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs index 5448e0bf3f0..eae29d62730 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs @@ -58,11 +58,15 @@ //! `service`. Missing fields and non-string values are encoded as absent so configured field //! positions remain distinct. use std::hash::Hasher; +use std::ops::Deref; use std::sync::Arc; use fnv::FnvHasher; -use quickwit_config::{ClusteringField, ClusteringPolicy, DocsClusteringConfig}; +use quickwit_config::{ + ClusteringField, ClusteringPolicy, DocsClusteringConfig, FingerprintPolicy, JsonPath, +}; use serde_json::Value as JsonValue; +use smallvec::SmallVec; use super::tokenize; @@ -74,72 +78,46 @@ const FIELD_BOUNDARY: u8 = 0xFD; const TOKENIZED_TOKEN_SEPARATOR: u8 = 0xFE; const MAX_GROUPING_TOKENS: usize = 50; -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub struct Fingerprint { - pub(super) schema: u64, - pub(super) grouping: u64, +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Fingerprint(SmallVec<[u64; 4]>); + +impl Deref for Fingerprint { + type Target = [u64]; + + fn deref(&self) -> &Self::Target { + &self.0 + } } #[cfg(test)] impl Fingerprint { - pub(crate) fn for_test(schema: u64, grouping: u64) -> Self { - Self { schema, grouping } + pub(crate) fn for_test(hashes: [u64; N]) -> Self { + let mut fingerprint = SmallVec::new(); + fingerprint.extend_from_slice(&hashes); + Self(fingerprint) } } -type JsonPath = Box<[String]>; - -#[derive(Clone, Copy, Debug, PartialEq)] -enum FingerprintFieldKind { - Tokenized, - Raw, -} - #[derive(Clone)] pub struct Fingerprinter { config: Arc, - grouping_fields: Arc<[(JsonPath, FingerprintFieldKind)]>, - ignored_paths: Arc<[JsonPath]>, + policies: Arc<[FingerprintPolicy]>, } impl Fingerprinter { pub fn new(config: &DocsClusteringConfig) -> Self { - let mut grouping_fields = Vec::new(); - let mut ignored_paths = Vec::new(); - debug_assert_eq!( - config.policies.len(), - 2, - "document clustering config must contain exactly two policies" - ); - - fn parse_path(path: &str) -> JsonPath { - path.split('.').map(ToString::to_string).collect() - } - + let mut policies = Vec::new(); for policy in &config.policies { - let ClusteringPolicy::Fingerprint { fingerprint } = policy; - for field in &fingerprint.fingerprint { - match field { - ClusteringField::Structure { exclude } => { - ignored_paths.extend(exclude.iter().map(|path| parse_path(path))); - } - ClusteringField::Raw { path } => { - grouping_fields.push((parse_path(path), FingerprintFieldKind::Raw)); - } - ClusteringField::Tokenized { path } => { - grouping_fields.push((parse_path(path), FingerprintFieldKind::Tokenized)); - } + match policy { + ClusteringPolicy::Fingerprint { fingerprint } => { + policies.push(fingerprint.clone()); } } } - // Sort the paths to ensure a stable ordering of the fingerprint. - grouping_fields.sort_unstable_by(|(left, _), (right, _)| left.cmp(right)); - Self { config: Arc::new(config.clone()), - grouping_fields: Arc::from(grouping_fields.into_boxed_slice()), - ignored_paths: Arc::from(ignored_paths.into_boxed_slice()), + policies: policies.into_boxed_slice().into(), } } @@ -148,29 +126,54 @@ impl Fingerprinter { } pub fn fingerprint(&self, json_value: &JsonValue) -> Fingerprint { - let mut schema_hasher = FnvHasher::default(); - self.hash_schema(json_value, &mut schema_hasher); - let mut grouping_hasher = FnvHasher::default(); - self.hash_grouping_fields(json_value, &mut grouping_hasher); - Fingerprint { - schema: schema_hasher.finish(), - grouping: grouping_hasher.finish(), + let mut fingerprint = SmallVec::new(); + for policy in self.policies.iter() { + let mut hasher = FnvHasher::default(); + + for field in policy.fingerprint.iter() { + match field { + ClusteringField::Structure { exclude } => { + self.hash_structure(json_value, exclude, &mut hasher); + } + ClusteringField::Raw { path } => { + self.hash_raw_field(json_value, path, &mut hasher); + } + ClusteringField::Tokenized { path } => { + self.hash_tokenized_field(json_value, path, true, &mut hasher); + } + } + } + + fingerprint.push(hasher.finish()); } + + Fingerprint(fingerprint) } - fn hash_schema(&self, value: &JsonValue, hasher: &mut FnvHasher) { + fn hash_structure(&self, value: &JsonValue, exclude: &[JsonPath], hasher: &mut FnvHasher) { fn walk<'a>( - fingerprinter: &Fingerprinter, json_value: &'a JsonValue, + exclude: &[JsonPath], current: &mut Vec<&'a str>, paths: &mut Vec>, ) { + fn is_excluded(exclude: &[JsonPath], path: &[&str]) -> bool { + exclude.iter().any(|excluded_path| { + excluded_path.len() == path.len() + && excluded_path.iter().zip(path.iter()).all( + |(excluded_component, component)| { + excluded_component.as_str() == *component + }, + ) + }) + } + match json_value { JsonValue::Object(obj) => { for (key, child_value) in obj.iter() { current.push(key.as_str()); - if !fingerprinter.is_ignored(current) { - walk(fingerprinter, child_value, current, paths); + if !is_excluded(exclude, current) { + walk(child_value, exclude, current, paths); } current.pop(); } @@ -181,7 +184,7 @@ impl Fingerprinter { let mut current = Vec::with_capacity(16); let mut paths = Vec::with_capacity(32); - walk(self, value, &mut current, &mut paths); + walk(value, exclude, &mut current, &mut paths); paths.sort_unstable(); for path in paths { @@ -193,37 +196,39 @@ impl Fingerprinter { } } - fn hash_grouping_fields(&self, json_value: &JsonValue, hasher: &mut FnvHasher) { - for (path, kind) in self.grouping_fields.iter() { - let Some(value) = get_leaf_string(json_value, path) else { - hasher.write_u8(FIELD_ABSENT); - hasher.write_u8(FIELD_BOUNDARY); - continue; - }; - hasher.write_u8(FIELD_PRESENT); - match kind { - FingerprintFieldKind::Tokenized => { - for span in tokenize(value).take(MAX_GROUPING_TOKENS) { - hasher.write_u8(span.token_type as u8); - hasher.write_u8(TOKENIZED_TOKEN_SEPARATOR); - } - } - FingerprintFieldKind::Raw => { - hasher.write(value.as_bytes()); - } - } + fn hash_raw_field(&self, json_value: &JsonValue, path: &JsonPath, hasher: &mut FnvHasher) { + let Some(value) = get_leaf_string(json_value, path) else { + hasher.write_u8(FIELD_ABSENT); hasher.write_u8(FIELD_BOUNDARY); - } + return; + }; + hasher.write_u8(FIELD_PRESENT); + hasher.write(value.as_bytes()); + hasher.write_u8(FIELD_BOUNDARY); } - fn is_ignored(&self, path: &[&str]) -> bool { - self.ignored_paths.iter().any(|ignored_path| { - ignored_path.len() == path.len() - && ignored_path - .iter() - .zip(path.iter()) - .all(|(ignored_component, component)| ignored_component == component) - }) + fn hash_tokenized_field( + &self, + json_value: &JsonValue, + path: &JsonPath, + tokenized: bool, + hasher: &mut FnvHasher, + ) { + let Some(value) = get_leaf_string(json_value, path) else { + hasher.write_u8(FIELD_ABSENT); + hasher.write_u8(FIELD_BOUNDARY); + return; + }; + hasher.write_u8(FIELD_PRESENT); + if tokenized { + for span in tokenize(value).take(MAX_GROUPING_TOKENS) { + hasher.write_u8(span.token_type as u8); + hasher.write_u8(TOKENIZED_TOKEN_SEPARATOR); + } + } else { + hasher.write(value.as_bytes()); + } + hasher.write_u8(FIELD_BOUNDARY); } } @@ -303,14 +308,8 @@ mod tests { let dotted_key_fingerprint = fingerprinter.fingerprint(&dotted_key_doc); let nested_path_fingerprint = fingerprinter.fingerprint(&nested_path_doc); - assert_ne!( - dotted_key_fingerprint.schema, - nested_path_fingerprint.schema - ); - assert_eq!( - dotted_key_fingerprint.grouping, - nested_path_fingerprint.grouping - ); + assert_ne!(dotted_key_fingerprint[0], nested_path_fingerprint[0]); + assert_eq!(dotted_key_fingerprint[1], nested_path_fingerprint[1]); } #[test] @@ -331,8 +330,8 @@ mod tests { let doc2 = parse(r#"{"message":"connection from 1.2.3.4","service":"api"}"#); let doc1_fingerprint = fingerprinter.fingerprint(&doc1); let doc2_fingerprint = fingerprinter.fingerprint(&doc2); - assert_eq!(doc1_fingerprint.schema, doc2_fingerprint.schema); - assert_ne!(doc1_fingerprint.grouping, doc2_fingerprint.grouping); + assert_eq!(doc1_fingerprint[0], doc2_fingerprint[0]); + assert_ne!(doc1_fingerprint[1], doc2_fingerprint[1]); } #[test] @@ -355,8 +354,8 @@ mod tests { let doc1 = parse(&format!(r#"{{"message":"{prefix}123"}}"#)); let doc2 = parse(&format!(r#"{{"message":"{prefix}beta"}}"#)); assert_eq!( - fingerprinter.fingerprint(&doc1).grouping, - fingerprinter.fingerprint(&doc2).grouping + fingerprinter.fingerprint(&doc1)[1], + fingerprinter.fingerprint(&doc2)[1] ); } @@ -367,8 +366,8 @@ mod tests { let doc2 = parse(r#"{"message":"server started at 8080","service":"worker"}"#); let doc1_fingerprint = fingerprinter.fingerprint(&doc1); let doc2_fingerprint = fingerprinter.fingerprint(&doc2); - assert_eq!(doc1_fingerprint.schema, doc2_fingerprint.schema); - assert_ne!(doc1_fingerprint.grouping, doc2_fingerprint.grouping); + assert_eq!(doc1_fingerprint[0], doc2_fingerprint[0]); + assert_ne!(doc1_fingerprint[1], doc2_fingerprint[1]); } #[test] @@ -391,8 +390,8 @@ mod tests { let doc2 = parse(r#"{"message":"server started at 8080","service":"api","host":"web-1"}"#); let doc1_fingerprint = fingerprinter.fingerprint(&doc1); let doc2_fingerprint = fingerprinter.fingerprint(&doc2); - assert_ne!(doc1_fingerprint.schema, doc2_fingerprint.schema); - assert_eq!(doc1_fingerprint.grouping, doc2_fingerprint.grouping); + assert_ne!(doc1_fingerprint[0], doc2_fingerprint[0]); + assert_eq!(doc1_fingerprint[1], doc2_fingerprint[1]); } #[test] @@ -415,8 +414,8 @@ mod tests { let doc2 = parse(r#"{"message":"same","host":"web-2"}"#); let doc1_fingerprint = fingerprinter.fingerprint(&doc1); let doc2_fingerprint = fingerprinter.fingerprint(&doc2); - assert_eq!(doc1_fingerprint.schema, doc2_fingerprint.schema); - assert_ne!(doc1_fingerprint.grouping, doc2_fingerprint.grouping); + assert_eq!(doc1_fingerprint[0], doc2_fingerprint[0]); + assert_ne!(doc1_fingerprint[1], doc2_fingerprint[1]); } #[test] @@ -468,7 +467,7 @@ mod tests { let doc1_fingerprint = fingerprinter.fingerprint(&doc1); let doc2_fingerprint = fingerprinter.fingerprint(&doc2); - assert_eq!(doc1_fingerprint.schema, doc2_fingerprint.schema); - assert_ne!(doc1_fingerprint.grouping, doc2_fingerprint.grouping); + assert_eq!(doc1_fingerprint[0], doc2_fingerprint[0]); + assert_ne!(doc1_fingerprint[1], doc2_fingerprint[1]); } } diff --git a/quickwit/quickwit-indexing/src/docs_clustering/mod.rs b/quickwit/quickwit-indexing/src/docs_clustering/mod.rs index 4eb35bf0502..d5888a34691 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/mod.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/mod.rs @@ -43,10 +43,10 @@ //! policy, unless `QW_DISABLE_DOCS_CLUSTERING=true`. Deployments should configure it only on //! indexers whose workloads are intended to use document clustering. //! -//! [`Fingerprinter`] computes two independent hashes: +//! [`Fingerprinter`] computes one hash for each configured fingerprint policy: //! -//! 1. A schema hash from the sorted set of leaf JSON paths, excluding configured paths. -//! 2. A grouping hash from configured raw string values and tokenized signatures. +//! 1. A structure policy hashes the sorted set of leaf JSON paths, excluding configured paths. +//! 2. Raw and tokenized policies hash configured string values and tokenized signatures. //! //! Tokenized fields hash token types rather than literal values, up to 50 tokens. This groups //! volatile values with the same shape: @@ -57,14 +57,14 @@ //! ``` //! //! Missing and non-string grouping values are encoded as absent so each configured field retains -//! its position in the grouping hash. +//! its position in the policy hash. //! //! ```text //! Raw document //! | //! v //! DocProcessor -//! | computes Fingerprint { schema, grouping } +//! | computes Fingerprint { hashes } //! v //! ProcessedDoc { doc, fingerprint_opt, ... } //! | @@ -73,8 +73,7 @@ //! | records each split-local doc ID in DocIdClusterer //! v //! IndexedSplitBuilder::finalize -//! | builds a DocIdMapping: largest schema groups first, -//! | then largest grouping-hash groups within each schema +//! | builds a DocIdMapping: largest groups first at every fingerprint level //! v //! Tantivy segment with similar documents stored together //! ``` From bd59e488f8d95b430330ba8d50acb4283e97d2b6 Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Mon, 3 Aug 2026 12:48:51 +0200 Subject: [PATCH 2/6] Remove document clusterer benchmark target Keep the benchmark evidence in the PR description while avoiding a permanent synthetic bench target in the crate. Co-authored-by: Cursor --- quickwit/quickwit-indexing/Cargo.toml | 4 - .../benches/doc_id_clusterer_bench.rs | 168 ------------------ 2 files changed, 172 deletions(-) delete mode 100644 quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs diff --git a/quickwit/quickwit-indexing/Cargo.toml b/quickwit/quickwit-indexing/Cargo.toml index 2ed527afbed..23d81a3e58c 100644 --- a/quickwit/quickwit-indexing/Cargo.toml +++ b/quickwit/quickwit-indexing/Cargo.toml @@ -147,10 +147,6 @@ required-features = ["fail/failpoints"] name = "doc_process_vrl_bench" harness = false -[[bench]] -name = "doc_id_clusterer_bench" -harness = false - [package.metadata.cargo-machete] # used to vendor/static build native dependencies ignored = ["libz-sys", "openssl"] diff --git a/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs b/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs deleted file mode 100644 index 6b8a9ae207f..00000000000 --- a/quickwit/quickwit-indexing/benches/doc_id_clusterer_bench.rs +++ /dev/null @@ -1,168 +0,0 @@ -// 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 std::hint::black_box; - -use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; -use quickwit_config::DocsClusteringConfig; -use quickwit_indexing::docs_clustering::{DocIdClusterer, Fingerprint, Fingerprinter}; -use serde_json::{Value as JsonValue, json}; -use tantivy::DocId; - -const NUM_DOCS: usize = 100_000; - -fn docs_clustering_config(num_levels: usize) -> DocsClusteringConfig { - let mut policies = vec![json!({ - "fingerprint": [ - { - "kind": "structure", - "exclude": ["message", "timestamp"] - } - ] - })]; - if num_levels >= 2 { - policies.push(json!({ - "fingerprint": [ - { - "kind": "raw", - "path": "service" - } - ] - })); - } - if num_levels >= 3 { - policies.push(json!({ - "fingerprint": [ - { - "kind": "tokenized", - "path": "message" - } - ] - })); - } - if num_levels >= 4 { - policies.push(json!({ - "fingerprint": [ - { - "kind": "raw", - "path": "status" - } - ] - })); - } - - let config: DocsClusteringConfig = serde_json::from_value(JsonValue::Array(policies)).unwrap(); - config.validate().unwrap(); - config -} - -fn generate_fingerprints( - num_docs: usize, - num_levels: usize, - cardinality: usize, -) -> Vec { - let config = docs_clustering_config(num_levels); - let fingerprinter = Fingerprinter::new(&config); - (0..num_docs) - .map(|doc_idx| { - let service_id = doc_idx % cardinality; - let template_id = (doc_idx / cardinality) % 128; - let status = match doc_idx % 5 { - 0 => "debug", - 1 => "info", - 2 => "warn", - 3 => "error", - _ => "critical", - }; - let doc = json!({ - "timestamp": doc_idx, - "service": format!("service-{service_id}"), - "message": format!("template {template_id} request {} completed in {} ms", doc_idx, doc_idx % 1_000), - "status": status, - "host": format!("host-{}", doc_idx % 1_024), - }); - fingerprinter.fingerprint(&doc) - }) - .collect() -} - -fn is_unsorted_doc(doc_idx: usize, unsorted_doc_frequency_opt: Option) -> bool { - match unsorted_doc_frequency_opt { - Some(frequency) => doc_idx % frequency == 0, - None => false, - } -} - -fn bench_doc_id_mapping( - group: &mut criterion::BenchmarkGroup<'_, criterion::measurement::WallTime>, - name: &str, - fingerprints: &[Fingerprint], - unsorted_doc_frequency_opt: Option, -) { - group.throughput(criterion::Throughput::Elements(fingerprints.len() as u64)); - group.bench_function(BenchmarkId::new(name, fingerprints.len()), |b| { - b.iter(|| { - let mut clusterer = DocIdClusterer::default(); - for (doc_idx, fingerprint) in fingerprints.iter().enumerate() { - let fingerprint_opt = if is_unsorted_doc(doc_idx, unsorted_doc_frequency_opt) { - None - } else { - Some(fingerprint.clone()) - }; - clusterer.push(fingerprint_opt, doc_idx as DocId); - } - black_box( - clusterer - .into_doc_id_mapping(fingerprints.len() as u64) - .unwrap(), - ) - }) - }); -} - -fn bench_doc_id_clusterer(c: &mut Criterion) { - let low_cardinality_fingerprints = generate_fingerprints(NUM_DOCS, 2, 16); - let high_cardinality_fingerprints = generate_fingerprints(NUM_DOCS, 2, 16_384); - let deep_fingerprints = generate_fingerprints(NUM_DOCS, 4, 256); - - let mut group = c.benchmark_group("DocIdClusterer"); - bench_doc_id_mapping( - &mut group, - "two-level/low-cardinality", - &low_cardinality_fingerprints, - None, - ); - bench_doc_id_mapping( - &mut group, - "two-level/high-cardinality", - &high_cardinality_fingerprints, - None, - ); - bench_doc_id_mapping( - &mut group, - "four-level/mixed-cardinality", - &deep_fingerprints, - None, - ); - bench_doc_id_mapping( - &mut group, - "four-level/with-unsorted-docs", - &deep_fingerprints, - Some(10), - ); - group.finish(); -} - -criterion_group!(benches, bench_doc_id_clusterer); -criterion_main!(benches); From 810de92df3085226b540ba353cf58b4e9863a4d2 Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Mon, 3 Aug 2026 14:13:24 +0200 Subject: [PATCH 3/6] Stream document cluster metrics during finalization Move cluster-size observation into the clusterer so recursive traversal stays allocation-free and mapping validation remains owned by Tantivy. Co-authored-by: Cursor --- .../quickwit-indexing/src/actors/indexer.rs | 14 ++--- .../src/docs_clustering/clusterer.rs | 54 ++++++++----------- .../src/docs_clustering/fingerprinter.rs | 2 +- .../src/models/indexed_split.rs | 11 ++-- 4 files changed, 32 insertions(+), 49 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 6ed248676ce..5de2eda20f0 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -1266,35 +1266,35 @@ mod tests { let docs = vec![ ProcessedDoc { doc: doc!(body_field=>"first"), - fingerprint_opt: Some(Fingerprint::for_test([1, 1, 1])), + fingerprint_opt: Some(Fingerprint::new([1, 1, 1])), timestamp_opt: None, partition: 0, num_bytes: 5, }, ProcessedDoc { doc: doc!(body_field=>"second"), - fingerprint_opt: Some(Fingerprint::for_test([1, 2, 1])), + fingerprint_opt: Some(Fingerprint::new([1, 2, 1])), timestamp_opt: None, partition: 0, num_bytes: 6, }, ProcessedDoc { doc: doc!(body_field=>"third"), - fingerprint_opt: Some(Fingerprint::for_test([1, 2, 1])), + fingerprint_opt: Some(Fingerprint::new([1, 2, 1])), timestamp_opt: None, partition: 0, num_bytes: 5, }, ProcessedDoc { doc: doc!(body_field=>"fourth"), - fingerprint_opt: Some(Fingerprint::for_test([1, 1, 1])), + fingerprint_opt: Some(Fingerprint::new([1, 1, 1])), timestamp_opt: None, partition: 0, num_bytes: 6, }, ProcessedDoc { doc: doc!(body_field=>"fifth"), - fingerprint_opt: Some(Fingerprint::for_test([1, 1, 2])), + fingerprint_opt: Some(Fingerprint::new([1, 1, 2])), timestamp_opt: None, partition: 0, num_bytes: 5, @@ -1314,10 +1314,6 @@ mod tests { index_serializer_inbox.drain_for_test_typed(); let mut split_batch = split_batches.pop().unwrap(); let split_builder = split_batch.splits.pop().unwrap(); - let clusterer = split_builder.doc_id_clusterer_opt.as_ref().unwrap(); - let mut sort_group_sizes = clusterer.sort_group_sizes().collect_vec(); - sort_group_sizes.sort_unstable(); - assert_eq!(sort_group_sizes, [1, 2, 2]); let indexed_split = split_builder.finalize()?; let reader = indexed_split.index.reader()?; diff --git a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs index aed78e7ae65..a3b48269983 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs @@ -46,11 +46,13 @@ use std::cmp::Reverse; use std::mem; use fnv::FnvHashMap; +use quickwit_metrics::{Labels, histogram}; use smallvec::SmallVec; use tantivy::DocId; use tantivy::indexer::DocIdMapping; use super::Fingerprint; +use crate::metrics::DOCS_SORT_GROUP_SIZE; // We inline as many DocIds as possible to avoid heap allocations. // This is done by calculating the inline capacity based on the size of the Vec and the @@ -87,21 +89,25 @@ impl DocIdClusterer { } } - pub fn sort_group_sizes(&self) -> impl Iterator + '_ { - self.root.leaf_sizes() - } - - pub fn into_doc_id_mapping(self, num_docs: u64) -> anyhow::Result { + pub fn into_doc_id_mapping(self) -> anyhow::Result { let doc_ids = self.into_sorted_doc_ids(); - debug_assert_eq!(doc_ids.len(), num_docs as usize); let doc_id_mapping = DocIdMapping::new_permutation(doc_ids)?; Ok(doc_id_mapping) } + /// Internal iteration avoids heap allocations and the complex traversal state required by an + /// external iterator over the recursive cluster tree. + pub(crate) fn observe_cluster_group_sizes(&self, labels: Labels) { + let h = histogram!(parent: DOCS_SORT_GROUP_SIZE, labels: [labels]); + self.root.for_each_leaf(&mut |cluster_group_doc_ids| { + h.observe(cluster_group_doc_ids.len() as f64); + }); + } + fn into_sorted_doc_ids(self) -> Vec { let mut doc_ids = Vec::with_capacity(self.root.num_docs + self.unclustered_docs.len()); self.root.append_sorted_doc_ids(&mut doc_ids); - doc_ids.extend(self.unclustered_docs); + doc_ids.extend_from_slice(&self.unclustered_docs); doc_ids } } @@ -121,7 +127,7 @@ impl ClusterGroup { } fn append_sorted_doc_ids(self, doc_ids: &mut Vec) { - doc_ids.extend(self.doc_ids); + doc_ids.extend_from_slice(&self.doc_ids); let mut children: Vec = self.children.into_values().collect(); // Larger groups are emitted first. children.sort_unstable_by_key(|child| Reverse(child.num_docs)); @@ -130,16 +136,13 @@ impl ClusterGroup { } } - /// Iterates over the leaf sort group sizes of this subtree. A leaf reports its document count, - /// while an empty tree reports nothing. - /// - /// The return type is boxed to break the otherwise self-referential (and thus infinitely sized) - /// recursive `impl Iterator` type. - fn leaf_sizes(&self) -> Box + '_> { - let leaf_size = (!self.doc_ids.is_empty()) - .then_some(self.doc_ids.len()) - .into_iter(); - Box::new(leaf_size.chain(self.children.values().flat_map(ClusterGroup::leaf_sizes))) + fn for_each_leaf(&self, f: &mut impl FnMut(&[DocId])) { + if !self.doc_ids.is_empty() { + f(&self.doc_ids); + } + for child in self.children.values() { + child.for_each_leaf(f); + } } } @@ -148,7 +151,7 @@ mod tests { use super::{DocIdClusterer, Fingerprint}; fn fingerprint(hashes: [u64; N]) -> Fingerprint { - Fingerprint::for_test(hashes) + Fingerprint::new(hashes) } #[test] @@ -236,17 +239,4 @@ mod tests { assert_eq!(clusterer.into_sorted_doc_ids(), [0, 2, 1, 3]); } - - #[test] - fn reports_leaf_grouping_sort_group_sizes() { - let mut clusterer = DocIdClusterer::default(); - clusterer.push(Some(fingerprint([1, 1])), 0); - clusterer.push(Some(fingerprint([1, 2])), 1); - clusterer.push(Some(fingerprint([1, 2])), 2); - clusterer.push(Some(fingerprint([2, 1])), 3); - - let mut sort_group_sizes = clusterer.sort_group_sizes().collect::>(); - sort_group_sizes.sort_unstable(); - assert_eq!(sort_group_sizes, [1, 1, 2]); - } } diff --git a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs index eae29d62730..0b0885e61af 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs @@ -91,7 +91,7 @@ impl Deref for Fingerprint { #[cfg(test)] impl Fingerprint { - pub(crate) fn for_test(hashes: [u64; N]) -> Self { + pub(crate) fn new(hashes: [u64; N]) -> Self { let mut fingerprint = SmallVec::new(); fingerprint.extend_from_slice(&hashes); Self(fingerprint) diff --git a/quickwit/quickwit-indexing/src/models/indexed_split.rs b/quickwit/quickwit-indexing/src/models/indexed_split.rs index 5f49c5d799c..09e6b47f831 100644 --- a/quickwit/quickwit-indexing/src/models/indexed_split.rs +++ b/quickwit/quickwit-indexing/src/models/indexed_split.rs @@ -19,7 +19,7 @@ use quickwit_common::io::IoControls; use quickwit_common::metrics::index_label; use quickwit_common::temp_dir::TempDirectory; use quickwit_metastore::checkpoint::IndexCheckpointDelta; -use quickwit_metrics::{GaugeGuard, histogram, label_values}; +use quickwit_metrics::{GaugeGuard, label_values}; use quickwit_proto::indexing::IndexingPipelineId; use quickwit_proto::types::{DocMappingUid, IndexUid, SplitId}; use tantivy::IndexBuilder; @@ -29,7 +29,7 @@ use tracing::{Span, error, instrument}; use crate::controlled_directory::ControlledDirectory; use crate::docs_clustering::DocIdClusterer; use crate::merge_policy::MergeTask; -use crate::metrics::{DOCS_SORT_GROUP_SIZE, INDEX_SOURCE}; +use crate::metrics::INDEX_SOURCE; use crate::models::{PublishLock, SplitAttrs}; pub struct IndexedSplitBuilder { @@ -147,14 +147,11 @@ impl IndexedSplitBuilder { INDEX_SOURCE => index_label.to_string(), split_attrs.source_id.to_string() ); - for sort_group_size in doc_id_clusterer.sort_group_sizes() { - histogram!(parent: DOCS_SORT_GROUP_SIZE, labels: [labels.clone()]) - .observe(sort_group_size as f64); - } + doc_id_clusterer.observe_cluster_group_sizes(labels); // Finalize the index with the doc id mapping. let doc_id_mapping = doc_id_clusterer - .into_doc_id_mapping(split_attrs.num_docs) + .into_doc_id_mapping() .inspect_err(|error| { error!(?error, "failed to create doc id mapping"); })?; From 62fb2fd1d783b34e05c9d31a715853e4a5254fcb Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Mon, 3 Aug 2026 14:49:00 +0200 Subject: [PATCH 4/6] Document multi-level fingerprint hashing Co-authored-by: Cursor --- .../src/docs_clustering/fingerprinter.rs | 29 ++++++++++--------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs index 0b0885e61af..7d6dd67ca07 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/fingerprinter.rs @@ -14,23 +14,22 @@ //! Fingerprint computation. //! -//! A fingerprint contains two independent components: -//! 1. the schema hash, represented by the sorted set of leaf JSON paths; -//! 2. the grouping hash, represented by raw string values or token types from the configured string -//! fields. +//! A fingerprint contains one hash for each configured fingerprint policy, in configuration order. +//! Each policy can combine: +//! 1. structure fields, represented by sorted sets of leaf JSON paths; +//! 2. raw fields, represented by their exact string values; +//! 3. tokenized fields, represented by their token types. //! //! ```text //! JSON document //! | -//! +--> schema paths, excluding paths configured by `structure.exclude` -//! | -//! +--> configured grouping fields -//! | -//! +--> Tokenized: hash token types, not literal values -//! +--> Raw: hash the exact string value +//! +--> policy 0 fields --> hash 0 +//! +--> policy 1 fields --> hash 1 +//! +--> ... +//! +--> policy N fields --> hash N //! | //! v -//! Fingerprint +//! Fingerprint [hash 0, hash 1, ..., hash N] //! ``` //! //! Tokenized fields use the sequence of token types as a lightweight message template. This keeps @@ -46,8 +45,8 @@ //! -> Word Gap Word Gap Word Gap Number //! ``` //! -//! These two values produce the same tokenized signature and can be ordered together. A different -//! shape produces a different signature: +//! These two values produce the same tokenized signature and policy hash, so they can be ordered +//! together. A different shape produces a different signature: //! //! ```text //! "connection from 1.2.3.4" @@ -56,7 +55,7 @@ //! //! Raw string fields keep exact-value differences, which is useful for dimensions such as //! `service`. Missing fields and non-string values are encoded as absent so configured field -//! positions remain distinct. +//! positions remain distinct within a policy hash. use std::hash::Hasher; use std::ops::Deref; use std::sync::Arc; @@ -78,6 +77,8 @@ const FIELD_BOUNDARY: u8 = 0xFD; const TOKENIZED_TOKEN_SEPARATOR: u8 = 0xFE; const MAX_GROUPING_TOKENS: usize = 50; +// Inline 4 hashes to avoid heap allocations. +// This is usually enough for most use cases. #[derive(Clone, Debug, Eq, PartialEq)] pub struct Fingerprint(SmallVec<[u64; 4]>); From 62b9e861c59635ad5929dbe4b6702b70afe03cc3 Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Mon, 3 Aug 2026 14:58:08 +0200 Subject: [PATCH 5/6] Clarify cluster group documentation Co-authored-by: Cursor --- .../quickwit-indexing/src/docs_clustering/clusterer.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs index a3b48269983..fa31ad55356 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs @@ -69,11 +69,10 @@ pub struct DocIdClusterer { unclustered_docs: ClusterDocIds, } -/// A node in the fingerprint prefix tree. +/// A group of documents that share the same fingerprint prefix. /// -/// Each level of the tree partitions documents by one fingerprint hash. Internal nodes hold their -/// children keyed by the hash at their level; leaf nodes (documents that share the exact same -/// fingerprint) hold the corresponding doc IDs in insertion order. +/// The root contains all documents. Each successive tree level groups them by the next hash in +/// their fingerprint. #[derive(Default)] struct ClusterGroup { num_docs: usize, @@ -95,8 +94,6 @@ impl DocIdClusterer { Ok(doc_id_mapping) } - /// Internal iteration avoids heap allocations and the complex traversal state required by an - /// external iterator over the recursive cluster tree. pub(crate) fn observe_cluster_group_sizes(&self, labels: Labels) { let h = histogram!(parent: DOCS_SORT_GROUP_SIZE, labels: [labels]); self.root.for_each_leaf(&mut |cluster_group_doc_ids| { From 16fca896950f5a707d08bb7b58ed79488122a5c5 Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Mon, 3 Aug 2026 15:02:22 +0200 Subject: [PATCH 6/6] Rename document cluster group metric Co-authored-by: Cursor --- quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs | 4 ++-- quickwit/quickwit-indexing/src/metrics.rs | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs index fa31ad55356..77c7fd5da9e 100644 --- a/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs +++ b/quickwit/quickwit-indexing/src/docs_clustering/clusterer.rs @@ -52,7 +52,7 @@ use tantivy::DocId; use tantivy::indexer::DocIdMapping; use super::Fingerprint; -use crate::metrics::DOCS_SORT_GROUP_SIZE; +use crate::metrics::DOCS_CLUSTER_GROUP_SIZE; // We inline as many DocIds as possible to avoid heap allocations. // This is done by calculating the inline capacity based on the size of the Vec and the @@ -95,7 +95,7 @@ impl DocIdClusterer { } pub(crate) fn observe_cluster_group_sizes(&self, labels: Labels) { - let h = histogram!(parent: DOCS_SORT_GROUP_SIZE, labels: [labels]); + let h = histogram!(parent: DOCS_CLUSTER_GROUP_SIZE, labels: [labels]); self.root.for_each_leaf(&mut |cluster_group_doc_ids| { h.observe(cluster_group_doc_ids.len() as f64); }); diff --git a/quickwit/quickwit-indexing/src/metrics.rs b/quickwit/quickwit-indexing/src/metrics.rs index b7431c6d032..9a874f7f578 100644 --- a/quickwit/quickwit-indexing/src/metrics.rs +++ b/quickwit/quickwit-indexing/src/metrics.rs @@ -36,9 +36,9 @@ pub(crate) static PROCESSED_BYTES: LazyCounter = lazy_counter!( subsystem: "indexing", ); -pub(crate) static DOCS_SORT_GROUP_SIZE: LazyHistogram = lazy_histogram!( - name: "docs_sort_group_size", - description: "Document sort group size when finalizing an indexed split.", +pub(crate) static DOCS_CLUSTER_GROUP_SIZE: LazyHistogram = lazy_histogram!( + name: "docs_cluster_group_size", + description: "Document cluster group size when finalizing an indexed split.", subsystem: "indexing", buckets: exponential_buckets(1.0, 10.0, 8).unwrap(), );