CompactionService decoupled from DatasetRepository

kamu-data · Oct 11, 2024 · 9cf7a83 · 9cf7a83
1 parent 7053f4a
commit 9cf7a83
Show file tree

Hide file tree

Showing 12 changed files with 107 additions and 75 deletions.
diff --git a/src/adapter/http/tests/tests/tests_pull/scenarios/scenario_existing_diverged_dataset.rs b/src/adapter/http/tests/tests/tests_pull/scenarios/scenario_existing_diverged_dataset.rs
@@ -91,10 +91,14 @@ impl<TServerHarness: ServerSideHarness> SmartPullExistingDivergedDatasetScenario
         )
         .await;
 
+        let server_dataset = server_harness
+            .cli_dataset_repository()
+            .get_dataset_by_handle(&server_create_result.dataset_handle);
+
         let compaction_service = server_harness.cli_compaction_service();
         let server_compaction_result = compaction_service
             .compact_dataset(
-                &server_create_result.dataset_handle,
+                ResolvedDataset::new(server_dataset, server_create_result.dataset_handle.clone()),
                 CompactionOptions::default(),
                 None,
             )

diff --git a/src/adapter/http/tests/tests/tests_push/scenarios/scenario_existing_diverged_dataset.rs b/src/adapter/http/tests/tests/tests_push/scenarios/scenario_existing_diverged_dataset.rs
@@ -93,11 +93,15 @@ impl<TServerHarness: ServerSideHarness> SmartPushExistingDivergedDatasetScenario
         )
         .await;
 
+        let client_dataset = client_harness
+            .dataset_repository()
+            .get_dataset_by_handle(&client_create_result.dataset_handle);
+
         // Compact at client side
         let compaction_service = client_harness.compaction_service();
         let client_compaction_result = compaction_service
             .compact_dataset(
-                &client_create_result.dataset_handle,
+                ResolvedDataset::new(client_dataset, client_create_result.dataset_handle.clone()),
                 CompactionOptions::default(),
                 None,
             )

diff --git a/src/domain/core/src/entities/mod.rs b/src/domain/core/src/entities/mod.rs
@@ -12,8 +12,10 @@ pub mod dataset_summary;
 pub mod engine;
 pub mod metadata_chain;
 pub mod metadata_stream;
+pub mod resolved_dataset;
 
 pub use dataset::*;
 pub use dataset_summary::*;
 pub use metadata_chain::*;
 pub use metadata_stream::*;
