Search enhacements:

- `SearchIndexerConfig` is respected both by natural and full-text search engine
- Ability to disable incremental full-text search indexing via config option

Author: zaychenko-sergei

CHANGELOG.md

@@ -11,6 +11,11 @@ Recommendation: for ease of reading, use the following order:
 - Fixed
 -->
 
+## [Unreleased]
+### Added
+- `SearchIndexerConfig` is respected both by natural and full-text search engine
+- Ability to disable incremental full-text search indexing via config option
+
 ## [0.256.0] - 2026-01-08
 ### Added
 - Prototyped full-text search capabilities via `Elasticsearch`:

resources/di.puml

@@ -245,7 +245,6 @@ class "ModifyAccountPasswordUseCaseImpl"
 class "MqttSourceConfig" <<singleton>>
 class "NaturalLanguageSearchConfig" <<singleton>>
 class "NaturalLanguageSearchImplLazyInit" <<singleton>>
-class "NaturalLanguageSearchIndexerConfig" <<singleton>>
 class "NotebookServerFactory"
 class "OAuthDeviceCodeGeneratorDefault"
 class "OAuthDeviceCodeServiceImpl"
@@ -294,6 +293,7 @@ class "ResetPlannerImpl"
 class "ResourceLoaderImpl"
 class "RunInfoDir" <<singleton>>
 class "SchemaServiceImpl"
+class "SearchIndexerConfig" <<singleton>>
 class "SearchServiceRemoteImpl"
 class "SessionAuthAnonymous"
 class "SessionAuthConfig" <<singleton>>
@@ -652,13 +652,15 @@ class "WsSmartTransferProtocolClient"
 "DidGenerator" <|-- "DidGeneratorDefault"
 "SearchEntitySchemaProvider" <|-- "DatasetSearchSchemaProvider"
 "DatasetSearchSchemaProvider"  --> "Catalog"
+"DatasetSearchSchemaProvider"  --> "SearchIndexerConfig"
 "MessageConsumer" <|-- "DatasetSearchUpdater"
 "MessageConsumerT" <|-- "DatasetSearchUpdater"
 "MessageConsumerT" <|-- "DatasetSearchUpdater"
 "MessageConsumerT" <|-- "DatasetSearchUpdater"
 "DatasetSearchUpdater"  --> "DatasetEntryService"
 "DatasetSearchUpdater"  --> "DatasetRegistry"
 "DatasetSearchUpdater"  --> "SearchService"
+"DatasetSearchUpdater"  --> "SearchIndexerConfig"
 "InitOnStartup" <|-- "DatasetBlockIndexer"
 "DatasetBlockIndexer"  --> "Catalog"
 "MessageConsumer" <|-- "DatasetBlockUpdateHandler"

src/app/cli/src/app.rs

@@ -39,6 +39,7 @@ use tracing::{Instrument, warn};
 
 use crate::accounts::AccountService;
 use crate::cli::Command;
+use crate::config::CLIConfig;
 use crate::error::*;
 use crate::output::*;
 use crate::{
@@ -129,15 +130,15 @@ pub async fn run(workspace_layout: WorkspaceLayout, args: cli::Cli) -> Result<()
             (_, true) => WorkspaceStatus::Created(tenancy_config),
         };
 
-        let config = load_config(&workspace_layout);
+        let cli_config = load_config(&workspace_layout);
         let current_account =
             AccountService::current_account_indication(args.account.clone(), tenancy_config)
                 .map_err(CLIError::usage_error_from)?;
 
         prepare_run_dir(&workspace_layout.run_info_dir);
 
         let app_database_config =
-            get_app_database_config(&workspace_layout, &config, workspace_status);
+            get_app_database_config(&workspace_layout, &cli_config, workspace_status);
         let (database_config, maybe_temp_database_path) = app_database_config.into_inner();
         let maybe_db_connection_settings =
             database_config.as_ref().map(build_db_connection_settings);
