Handle ingest settings during index config update

Author: guilload

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

@@ -53,9 +53,6 @@
         "period": "90 days",
         "schedule": "daily"
     },
-    "ingest_settings": {
-        "min_shards": 12
-    },
     "indexing_settings": {
         "commit_timeout_secs": 61,
         "split_num_docs_target": 10000001,
@@ -69,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

@@ -21,13 +21,13 @@ timestamp_field = "timestamp"
 period = "90 days"
 schedule = "daily"
 
-[ingest_settings]
-min_shards = 12
-
 [indexing_settings]
 commit_timeout_secs = 61
 split_num_docs_target = 10_000_001
 
+[ingest_settings]
+min_shards = 12
+
 [indexing_settings.merge_policy]
 type = "stable_log"
 merge_factor = 9

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

@@ -35,9 +35,6 @@ retention:
   period: 90 days
   schedule: daily
 
-ingest_settings:
-  min_shards: 12
-
 indexing_settings:
   commit_timeout_secs: 61
   split_num_docs_target: 10000001
@@ -49,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

@@ -36,28 +36,6 @@ use tracing::warn;
 use crate::index_config::serialize::VersionedIndexConfig;
 use crate::merge_policy_config::MergePolicyConfig;
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
-#[serde(deny_unknown_fields)]
-pub struct IngestSettings {
-    #[schema(default = 1)]
-    #[serde(default = "IngestSettings::default_min_shards")]
-    pub min_shards: NonZeroUsize,
-}
-
-impl IngestSettings {
-    pub fn default_min_shards() -> NonZeroUsize {
-        NonZeroUsize::new(1).unwrap()
-    }
-}
-
-impl Default for IngestSettings {
-    fn default() -> Self {
-        Self {
-            min_shards: Self::default_min_shards(),
-        }
-    }
-}
-
 #[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema)]
 #[serde(deny_unknown_fields)]
 pub struct IndexingResources {
@@ -184,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)]
+    #[serde(default = "IngestSettings::default_min_shards")]
+    pub min_shards: NonZeroUsize,
+}
+
+impl IngestSettings {
+    pub fn default_min_shards() -> NonZeroUsize {
+        NonZeroUsize::new(1).unwrap()
+    }
+}
+
+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 {
@@ -273,8 +275,8 @@ pub struct IndexConfig {
     pub index_id: IndexId,
     pub index_uri: Uri,
     pub doc_mapping: DocMapping,
-    pub ingest_settings: IngestSettings,
     pub indexing_settings: IndexingSettings,
+    pub ingest_settings: IngestSettings,
     pub search_settings: SearchSettings,
     pub retention_policy_opt: Option<RetentionPolicy>,
 }
@@ -382,8 +384,8 @@ impl IndexConfig {
             index_id: index_id.to_string(),
             index_uri,
             doc_mapping,
-            ingest_settings: IngestSettings::default(),
             indexing_settings,
+            ingest_settings: IngestSettings::default(),
             search_settings,
             retention_policy_opt: None,
         }
@@ -481,17 +483,20 @@ 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()],
         };
         IndexConfig {
             index_id: "my-index".to_string(),
             index_uri: Uri::for_test("s3://quickwit-indexes/my-index"),
             doc_mapping,
-            ingest_settings: IngestSettings::default(),
             indexing_settings,
-            retention_policy_opt: retention_policy,
+            ingest_settings,
             search_settings,
+            retention_policy_opt: retention_policy,
         }
     }
 
@@ -500,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);
     }
 }
 
@@ -514,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
@@ -607,7 +615,6 @@ mod tests {
             index_config.doc_mapping.timestamp_field.unwrap(),
             "timestamp"
         );
-        assert_eq!(index_config.ingest_settings.min_shards.get(), 12);
         assert_eq!(index_config.indexing_settings.commit_timeout_secs, 61);
         assert_eq!(
             index_config.indexing_settings.merge_policy,
@@ -625,6 +632,7 @@ mod tests {
                 ..Default::default()
             }
         );
