DatasetEntryIndexer: memorization + Updated our crate dependencies …

…so they can be built in isolation (#874)

* DatasetEntryIndexer::concurrent_dataset_handles_processing(): use precalculated mapping

* Update cargo deps

* CHANGELOG.md: update

CHANGELOG.md

@@ -23,6 +23,7 @@ Recommendation: for ease of reading, use the following order:
   - Introducing `DatasetLifecycleMessageRenamed`
 - Simplified error handling code in repositories
 - Hidden part of the test code behind the feature gate
+- Updated our crate dependencies so they can be built in isolation
 
 ## [0.204.4] - 2024-09-30
 ### Changed

src/adapter/flight-sql/Cargo.toml

@@ -36,7 +36,8 @@ uuid = { version = "1", default-features = false }
 
 
 [dev-dependencies]
-kamu-data-utils = { workspace = true }
+kamu-data-utils = { workspace = true, features = ["testing"] }
+
 indoc = "2"
 test-log = { version = "0.2", features = ["trace"] }
 tokio = { version = "1", default-features = false, features = [] }

src/adapter/graphql/Cargo.toml

@@ -64,6 +64,7 @@ uuid = { version = "1", default-features = false }
 # TODO: Limit to mock or in-memory implementations only
 container-runtime = { workspace = true }
 init-on-startup = { workspace = true }
+kamu = { workspace = true, features = ["testing"] }
 kamu-accounts = { workspace = true, features = ["testing"] }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }

src/adapter/http/Cargo.toml

@@ -90,6 +90,7 @@ uuid = { version = "1", default-features = false, features = ["v4"] }
 [dev-dependencies]
 container-runtime = { workspace = true }
 init-on-startup = { workspace = true }
+kamu = { workspace = true, features = ["testing"] }
 kamu-accounts = { workspace = true, features = ["testing"] }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }

src/adapter/odata/Cargo.toml

@@ -43,12 +43,12 @@ tracing = "0.1"
 
 
 [dev-dependencies]
+kamu = { workspace = true, features = ["testing"] }
 messaging-outbox = { workspace = true }
 time-source = { workspace = true }
 
 hyper = { version = "1", default-features = false }
 indoc = { version = "2" }
-kamu = { workspace = true }
 pretty_assertions = { version = "1" }
 reqwest = { version = "0.12", default-features = false }
 tempfile = { version = "3" }

src/app/cli/Cargo.toml

@@ -198,6 +198,7 @@ libc = "0.2" # For getting uid:gid
 
 
 [dev-dependencies]
+kamu = { workspace = true, features = ["testing"] }
 kamu-cli-puppet = { workspace = true, default-features = false, features = [
     "extensions",
 ] }

src/domain/datasets/domain/Cargo.toml

@@ -50,3 +50,4 @@ mockall = { optional = true, version = "0.13", default-features = false }
 
 
 [dev-dependencies]
+mockall = { version = "0.13", default-features = false }

src/domain/datasets/services/src/dataset_entry_indexer.rs