@@ -149,6 +150,7 @@ pub async fn run(workspace_layout: WorkspaceLayout, args: cli::Cli) -> Result<()
             let mut base_catalog_builder = configure_base_catalog(
                 &workspace_layout,
                 workspace_status,
+                &cli_config,
                 tenancy_config,
                 args.system_time.map(Into::into),
                 is_e2e_testing,
@@ -171,7 +173,7 @@ pub async fn run(workspace_layout: WorkspaceLayout, args: cli::Cli) -> Result<()
             base_catalog_builder.add_value(Interact::new(args.yes, output_config.is_tty));
 
             register_config_in_catalog(
-                &config,
+                &cli_config,
                 &mut base_catalog_builder,
                 workspace_status,
                 args.password_hashing_mode,
@@ -417,6 +419,7 @@ where
 pub fn configure_base_catalog(
     workspace_layout: &WorkspaceLayout,
     workspace_status: WorkspaceStatus,
+    cli_config: &CLIConfig,
     tenancy_config: TenancyConfig,
     system_time: Option<DateTime<Utc>>,
     is_e2e_testing: bool,
@@ -545,16 +548,36 @@ pub fn configure_base_catalog(
         kamu_adapter_oauth::register_dependencies(&mut b, !is_e2e_testing);
     }
 
+    let incremental_search_indexing = cli_config
+        .search
+        .as_ref()
+        .unwrap()
+        .indexer
+        .as_ref()
+        .unwrap()
+        .incremental_indexing;
+
     kamu_accounts_services::register_dependencies(
         &mut b,
-        workspace_status.is_indexing_needed(),
-        !is_e2e_testing,
+        kamu_accounts_services::AccountDomainDependenciesOptions {
+            needs_indexing: workspace_status.is_indexing_needed(),
+            production: !is_e2e_testing,
+            incremental_search_indexing,
+        },
     );
 
     odf_server::register_dependencies(&mut b, is_e2e_testing);
 
     kamu_adapter_auth_oso_rebac::register_dependencies(&mut b);
-    kamu_datasets_services::register_dependencies(&mut b, workspace_status.is_indexing_needed());
+
+    kamu_datasets_services::register_dependencies(
+        &mut b,
+        kamu_datasets_services::DatasetDomainDependenciesOptions {
+            needs_indexing: workspace_status.is_indexing_needed(),
+            incremental_search_indexing,
+        },
+    );
+
     kamu_auth_rebac_services::register_dependencies(&mut b, workspace_status.is_indexing_needed());
 
     b.add::<DatabaseTransactionRunner>();
@@ -1093,7 +1116,7 @@ pub fn register_config_in_catalog(
     });
 
     let indexer = indexer.unwrap_or_default();
-    catalog_builder.add_value(kamu_search_services::NaturalLanguageSearchIndexerConfig {
+    catalog_builder.add_value(kamu_search::SearchIndexerConfig {
         clear_on_start: indexer.clear_on_start,
         skip_datasets_with_no_description: indexer.skip_datasets_with_no_description,
         skip_datasets_with_no_data: indexer.skip_datasets_with_no_data,

src/app/cli/src/services/config/models.rs

@@ -1140,10 +1140,13 @@ impl Default for SearchConfig {
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 #[skip_serializing_none]
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 #[serde(rename_all = "camelCase")]
 pub struct SearchIndexerConfig {
+    /// Whether incremental indexing is enabled
+    pub incremental_indexing: bool,
+
     /// Whether to clear and re-index on start or use existing vectors if any
     pub clear_on_start: bool,
 
@@ -1159,6 +1162,18 @@ pub struct SearchIndexerConfig {
     pub payload_include_content: bool,
 }
 
+impl Default for SearchIndexerConfig {
+    fn default() -> Self {
+        Self {
+            incremental_indexing: true,
+            clear_on_start: false,
+            skip_datasets_with_no_description: true,
+            skip_datasets_with_no_data: true,
+            payload_include_content: false,
+        }
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 #[skip_serializing_none]

src/app/cli/tests/tests/test_di_graph.rs

@@ -15,6 +15,7 @@ use kamu::domain::{KamuBackgroundCatalog, ServerUrlConfig, TenancyConfig};
 use kamu_accounts::{CurrentAccountSubject, JwtAuthenticationConfig};
 use kamu_adapter_http::AccessToken;
 use kamu_adapter_oauth::GithubAuthenticationConfig;
+use kamu_cli::config::CLIConfig;
 use kamu_cli::services::config::DatabaseConfig;
 use kamu_cli::{self, Interact, OutputConfig, WorkspaceLayout, WorkspaceStatus};
 
@@ -75,6 +76,7 @@ fn test_di_cli_graph_validates(
     let mut base_catalog_builder = kamu_cli::configure_base_catalog(
         &workspace_layout,
         WorkspaceStatus::Created(tenancy_config),
+        &CLIConfig::default(),
         tenancy_config,
         None,
         false,
@@ -118,6 +120,7 @@ fn test_di_server_graph_validates(
     let mut base_catalog_builder = kamu_cli::configure_base_catalog(
         &workspace_layout,
         WorkspaceStatus::Created(tenancy_config),
+        &CLIConfig::default(),
         tenancy_config,
         None,
         false,
@@ -181,6 +184,7 @@ async fn update_di_graph() {
     let mut base_catalog_builder = kamu_cli::configure_base_catalog(
         &workspace_layout,
         WorkspaceStatus::Created(tenancy_config),
+        &CLIConfig::default(),
         tenancy_config,
         None,
         false,

src/app/cli/tests/tests/test_system_info_command.rs

@@ -13,6 +13,7 @@ use std::sync::Arc;
 use container_runtime::ContainerRuntimeConfig;
 use dill::TypedBuilder;
 use kamu::domain::TenancyConfig;
+use kamu_cli::config::CLIConfig;
 use kamu_cli::*;
 use observability::build_info::BuildInfo;
 
@@ -24,6 +25,7 @@ async fn test_system_info_command() {
     let cat = kamu_cli::configure_base_catalog(
         &workspace_layout,
         WorkspaceStatus::NoWorkspace,
+        &CLIConfig::default(),
         TenancyConfig::SingleTenant,
         None,
         false,

src/domain/accounts/services/src/dependencies.rs

@@ -13,9 +13,20 @@ use crate::*;
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
-pub fn register_dependencies(b: &mut CatalogBuilder, needs_indexing: bool, production: bool) {
+#[derive(Clone, Copy)]
+pub struct AccountDomainDependenciesOptions {
+    pub needs_indexing: bool,
+    pub production: bool,
+    pub incremental_search_indexing: bool,
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+pub fn register_dependencies(b: &mut CatalogBuilder, options: AccountDomainDependenciesOptions) {
     b.add::<AccountSearchSchemaProvider>();
-    b.add::<AccountSearchUpdater>();
+    if options.incremental_search_indexing {
+        b.add::<AccountSearchUpdater>();
+    }
 
     b.add::<AccessTokenServiceImpl>();
     b.add::<AccountServiceImpl>();
@@ -32,10 +43,10 @@ pub fn register_dependencies(b: &mut CatalogBuilder, needs_indexing: bool, produ
 
     b.add::<DidSecretService>();
 
-    if needs_indexing {
+    if options.needs_indexing {
         b.add::<OAuthDeviceCodeServiceImpl>();
 
-        if production {
+        if options.production {
             b.add::<OAuthDeviceCodeGeneratorDefault>();
         } else {
             b.add::<PredefinedOAuthDeviceCodeGenerator>();

src/domain/datasets/services/src/dependencies.rs

@@ -7,18 +7,29 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use dill::CatalogBuilder;
-
 use crate::utils::CreateDatasetUseCaseHelper;
 use crate::*;
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
-pub fn register_dependencies(b: &mut CatalogBuilder, needs_indexing: bool) {
+#[derive(Clone, Copy)]
+pub struct DatasetDomainDependenciesOptions {
+    pub needs_indexing: bool,
+    pub incremental_search_indexing: bool,
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+pub fn register_dependencies(
+    b: &mut dill::CatalogBuilder,
+    options: DatasetDomainDependenciesOptions,
+) {
     b.add::<DatasetSearchSchemaProvider>();
-    b.add::<DatasetSearchUpdater>();
+    if options.incremental_search_indexing {
+        b.add::<DatasetSearchUpdater>();
+    }
 
-    if needs_indexing {
+    if options.needs_indexing {
         b.add::<DatasetEntryIndexer>();
         b.add::<DatasetReferenceIndexer>();
         b.add::<DatasetStatisticsIndexer>();

src/domain/datasets/services/src/search/dataset_search_indexer.rs

@@ -17,9 +17,10 @@ use kamu_search::*;
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 pub(crate) async fn index_dataset_from_scratch(
+    indexer_config: &SearchIndexerConfig,
     dataset: ResolvedDataset,
     owner_id: &odf::AccountID,
-) -> Result<serde_json::Value, InternalError> {
+) -> Result<Option<serde_json::Value>, InternalError> {
     // Find key blocks that contribute to search
     let mut attachments_visitor = odf::dataset::SearchSetAttachmentsVisitor::new();
     let mut info_visitor = odf::dataset::SearchSetInfoVisitor::new();
@@ -46,6 +47,17 @@ pub(crate) async fn index_dataset_from_scratch(
             let (description, keywords) = extract_description_and_keywords(info_visitor);
             let attachments = extract_attachment_contents(attachments_visitor);
 
+            // Apply indexing configuration filters
+            if indexer_config.skip_datasets_with_no_data && !schema_fields.is_present() {
+                return Ok(None);
+            }
+            if indexer_config.skip_datasets_with_no_description
+                && !description.is_present()
+                && !attachments.is_present()
+            {
+                return Ok(None);
+            }
+
             // Seed event: created_at
             let seed_event_time = seed_visitor.into_block().unwrap().system_time;
 
@@ -91,7 +103,7 @@ pub(crate) async fn index_dataset_from_scratch(
                 index_doc_mut.insert(fields::ATTACHMENTS.to_string(), serde_json::json!(v));
             }
 
-            Ok(index_doc)
+            Ok(Some(index_doc))
         }
 
         Err(e) => Err(e.int_err()),
@@ -101,11 +113,12 @@ pub(crate) async fn index_dataset_from_scratch(
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 pub(crate) async fn partial_update_for_new_interval(
+    indexer_config: &SearchIndexerConfig,
     dataset: ResolvedDataset,
     owner_id: &odf::AccountID,
     new_head: &odf::Multihash,
     maybe_prev_head: Option<&odf::Multihash>,
-) -> Result<serde_json::Value, InternalError> {
+) -> Result<Option<serde_json::Value>, InternalError> {
     // Extract key blocks that contribute to search
     let mut attachments_visitor = odf::dataset::SearchSetAttachmentsVisitor::new();
     let mut info_visitor = odf::dataset::SearchSetInfoVisitor::new();
@@ -136,7 +149,7 @@ pub(crate) async fn partial_update_for_new_interval(
     match result {
         Ok(_) => {}
         Err(AcceptVisitorError::Traversal(IterBlocksError::InvalidInterval(_))) => {
-            return index_dataset_from_scratch(dataset, owner_id).await;
+            return index_dataset_from_scratch(indexer_config, dataset, owner_id).await;
         }
         Err(e) => return Err(e.int_err()),
     }
@@ -170,7 +183,7 @@ pub(crate) async fn partial_update_for_new_interval(
     insert_search_incremental_update_field(&mut update, fields::KEYWORDS, keywords);
     insert_search_incremental_update_field(&mut update, fields::ATTACHMENTS, attachments);
 
-    Ok(serde_json::Value::Object(update))
+    Ok(Some(serde_json::Value::Object(update)))
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -270,6 +283,7 @@ pub(crate) async fn index_datasets(
     dataset_entry_service: &dyn DatasetEntryService,
     dataset_registry: &dyn DatasetRegistry,
     search_repo: &dyn SearchRepository,
+    indexer_config: &SearchIndexerConfig,
 ) -> Result<usize, InternalError> {
     const BULK_SIZE: usize = 500;
 
@@ -288,7 +302,8 @@ pub(crate) async fn index_datasets(
             .int_err()?;
 
         // Index dataset
-        let dataset_document = index_dataset_from_scratch(dataset, &entry.owner_id).await?;
+        let dataset_document =
+            index_dataset_from_scratch(indexer_config, dataset, &entry.owner_id).await?;
         let dataset_document_json = serde_json::to_value(dataset_document).int_err()?;
 
         tracing::debug!(