Generalize document clustering fingerprints

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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where S: Serializer {
+        serializer.serialize_str(&self.0.join("."))
+    }
+}
+
+impl<'de> Deserialize<'de> for JsonPath {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    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,68 +197,31 @@ pub enum ClusteringField {
     Structure {
         /// Paths omitted from the structure fingerprint.
         #[serde(default, skip_serializing_if = "Vec::is_empty")]
-        exclude: Vec<String>,
+        exclude: Vec<JsonPath>,
     },
     /// Groups documents by the exact string value at `path`.
     ///
     /// For example, `{"service":"api"}` and `{"service":"worker"}` belong to different groups
     /// 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`.
     ///
     /// For example, `{"message":"request 123"}` and `{"message":"request 456"}` belong to the same
     /// 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<Option<DocsClusteringConfig>> {
         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::<JsonPath>(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::<DocsClusteringConfigBuilder>("- fingerprint: [")

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

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<ProcessedDocBatch> = 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;
     }