Add min_shards parameter to index config (#5781)

Author: guilload

quickwit/quickwit-cli/src/source.rs

@@ -823,7 +823,7 @@ mod tests {
             .collect();
         let sources = vec![SourceConfig {
             source_id: "foo-source".to_string(),
-            num_pipelines: NonZeroUsize::new(1).unwrap(),
+            num_pipelines: NonZeroUsize::MIN,
             enabled: true,
             source_params: SourceParams::file_from_str("path/to/file").unwrap(),
             transform_config: None,
@@ -884,15 +884,15 @@ mod tests {
         let sources = [
             SourceConfig {
                 source_id: "foo-source".to_string(),
-                num_pipelines: NonZeroUsize::new(1).unwrap(),
+                num_pipelines: NonZeroUsize::MIN,
                 enabled: true,
                 source_params: SourceParams::stdin(),
                 transform_config: None,
                 input_format: SourceInputFormat::Json,
             },
             SourceConfig {
                 source_id: "bar-source".to_string(),
-                num_pipelines: NonZeroUsize::new(1).unwrap(),
+                num_pipelines: NonZeroUsize::MIN,
                 enabled: true,
                 source_params: SourceParams::stdin(),
                 transform_config: None,

quickwit/quickwit-cli/src/tool.rs

@@ -415,7 +415,7 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result<
         .map(|vrl_script| TransformConfig::new(vrl_script, None));
     let source_config = SourceConfig {
         source_id: CLI_SOURCE_ID.to_string(),
-        num_pipelines: NonZeroUsize::new(1).expect("1 is always non-zero."),
+        num_pipelines: NonZeroUsize::MIN,
         enabled: true,
         source_params,
         transform_config,
@@ -605,7 +605,7 @@ pub async fn merge_cli(args: MergeArgs) -> anyhow::Result<()> {
             index_id: args.index_id,
             source_config: SourceConfig {
                 source_id: args.source_id,
-                num_pipelines: NonZeroUsize::new(1).unwrap(),
+                num_pipelines: NonZeroUsize::MIN,
                 enabled: true,
                 source_params: SourceParams::Vec(VecSourceParams::default()),
                 transform_config: None,

quickwit/quickwit-config/resources/tests/index_config/hdfs-logs.json

@@ -66,6 +66,9 @@
             "heap_size": "3G"
         }
     },
+    "ingest_settings": {
+        "min_shards": 12
+    },
     "search_settings": {
         "default_search_fields": ["severity_text", "body"]
     }

quickwit/quickwit-config/resources/tests/index_config/hdfs-logs.toml

@@ -34,5 +34,8 @@ maturation_period = "48 hours"
 [indexing_settings.resources]
 heap_size = "3G"
 
+[ingest_settings]
+min_shards = 12
+
 [search_settings]
 default_search_fields = [ "severity_text", "body" ]

quickwit/quickwit-config/resources/tests/index_config/hdfs-logs.yaml

@@ -46,5 +46,8 @@ indexing_settings:
   resources:
     heap_size: 3G
 
+ingest_settings:
+  min_shards: 12
+
 search_settings:
   default_search_fields: [severity_text, body]

quickwit/quickwit-config/src/index_config/mod.rs

@@ -15,6 +15,7 @@
 pub(crate) mod serialize;
 
 use std::hash::{Hash, Hasher};
+use std::num::NonZeroUsize;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -161,6 +162,30 @@ impl Default for IndexingSettings {
     }
 }
 
+/// Settings for ingestion.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct IngestSettings {
+    /// Configures the minimum number of shards to use for ingestion.
+    #[schema(default = 1, value_type = usize)]
+    #[serde(default = "IngestSettings::default_min_shards")]
+    pub min_shards: NonZeroUsize,
+}
+
+impl IngestSettings {
+    pub fn default_min_shards() -> NonZeroUsize {
+        NonZeroUsize::MIN
+    }
+}
+
+impl Default for IngestSettings {
+    fn default() -> Self {
+        Self {
+            min_shards: Self::default_min_shards(),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
 #[serde(deny_unknown_fields)]
 pub struct SearchSettings {
@@ -251,6 +276,7 @@ pub struct IndexConfig {
     pub index_uri: Uri,
     pub doc_mapping: DocMapping,
     pub indexing_settings: IndexingSettings,
+    pub ingest_settings: IngestSettings,
     pub search_settings: SearchSettings,
     pub retention_policy_opt: Option<RetentionPolicy>,
 }
@@ -359,8 +385,9 @@ impl IndexConfig {
             index_uri,
             doc_mapping,
             indexing_settings,
+            ingest_settings: IngestSettings::default(),
             search_settings,
-            retention_policy_opt: Default::default(),
+            retention_policy_opt: None,
         }
     }
 }
@@ -435,10 +462,6 @@ impl crate::TestableForRegression for IndexConfig {
             store_source: true,
             tokenizers: vec![tokenizer],
         };
-        let retention_policy = Some(RetentionPolicy {
-            retention_period: "90 days".to_string(),
-            evaluation_schedule: "daily".to_string(),
-        });
         let stable_log_config = StableLogMergePolicyConfig {
             merge_factor: 9,
             max_merge_factor: 11,
@@ -456,16 +479,24 @@ impl crate::TestableForRegression for IndexConfig {
             resources: indexing_resources,
             ..Default::default()
         };
+        let ingest_settings = IngestSettings {
+            min_shards: NonZeroUsize::new(12).unwrap(),
+        };
         let search_settings = SearchSettings {
             default_search_fields: vec!["message".to_string()],
         };
+        let retention_policy_opt = Some(RetentionPolicy {
+            retention_period: "90 days".to_string(),
+            evaluation_schedule: "daily".to_string(),
+        });
         IndexConfig {
             index_id: "my-index".to_string(),
             index_uri: Uri::for_test("s3://quickwit-indexes/my-index"),
             doc_mapping,
             indexing_settings,
-            retention_policy_opt: retention_policy,
+            ingest_settings,
             search_settings,
+            retention_policy_opt,
         }
     }
 
@@ -474,7 +505,9 @@ impl crate::TestableForRegression for IndexConfig {
         assert_eq!(self.index_uri, other.index_uri);
         assert_eq!(self.doc_mapping, other.doc_mapping);
         assert_eq!(self.indexing_settings, other.indexing_settings);
+        assert_eq!(self.ingest_settings, other.ingest_settings);
         assert_eq!(self.search_settings, other.search_settings);
+        assert_eq!(self.retention_policy_opt, other.retention_policy_opt);
     }
 }
 
@@ -488,7 +521,8 @@ pub fn build_doc_mapper(
         default_search_fields: search_settings.default_search_fields.clone(),
         legacy_type_tag: None,
     };
-    Ok(Arc::new(builder.try_build()?))
+    let doc_mapper = builder.try_build()?;
+    Ok(Arc::new(doc_mapper))
 }
 
 /// Validates the objects that make up an index configuration. This is a "free" function as opposed
@@ -598,6 +632,7 @@ mod tests {
                 ..Default::default()
             }
         );
+        assert_eq!(index_config.ingest_settings.min_shards.get(), 12);
         assert_eq!(
             index_config.search_settings,
             SearchSettings {
@@ -640,12 +675,13 @@ mod tests {
             assert_eq!(index_config.doc_mapping.field_mappings[0].name, "body");
             assert!(!index_config.doc_mapping.store_source);
             assert_eq!(index_config.indexing_settings, IndexingSettings::default());
-            assert_eq!(
-                index_config.search_settings,
-                SearchSettings {
-                    default_search_fields: vec!["body".to_string()],
-                }
-            );
+            assert_eq!(index_config.ingest_settings, IngestSettings::default());
+
+            let expected_search_settings = SearchSettings {
+                default_search_fields: vec!["body".to_string()],
+            };
+            assert_eq!(index_config.search_settings, expected_search_settings);
+            assert!(index_config.retention_policy_opt.is_none());
         }
         {
             let index_config_filepath = get_index_config_filepath("partial-hdfs-logs.yaml");
@@ -903,4 +939,21 @@ mod tests {
         schedule_test_helper_fn("monthly");
         schedule_test_helper_fn("* * * ? * ?");
     }
+
+    #[test]
+    fn test_ingest_settings_serde() {
+        let ingest_settings = IngestSettings {
+            min_shards: NonZeroUsize::MIN,
+        };
+        let ingest_settings_yaml = serde_yaml::to_string(&ingest_settings).unwrap();
+        let ingest_settings_roundtrip: IngestSettings =
+            serde_yaml::from_str(&ingest_settings_yaml).unwrap();
+        assert_eq!(ingest_settings, ingest_settings_roundtrip);
+
+        let ingest_settings_yaml = r#"
+            min_shards: 0
+        "#;
+        let error = serde_yaml::from_str::<IngestSettings>(ingest_settings_yaml).unwrap_err();
+        assert!(error.to_string().contains("expected a nonzero"));
+    }
 }

quickwit/quickwit-config/src/index_config/serialize.rs

@@ -21,7 +21,7 @@ use quickwit_proto::types::{DocMappingUid, IndexId};
 use serde::{Deserialize, Serialize};
 use tracing::info;
 
-use super::validate_index_config;
+use super::{IngestSettings, validate_index_config};
 use crate::{
     ConfigFormat, DocMapping, IndexConfig, IndexingSettings, RetentionPolicy, SearchSettings,
     validate_identifier,
@@ -185,6 +185,7 @@ impl IndexConfigForSerialization {
             index_uri,
             doc_mapping: self.doc_mapping,
             indexing_settings: self.indexing_settings,
+            ingest_settings: self.ingest_settings,
             search_settings: self.search_settings,
             retention_policy_opt: self.retention_policy_opt,
         };
@@ -231,6 +232,8 @@ pub struct IndexConfigV0_8 {
     #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(rename = "retention")]
     #[serde(default)]
@@ -244,6 +247,7 @@ impl From<IndexConfig> for IndexConfigV0_8 {
             index_uri: Some(index_config.index_uri),
             doc_mapping: index_config.doc_mapping,
             indexing_settings: index_config.indexing_settings,
+            ingest_settings: index_config.ingest_settings,
             search_settings: index_config.search_settings,
             retention_policy_opt: index_config.retention_policy_opt,
         }

quickwit/quickwit-config/src/index_template/mod.rs

@@ -20,7 +20,7 @@ use quickwit_proto::types::{DocMappingUid, IndexId};
 use serde::{Deserialize, Serialize};
 pub use serialize::{IndexTemplateV0_8, VersionedIndexTemplate};
 
-use crate::index_config::validate_index_config;
+use crate::index_config::{IngestSettings, validate_index_config};
 use crate::{
     DocMapping, IndexConfig, IndexingSettings, RetentionPolicy, SearchSettings,
     validate_identifier, validate_index_id_pattern,
@@ -45,6 +45,8 @@ pub struct IndexTemplate {
     #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(rename = "retention")]
     #[serde(default)]
@@ -72,6 +74,7 @@ impl IndexTemplate {
             index_uri,
             doc_mapping,
             indexing_settings: self.indexing_settings.clone(),
+            ingest_settings: self.ingest_settings.clone(),
             search_settings: self.search_settings.clone(),
             retention_policy_opt: self.retention_policy_opt.clone(),
         };
@@ -128,6 +131,7 @@ impl IndexTemplate {
             description: Some("Test description.".to_string()),
             doc_mapping,
             indexing_settings: IndexingSettings::default(),
+            ingest_settings: IngestSettings::default(),
             search_settings: SearchSettings::default(),
             retention_policy_opt: None,
         }
@@ -168,6 +172,7 @@ impl crate::TestableForRegression for IndexTemplate {
             description: Some("Test description.".to_string()),
             doc_mapping,
             indexing_settings: IndexingSettings::default(),
+            ingest_settings: IngestSettings::default(),
             search_settings: SearchSettings::default(),
             retention_policy_opt: Some(RetentionPolicy {
                 retention_period: "42 days".to_string(),

quickwit/quickwit-config/src/index_template/serialize.rs

@@ -16,6 +16,7 @@ use quickwit_common::uri::Uri;
 use serde::{Deserialize, Serialize};
 
 use super::{IndexIdPattern, IndexTemplate, IndexTemplateId};
+use crate::index_config::IngestSettings;
 use crate::{DocMapping, IndexingSettings, RetentionPolicy, SearchSettings};
 
 #[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema)]
@@ -49,6 +50,8 @@ pub struct IndexTemplateV0_8 {
     #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(default)]
     pub retention: Option<RetentionPolicy>,
@@ -78,6 +81,7 @@ impl From<IndexTemplateV0_8> for IndexTemplate {
             description: index_template_v0_8.description,
             doc_mapping: index_template_v0_8.doc_mapping,
             indexing_settings: index_template_v0_8.indexing_settings,
+            ingest_settings: index_template_v0_8.ingest_settings,
             search_settings: index_template_v0_8.search_settings,
             retention_policy_opt: index_template_v0_8.retention,
         }
@@ -94,6 +98,7 @@ impl From<IndexTemplate> for IndexTemplateV0_8 {
             description: index_template.description,
             doc_mapping: index_template.doc_mapping,
             indexing_settings: index_template.indexing_settings,
+            ingest_settings: index_template.ingest_settings,
             search_settings: index_template.search_settings,
             retention: index_template.retention_policy_opt,
         }