+pub use resolved_dataset::*;
diff --git a/src/domain/core/src/entities/resolved_dataset.rs b/src/domain/core/src/entities/resolved_dataset.rs
@@ -0,0 +1,29 @@
+// Copyright Kamu Data, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use std::sync::Arc;
+
+use opendatafabric::DatasetHandle;
+
+use super::Dataset;
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+pub struct ResolvedDataset {
+    pub dataset: Arc<dyn Dataset>,
+    pub handle: DatasetHandle,
+}
+
+impl ResolvedDataset {
+    pub fn new(dataset: Arc<dyn Dataset>, handle: DatasetHandle) -> Self {
+        Self { dataset, handle }
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/src/domain/core/src/services/compaction_service.rs b/src/domain/core/src/services/compaction_service.rs
@@ -27,7 +27,7 @@ pub const DEFAULT_MAX_SLICE_RECORDS: u64 = 10_000;
 pub trait CompactionService: Send + Sync {
     async fn compact_dataset(
         &self,
-        dataset_handle: &DatasetHandle,
+        target: ResolvedDataset,
         options: CompactionOptions,
         listener: Option<Arc<dyn CompactionListener>>,
     ) -> Result<CompactionResult, CompactionError>;
@@ -114,9 +114,9 @@ impl From<SetRefError> for CompactionError {
 }
 
 #[derive(Error, Debug)]
-#[error("Dataset {dataset_name} in not root kind")]
+#[error("Dataset '{dataset_alias}' in not root kind")]
 pub struct InvalidDatasetKindError {
-    pub dataset_name: DatasetName,
+    pub dataset_alias: DatasetAlias,
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

diff --git a/src/domain/task-system/services/src/task_logical_plan_runner_impl.rs b/src/domain/task-system/services/src/task_logical_plan_runner_impl.rs
@@ -23,6 +23,7 @@ use kamu_core::{
     PullService,
     ResetError,
     ResetService,
+    ResolvedDataset,
     TransformError,
 };
 use kamu_datasets::{DatasetEnvVar, DatasetEnvVarService};
@@ -137,14 +138,18 @@ impl TaskLogicalPlanRunnerImpl {
     ) -> Result<TaskOutcome, InternalError> {
         let compaction_svc = self.catalog.get_one::<dyn CompactionService>().int_err()?;
         let dataset_repo = self.catalog.get_one::<dyn DatasetRepository>().int_err()?;
+
         let dataset_handle = dataset_repo
             .resolve_dataset_ref(&args.dataset_id.as_local_ref())
             .await
             .int_err()?;
 
         let compaction_result = compaction_svc
             .compact_dataset(
-                &dataset_handle,
+                ResolvedDataset::new(
+                    dataset_repo.get_dataset_by_handle(&dataset_handle),
+                    dataset_handle.clone(),
+                ),
                 CompactionOptions {
                     max_slice_size: args.max_slice_size,
                     max_slice_records: args.max_slice_records,

diff --git a/src/infra/core/src/compaction_service_impl.rs b/src/infra/core/src/compaction_service_impl.rs
@@ -32,7 +32,6 @@ use internal_error::ResultIntoInternal;
 use kamu_core::*;
 use opendatafabric::{
     Checkpoint,
-    DatasetHandle,
     DatasetKind,
     DatasetVocabulary,
     MetadataEvent,
@@ -50,7 +49,6 @@ use crate::*;
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 pub struct CompactionServiceImpl {
-    dataset_repo: Arc<dyn DatasetRepository>,
     object_store_registry: Arc<dyn ObjectStoreRegistry>,
     time_source: Arc<dyn SystemTimeSource>,
     run_info_dir: Arc<RunInfoDir>,
@@ -102,13 +100,11 @@ struct ChainFilesInfo {
 #[interface(dyn CompactionService)]
 impl CompactionServiceImpl {
     pub fn new(
-        dataset_repo: Arc<dyn DatasetRepository>,
         object_store_registry: Arc<dyn ObjectStoreRegistry>,
         time_source: Arc<dyn SystemTimeSource>,
         run_info_dir: Arc<RunInfoDir>,
     ) -> Self {
         Self {
-            dataset_repo,
             object_store_registry,
             time_source,
             run_info_dir,
@@ -463,13 +459,12 @@ impl CompactionService for CompactionServiceImpl {
     #[tracing::instrument(level = "info", skip_all)]
     async fn compact_dataset(
         &self,
-        dataset_handle: &DatasetHandle,
+        target: ResolvedDataset,
         options: CompactionOptions,
         maybe_listener: Option<Arc<dyn CompactionListener>>,
     ) -> Result<CompactionResult, CompactionError> {
-        let dataset = self.dataset_repo.get_dataset_by_handle(dataset_handle);
-
-        let dataset_kind = dataset
+        let dataset_kind = target
+            .dataset
             .get_summary(GetSummaryOpts::default())
             .await
             .int_err()?
@@ -478,7 +473,7 @@ impl CompactionService for CompactionServiceImpl {
         if !options.keep_metadata_only && dataset_kind != DatasetKind::Root {
             return Err(CompactionError::InvalidDatasetKind(
                 InvalidDatasetKindError {
-                    dataset_name: dataset_handle.alias.dataset_name.clone(),
+                    dataset_alias: target.handle.alias.clone(),
                 },
             ));
         }
@@ -492,7 +487,7 @@ impl CompactionService for CompactionServiceImpl {
 
         match self
             .compact_dataset_impl(
-                dataset,
+                target.dataset,
                 max_slice_size,
                 max_slice_records,
                 options.keep_metadata_only,

diff --git a/src/infra/core/src/transform_service_impl.rs b/src/infra/core/src/transform_service_impl.rs
@@ -710,7 +710,10 @@ impl TransformServiceImpl {
                 let compaction_result = self
                     .compaction_svc
                     .compact_dataset(
-                        &dataset_handle,
+                        ResolvedDataset::new(
+                            self.dataset_repo.get_dataset_by_handle(&dataset_handle),
+                            dataset_handle.clone(),
+                        ),
                         CompactionOptions {
                             keep_metadata_only: true,
                             ..Default::default()

diff --git a/src/infra/core/src/use_cases/compact_dataset_use_case_impl.rs b/src/infra/core/src/use_cases/compact_dataset_use_case_impl.rs
@@ -20,7 +20,9 @@ use kamu_core::{
     CompactionResponse,
     CompactionResult,
     CompactionService,
+    DatasetRegistry,
     NullCompactionMultiListener,
+    ResolvedDataset,
 };
 use opendatafabric::DatasetHandle;
 
@@ -30,16 +32,19 @@ use opendatafabric::DatasetHandle;
 #[interface(dyn CompactDatasetUseCase)]
 pub struct CompactDatasetUseCaseImpl {
     compaction_service: Arc<dyn CompactionService>,
+    dataset_registry: Arc<dyn DatasetRegistry>,
     dataset_action_authorizer: Arc<dyn DatasetActionAuthorizer>,
 }
 
 impl CompactDatasetUseCaseImpl {
     pub fn new(
         compaction_service: Arc<dyn CompactionService>,
+        dataset_registry: Arc<dyn DatasetRegistry>,
         dataset_action_authorizer: Arc<dyn DatasetActionAuthorizer>,
     ) -> Self {
         Self {
             compaction_service,
+            dataset_registry,
             dataset_action_authorizer,
         }
     }
@@ -58,9 +63,16 @@ impl CompactDatasetUseCase for CompactDatasetUseCaseImpl {
             .check_action_allowed(dataset_handle, DatasetAction::Write)
             .await?;
 
+        // Resolve dataset
+        let dataset = self.dataset_registry.get_dataset_by_handle(dataset_handle);
+
         // Actual action
         self.compaction_service
-            .compact_dataset(dataset_handle, options, maybe_listener)
+            .compact_dataset(
+                ResolvedDataset::new(dataset, dataset_handle.clone()),
+                options,
+                maybe_listener,
+            )
             .await
     }
 

diff --git a/src/infra/core/tests/tests/engine/test_engine_io.rs b/src/infra/core/tests/tests/engine/test_engine_io.rs
@@ -69,7 +69,6 @@ async fn test_engine_io_common<
         engine_provisioner.clone(),
         Arc::new(SystemTimeSourceDefault),
         Arc::new(CompactionServiceImpl::new(
-            dataset_repo.clone(),
             object_store_registry.clone(),
             time_source.clone(),
             run_info_dir.clone(),