@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use dill::{component, interface, meta};
@@ -83,21 +84,18 @@ impl DatasetEntryIndexer {
     ) -> Result<(), InternalError> {
         let mut join_set = tokio::task::JoinSet::new();
         let now = self.time_source.now();
+        let account_name_id_mapping = self.build_account_name_id_mapping(&dataset_handles).await?;
 
         for dataset_handle in dataset_handles {
-            let task_account_repository = self.account_repository.clone();
-            let task_dataset_entry_repo = self.dataset_entry_repo.clone();
+            let task_owner_account_id =
+                account_name_id_mapping[&dataset_handle.alias.account_name].clone();
             let task_now = now;
+            let task_dataset_entry_repo = self.dataset_entry_repo.clone();
 
             join_set.spawn(async move {
-                let owner_account_id = get_dataset_owner_id(
-                    &task_account_repository,
-                    &dataset_handle.alias.account_name,
-                )
-                .await?;
                 let dataset_entry = DatasetEntry::new(
                     dataset_handle.id,
-                    owner_account_id.clone(),
+                    task_owner_account_id,
                     dataset_handle.alias.dataset_name,
                     task_now,
                 );
@@ -117,6 +115,45 @@ impl DatasetEntryIndexer {
 
         Ok(())
     }
+
+    async fn build_account_name_id_mapping(
+        &self,
+        dataset_handles: &[DatasetHandle],
+    ) -> Result<HashMap<Option<odf::AccountName>, odf::AccountID>, InternalError> {
+        let mut map = HashMap::new();
+
+        for dataset_handle in dataset_handles {
+            let maybe_owner_name = &dataset_handle.alias.account_name;
+
+            if map.contains_key(maybe_owner_name) {
+                continue;
+            }
+
+            let owner_account_id = self.get_dataset_owner_id(maybe_owner_name).await?;
+
+            map.insert(maybe_owner_name.clone(), owner_account_id);
+        }
+
+        Ok(map)
+    }
+
+    async fn get_dataset_owner_id(
+        &self,
+        maybe_owner_name: &Option<odf::AccountName>,
+    ) -> Result<odf::AccountID, InternalError> {
+        match &maybe_owner_name {
+            Some(account_name) => {
+                let account = self
+                    .account_repository
+                    .get_account_by_name(account_name)
+                    .await
+                    .int_err()?;
+
+                Ok(account.id)
+            }
+            None => Ok(DEFAULT_ACCOUNT_ID.clone()),
+        }
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -140,24 +177,3 @@ impl InitOnStartup for DatasetEntryIndexer {
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-// Helpers
-////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-
-async fn get_dataset_owner_id(
-    account_repository: &Arc<dyn AccountRepository>,
-    maybe_owner_name: &Option<odf::AccountName>,
-) -> Result<odf::AccountID, InternalError> {
-    match &maybe_owner_name {
-        Some(account_name) => {
-            let account = account_repository
-                .get_account_by_name(account_name)
-                .await
-                .int_err()?;
-
-            Ok(account.id)
-        }
-        None => Ok(DEFAULT_ACCOUNT_ID.clone()),
-    }
-}
-
-////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

src/domain/flow-system/services/Cargo.toml

@@ -53,7 +53,7 @@ serde_with = { version = "3", default-features = false }
 
 
 [dev-dependencies]
-kamu = { workspace = true }
+kamu = { workspace = true, features = ["testing"] }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }
 kamu-flow-system-inmem = { workspace = true }

src/e2e/app/cli/common/Cargo.toml

@@ -24,15 +24,15 @@ doctest = false
 [dependencies]
 internal-error = { workspace = true }
 kamu-cli-e2e-common-macros = { workspace = true }
-kamu-cli-puppet = { workspace = true, default-features = false }
+kamu-cli-puppet = { workspace = true, default-features = false, features = ["extensions"] }
 opendatafabric = { workspace = true }
 
 async-trait = "0.1"
 chrono = { version = "0.4", default-features = false, features = ["now"] }
 indoc = "2"
 pretty_assertions = "1"
 regex = "1"
-reqwest = { version = "0.12", default-features = false, features = [] }
+reqwest = { version = "0.12", default-features = false, features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 sqlx = { version = "0.8", default-features = false, features = [

src/e2e/app/cli/repo-tests/Cargo.toml

@@ -27,7 +27,7 @@ normal = ["kamu-cli"]
 
 [dependencies]
 internal-error = { workspace = true }
-kamu = { workspace = true }
+kamu = { workspace = true, features = ["testing"] }
 # We add a dependency to ensure kamu-cli is up to date before calling tests
 kamu-cli = { workspace = true }
 kamu-cli-e2e-common = { workspace = true }

src/infra/core/Cargo.toml

@@ -28,7 +28,7 @@ ingest-evm = ["dep:alloy", "dep:datafusion-ethers"]
 ingest-ftp = ["dep:curl", "dep:curl-sys"]
 ingest-mqtt = ["dep:rumqttc"]
 query-extensions-json = ["dep:datafusion-functions-json"]
-testing = ["dep:mockall"]
+testing = ["dep:mockall", "kamu-data-utils/testing"]
 
 
 [dependencies]
@@ -153,6 +153,7 @@ libc = "0.2" # For getting uid:gid
 database-common = { workspace = true }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }
+kamu-data-utils = { workspace = true, features = ["testing"] }
 kamu-datasets-services = { workspace = true }
 
 criterion = { version = "0.5", features = ["async_tokio"] }

src/infra/ingest-datafusion/Cargo.toml

@@ -59,6 +59,8 @@ url = { version = "2", features = ["serde"] }
 
 
 [dev-dependencies]
+kamu-data-utils = { workspace = true, features = ["testing"] }
+
 criterion = { version = "0.5", features = ["async_tokio"] }
 indoc = "2"
 pretty_assertions = "1"

src/utils/data-utils/Cargo.toml

@@ -23,7 +23,7 @@ doctest = false
 
 [features]
 default = []
-testing = []
+testing = ["dep:pretty_assertions"]
 
 
 [dependencies]
@@ -39,17 +39,20 @@ datafusion = { version = "42", default-features = false, features = [
 ] }
 digest = "0.10"
 hex = "0.4"
-pretty_assertions = { version = "1" }
 sha3 = "0.10"
 tracing = { version = "0.1", default-features = false }
 thiserror = { version = "1", default-features = false }
 url = "2"
 serde = { version = "1", default-features = false }
 serde_json = { version = "1" }
 
+# Optional
+pretty_assertions = { optional = true, version = "1" }
+
 
 [dev-dependencies]
 indoc = "2"
+pretty_assertions = { version = "1" }
 test-log = { version = "0.2", features = ["trace"] }
 tokio = { version = "1", default-features = false, features = ["rt", "macros"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

src/utils/database-common-macros/Cargo.toml

@@ -23,4 +23,9 @@ proc-macro = true
 
 [dependencies]
 quote = "1"
-syn = { version = "2", default-features = false }
+syn = { version = "2", default-features = false, features = [
+    "full",
+    "parsing",
+    "printing",
+    "proc-macro",
+] }

src/utils/database-common-macros/src/lib.rs

@@ -235,7 +235,7 @@ impl Parse for CatalogItem2 {
 /// The structure must contain the Catalog as a field
 ///
 /// # Examples
-/// ```
+/// ```compile_fail
 /// #[transactional_method()]
 /// async fn set_system_flow_schedule(&self) {
 ///     // `transaction_catalog` is available inside the method body
@@ -269,7 +269,7 @@ pub fn transactional_method(_attr: TokenStream, item: TokenStream) -> TokenStrea
 /// The structure must contain the Catalog as a field
 ///
 /// # Examples
-/// ```
+/// ```compile_fail
 /// // `service` request from a transactional Catalog
 /// #[transactional_method1(service: Arc<dyn Service>)]
 /// async fn set_system_flow_schedule(&self) {
@@ -308,7 +308,7 @@ pub fn transactional_method1(attr: TokenStream, item: TokenStream) -> TokenStrea
 /// The structure must contain the Catalog as a field
 ///
 /// # Examples
-/// ```
+/// ```compile_fail
 /// // `service` request from a transactional Catalog
 /// #[transactional_method2(service1: Arc<dyn Service1>, service2: Arc<dyn Service2>)]
 /// async fn set_system_flow_schedule(&self) {

src/utils/event-sourcing-macros/Cargo.toml

@@ -22,5 +22,10 @@ proc-macro = true
 
 
 [dependencies]
-syn = { version = "2", default-features = false }
 quote = "1"
+syn = { version = "2", default-features = false, features = [
+    "derive",
+    "parsing",
+    "printing",
+    "proc-macro",
+] }

src/utils/time-source/Cargo.toml

@@ -23,10 +23,15 @@ doctest = false
 
 [dependencies]
 async-trait = { version = "0.1", default-features = false }
-chrono = { version = "0.4", default-features = false }
+chrono = { version = "0.4", default-features = false, features = ["now"] }
 dill = "0.9"
-tokio = { version = "1", default-features = false }
+tokio = { version = "1", default-features = false, features = [
+    "macros",
+    "rt",
+    "sync",
+    "time",
+] }
 
 
 [dev-dependencies]
-futures = { version = "0.3", default-features = false }
+futures = { version = "0.3", default-features = false }