Experiment/dependencies graph (#364)

* Simplest startup job to initialize dependencies graph.
* Removed dependency query from DatasetRepository.
* Integrated dependencies graph into GraphQL queries for upstream/downstream links.
* Integrated dependencies graph into dataset deletion.
* Reacting on `DatasetCreated` events.
* Implemented reaction of dependencies graph on changes in dataset inputs
* Implemented lazy vs eager dependencies initialization

Author: zaychenko-sergei

Cargo.lock

@@ -14,7 +14,7 @@ use std::sync::Arc;
 use dill::Component;
 use event_bus::EventBus;
 use kamu::testing::MetadataFactory;
-use kamu::DatasetRepositoryLocalFs;
+use kamu::{DatasetRepositoryLocalFs, DependencyGraphServiceInMemory};
 use kamu_adapter_auth_oso::{KamuAuthOso, OsoDatasetAuthorizer};
 use kamu_core::auth::{DatasetAction, DatasetActionAuthorizer, DatasetActionUnauthorizedError};
 use kamu_core::{AccessError, CurrentAccountSubject, DatasetRepository};
@@ -111,6 +111,7 @@ impl DatasetAuthorizerHarness {
             )))
             .add::<KamuAuthOso>()
             .add::<OsoDatasetAuthorizer>()