+        assert_eq!(index_config.ingest_settings.min_shards.get(), 12);
         assert_eq!(
             index_config.search_settings,
             SearchSettings {

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

@@ -184,8 +184,8 @@ impl IndexConfigForSerialization {
             index_id: self.index_id,
             index_uri,
             doc_mapping: self.doc_mapping,
-            ingest_settings: self.ingest_settings,
             indexing_settings: self.indexing_settings,
+            ingest_settings: self.ingest_settings,
             search_settings: self.search_settings,
             retention_policy_opt: self.retention_policy_opt,
         };
@@ -230,10 +230,10 @@ pub struct IndexConfigV0_8 {
     pub index_uri: Option<Uri>,
     pub doc_mapping: DocMapping,
     #[serde(default)]
-    pub ingest_settings: IngestSettings,
-    #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(rename = "retention")]
     #[serde(default)]
@@ -246,8 +246,8 @@ impl From<IndexConfig> for IndexConfigV0_8 {
             index_id: index_config.index_id,
             index_uri: Some(index_config.index_uri),
             doc_mapping: index_config.doc_mapping,
-            ingest_settings: index_config.ingest_settings,
             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

@@ -43,10 +43,10 @@ pub struct IndexTemplate {
     pub description: Option<String>,
     pub doc_mapping: DocMapping,
     #[serde(default)]
-    pub ingest_settings: IngestSettings,
-    #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(rename = "retention")]
     #[serde(default)]
@@ -73,8 +73,8 @@ impl IndexTemplate {
             index_id,
             index_uri,
             doc_mapping,
-            ingest_settings: self.ingest_settings.clone(),
             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(),
         };
@@ -130,8 +130,8 @@ impl IndexTemplate {
             priority,
             description: Some("Test description.".to_string()),
             doc_mapping,
-            ingest_settings: IngestSettings::default(),
             indexing_settings: IndexingSettings::default(),
+            ingest_settings: IngestSettings::default(),
             search_settings: SearchSettings::default(),
             retention_policy_opt: None,
         }
@@ -171,8 +171,8 @@ impl crate::TestableForRegression for IndexTemplate {
             priority: 100,
             description: Some("Test description.".to_string()),
             doc_mapping,
-            ingest_settings: IngestSettings::default(),
             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

@@ -48,10 +48,10 @@ pub struct IndexTemplateV0_8 {
 
     pub doc_mapping: DocMapping,
     #[serde(default)]
-    pub ingest_settings: IngestSettings,
-    #[serde(default)]
     pub indexing_settings: IndexingSettings,
     #[serde(default)]
+    pub ingest_settings: IngestSettings,
+    #[serde(default)]
     pub search_settings: SearchSettings,
     #[serde(default)]
     pub retention: Option<RetentionPolicy>,
@@ -80,8 +80,8 @@ impl From<IndexTemplateV0_8> for IndexTemplate {
             priority: index_template_v0_8.priority,
             description: index_template_v0_8.description,
             doc_mapping: index_template_v0_8.doc_mapping,
-            ingest_settings: index_template_v0_8.ingest_settings,
             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,
         }
@@ -97,8 +97,8 @@ impl From<IndexTemplate> for IndexTemplateV0_8 {
             priority: index_template.priority,
             description: index_template.description,
             doc_mapping: index_template.doc_mapping,
-            ingest_settings: index_template.ingest_settings,
             indexing_settings: index_template.indexing_settings,
+            ingest_settings: index_template.ingest_settings,
             search_settings: index_template.search_settings,
             retention: index_template.retention_policy_opt,
         }

quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs

@@ -26,7 +26,7 @@ use std::ops::Bound;
 use itertools::Itertools;
 use quickwit_common::pretty::PrettySample;
 use quickwit_config::{
-    DocMapping, IndexingSettings, RetentionPolicy, SearchSettings, SourceConfig,
+    DocMapping, IndexingSettings, IngestSettings, RetentionPolicy, SearchSettings, SourceConfig,
 };
 use quickwit_proto::metastore::{
     AcquireShardsRequest, AcquireShardsResponse, DeleteQuery, DeleteShardsRequest,
@@ -214,24 +214,21 @@ impl FileBackedIndex {
         &self.metadata
     }
 
-    /// Replaces the retention policy in the index config, returning whether a mutation occurred.
-    pub fn set_retention_policy(&mut self, retention_policy_opt: Option<RetentionPolicy>) -> bool {
-        self.metadata.set_retention_policy(retention_policy_opt)
-    }
-
-    /// Replaces the search settings in the index config, returning whether a mutation occurred.
-    pub fn set_search_settings(&mut self, search_settings: SearchSettings) -> bool {
-        self.metadata.set_search_settings(search_settings)
-    }
-
-    /// Replaces the indexing settings in the index config, returning whether a mutation occurred.
-    pub fn set_indexing_settings(&mut self, search_settings: IndexingSettings) -> bool {
-        self.metadata.set_indexing_settings(search_settings)
-    }
-
-    /// Replaces the doc mapping in the index config, returning whether a mutation occurred.
-    pub fn set_doc_mapping(&mut self, doc_mapping: DocMapping) -> bool {
-        self.metadata.set_doc_mapping(doc_mapping)
+    pub fn update_index_config(
+        &mut self,
+        doc_mapping: DocMapping,
+        indexing_settings: IndexingSettings,
+        ingest_settings: IngestSettings,
+        search_settings: SearchSettings,
+        retention_policy_opt: Option<RetentionPolicy>,
+    ) -> bool {
+        self.metadata.update_index_config(
+            doc_mapping,
+            indexing_settings,
+            ingest_settings,
+            search_settings,
+            retention_policy_opt,
+        )
     }
 
     /// Stages a single split.

quickwit/quickwit-metastore/src/metastore/file_backed/mod.rs

@@ -558,19 +558,22 @@ impl MetastoreService for FileBackedMetastore {
         &self,
         request: UpdateIndexRequest,
     ) -> MetastoreResult<IndexMetadataResponse> {
-        let retention_policy_opt = request.deserialize_retention_policy()?;
-        let search_settings = request.deserialize_search_settings()?;
-        let indexing_settings = request.deserialize_indexing_settings()?;
-        let doc_mapping = request.deserialize_doc_mapping()?;
         let index_uid = request.index_uid();
+        let doc_mapping = request.deserialize_doc_mapping()?;
+        let indexing_settings = request.deserialize_indexing_settings()?;
+        let ingest_settings = request.deserialize_ingest_settings()?;
+        let search_settings = request.deserialize_search_settings()?;
+        let retention_policy_opt = request.deserialize_retention_policy()?;
 
         let index_metadata = self
             .mutate(index_uid, |index| {
-                let mut mutation_occurred = index.set_retention_policy(retention_policy_opt);
-                mutation_occurred |= index.set_search_settings(search_settings);
-                mutation_occurred |= index.set_indexing_settings(indexing_settings);
-                mutation_occurred |= index.set_doc_mapping(doc_mapping);
-
+                let mutation_occurred = index.update_index_config(
+                    doc_mapping,
+                    indexing_settings,
+                    ingest_settings,
+                    search_settings,
+                    retention_policy_opt,
+                );
                 let index_metadata = index.metadata().clone();
 
                 if mutation_occurred {