+            .add::<DependencyGraphServiceInMemory>()
             .add_builder(
                 DatasetRepositoryLocalFs::builder()
                     .with_root(datasets_dir)

src/adapter/auth-oso/tests/tests/test_oso_dataset_authorizer.rs

@@ -25,6 +25,7 @@ kamu-data-utils = { workspace = true }
 kamu-core = { workspace = true }
 kamu-task-system = { workspace = true }
 kamu-flow-system  = { workspace = true }
+event-bus = { workspace = true }
 
 async-graphql = { version = "6", features = ["chrono", "url", "apollo_tracing"] }
 async-trait = { version = "0.1", default-features = false }
@@ -36,6 +37,7 @@ indoc = "2"
 serde = "1"
 serde_json = "1"
 tokio = { version = "1", default-features = false, features = [] }
+tokio-stream = { version = "0.1", default-features = false }
 tracing = "0.1"
 thiserror = { version = "1", default-features = false }
 

src/adapter/graphql/Cargo.toml

@@ -10,6 +10,7 @@
 #![feature(error_generic_member_access)]
 #![feature(error_in_core)]
 #![feature(int_roundings)]
+#![feature(async_closure)]
 
 pub mod extensions;
 pub(crate) mod mutations;

src/adapter/graphql/src/lib.rs

@@ -8,7 +8,6 @@
 // by the Apache License, Version 2.0.
 
 use chrono::prelude::*;
-use futures::TryStreamExt;
 use kamu_core::{self as domain, MetadataChainExt, TryStreamExtExt};
 use opendatafabric as odf;
 use opendatafabric::{AsTypedBlock, VariantOf};
@@ -93,39 +92,59 @@ impl DatasetMetadata {
 
     /// Current upstream dependencies of a dataset
     async fn current_upstream_dependencies(&self, ctx: &Context<'_>) -> Result<Vec<Dataset>> {
-        let dataset_repo = from_catalog::<dyn domain::DatasetRepository>(ctx).unwrap();
+        let dependency_graph_service =
+            from_catalog::<dyn domain::DependencyGraphService>(ctx).unwrap();
 
-        let dataset = self.get_dataset(ctx).await?;
-        let summary = dataset
-            .get_summary(domain::GetSummaryOpts::default())
+        use tokio_stream::StreamExt;
+        let upstream_dataset_ids: Vec<_> = dependency_graph_service
+            .get_upstream_dependencies(&self.dataset_handle.id)
             .await
-            .int_err()?;
+            .int_err()?
+            .collect()
+            .await;
 
-        let mut dependencies: Vec<_> = Vec::new();
-        for input in summary.dependencies.into_iter() {
-            let dataset_id = input.id.unwrap().clone();
-            let dataset_handle = dataset_repo
-                .resolve_dataset_ref(&dataset_id.as_local_ref())
+        let dataset_repo = from_catalog::<dyn domain::DatasetRepository>(ctx).unwrap();
+        let mut upstream = Vec::with_capacity(upstream_dataset_ids.len());
+        for upstream_dataset_id in upstream_dataset_ids {
+            let hdl = dataset_repo
+                .resolve_dataset_ref(&upstream_dataset_id.as_local_ref())
                 .await
                 .int_err()?;
-            dependencies.push(Dataset::new(
-                Account::from_dataset_alias(ctx, &dataset_handle.alias),
-                dataset_handle,
+            upstream.push(Dataset::new(
+                Account::from_dataset_alias(ctx, &hdl.alias),
+                hdl,
             ));
         }
-        Ok(dependencies)
+
+        Ok(upstream)
     }
 
     // TODO: Convert to collection
     /// Current downstream dependencies of a dataset
     async fn current_downstream_dependencies(&self, ctx: &Context<'_>) -> Result<Vec<Dataset>> {
-        let dataset_repo = from_catalog::<dyn domain::DatasetRepository>(ctx).unwrap();
+        let dependency_graph_service =
+            from_catalog::<dyn domain::DependencyGraphService>(ctx).unwrap();
 
-        let downstream: Vec<_> = dataset_repo
-            .get_downstream_dependencies(&self.dataset_handle.as_local_ref())
-            .map_ok(|hdl| Dataset::new(Account::from_dataset_alias(ctx, &hdl.alias), hdl))
-            .try_collect()
-            .await?;
+        use tokio_stream::StreamExt;
+        let downstream_dataset_ids: Vec<_> = dependency_graph_service
+            .get_downstream_dependencies(&self.dataset_handle.id)
+            .await
+            .int_err()?
+            .collect()
+            .await;
+
+        let dataset_repo = from_catalog::<dyn domain::DatasetRepository>(ctx).unwrap();
+        let mut downstream = Vec::with_capacity(downstream_dataset_ids.len());
+        for downstream_dataset_id in downstream_dataset_ids {
+            let hdl = dataset_repo
+                .resolve_dataset_ref(&downstream_dataset_id.as_local_ref())
+                .await
+                .int_err()?;
+            downstream.push(Dataset::new(
+                Account::from_dataset_alias(ctx, &hdl.alias),
+                hdl,
+            ));
+        }
 
         Ok(downstream)
     }

src/adapter/graphql/src/queries/datasets/dataset_metadata.rs

@@ -61,6 +61,7 @@ async fn test_internal_error() {
         .add::<EventBus>()
         .add_value(CurrentAccountSubject::new_test())
         .add::<auth::AlwaysHappyDatasetActionAuthorizer>()
+        .add::<DependencyGraphServiceInMemory>()
         .add_builder(
             DatasetRepositoryLocalFs::builder()
                 .with_root(tempdir.path().join("datasets"))

src/adapter/graphql/tests/tests/test_error_handling.rs

@@ -25,6 +25,7 @@ use opendatafabric::*;
 async fn create_catalog_with_local_workspace(tempdir: &Path) -> dill::Catalog {
     dill::CatalogBuilder::new()
         .add::<EventBus>()
+        .add::<DependencyGraphServiceInMemory>()
         .add_builder(
             DatasetRepositoryLocalFs::builder()
                 .with_root(tempdir.join("datasets"))

src/adapter/graphql/tests/tests/test_gql_data.rs

@@ -377,6 +377,7 @@ async fn dataset_rename_name_collision() {
 #[test_log::test(tokio::test)]
 async fn dataset_delete_success() {
     let harness = GraphQLDatasetsHarness::new();
+    harness.init_dependencies_graph().await;
 
     let foo_result = harness
         .create_root_dataset(DatasetName::new_unchecked("foo"))
@@ -426,6 +427,7 @@ async fn dataset_delete_success() {
 #[test_log::test(tokio::test)]
 async fn dataset_delete_dangling_ref() {
     let harness = GraphQLDatasetsHarness::new();
+    harness.init_dependencies_graph().await;
 
     let foo_result = harness
         .create_root_dataset(DatasetName::new_unchecked("foo"))
@@ -531,7 +533,7 @@ async fn dataset_view_permissions() {
 
 struct GraphQLDatasetsHarness {
     _tempdir: tempfile::TempDir,
-    _base_catalog: dill::Catalog,
+    base_catalog: dill::Catalog,
     catalog_authorized: dill::Catalog,
     catalog_anonymous: dill::Catalog,
 }
@@ -544,6 +546,7 @@ impl GraphQLDatasetsHarness {
 
         let base_catalog = dill::CatalogBuilder::new()
             .add::<EventBus>()
+            .add::<DependencyGraphServiceInMemory>()
             .add_builder(
                 DatasetRepositoryLocalFs::builder()
                     .with_root(datasets_dir)
@@ -559,12 +562,27 @@ impl GraphQLDatasetsHarness {
 
         Self {
             _tempdir: tempdir,
-            _base_catalog: base_catalog,
+            base_catalog,
             catalog_anonymous,
             catalog_authorized,
         }
     }
 
+    pub async fn init_dependencies_graph(&self) {
+        let dataset_repo = self
+            .catalog_authorized
+            .get_one::<dyn DatasetRepository>()
+            .unwrap();
+        let dependency_graph_service = self
+            .base_catalog
+            .get_one::<dyn DependencyGraphService>()
+            .unwrap();
+        dependency_graph_service
+            .eager_initialization(&DependencyGraphRepositoryInMemory::new(dataset_repo))
+            .await
+            .unwrap();
+    }
+
     pub async fn create_root_dataset(&self, name: DatasetName) -> CreateDatasetResult {
         let dataset_repo = self
             .catalog_authorized

src/adapter/graphql/tests/tests/test_gql_datasets.rs

@@ -29,6 +29,7 @@ async fn metadata_chain_append_event() {
 
     let base_catalog = dill::CatalogBuilder::new()
         .add::<EventBus>()
+        .add::<DependencyGraphServiceInMemory>()
         .add_builder(
             DatasetRepositoryLocalFs::builder()
                 .with_root(tempdir.path().join("datasets"))
@@ -125,6 +126,7 @@ async fn metadata_update_readme_new() {
 
     let base_catalog = dill::CatalogBuilder::new()
         .add::<EventBus>()
+        .add::<DependencyGraphServiceInMemory>()
         .add_builder(
             DatasetRepositoryLocalFs::builder()
                 .with_root(tempdir.path().join("datasets"))

src/adapter/graphql/tests/tests/test_gql_metadata_chain.rs

@@ -21,6 +21,7 @@ async fn query() {
 
     let cat = dill::CatalogBuilder::new()
         .add::<EventBus>()
+        .add::<DependencyGraphServiceInMemory>()
         .add_value(CurrentAccountSubject::new_test())
         .add::<auth::AlwaysHappyDatasetActionAuthorizer>()
         .add_builder(