kamu-data
diff --git a/‎CHANGELOG.md
+2 b/‎CHANGELOG.md
+2
diff --git a/‎Cargo.lock
+1-1 b/‎Cargo.lock
+1-1
diff --git a/‎Makefile
+3 b/‎Makefile
+3
diff --git a/‎src/adapter/graphql/src/mutations/dataset_metadata_mut.rs
+10-23 b/‎src/adapter/graphql/src/mutations/dataset_metadata_mut.rs
+10-23
diff --git a/‎src/adapter/graphql/src/queries/datasets/dataset.rs
+12-16 b/‎src/adapter/graphql/src/queries/datasets/dataset.rs
+12-16
diff --git a/‎src/adapter/graphql/src/queries/datasets/dataset_metadata.rs
+56-49 b/‎src/adapter/graphql/src/queries/datasets/dataset_metadata.rs
+56-49
diff --git a/‎src/app/cli/src/commands/list_command.rs
+15-14 b/‎src/app/cli/src/commands/list_command.rs
+15-14
diff --git a/‎src/domain/core/Cargo.toml
-1 b/‎src/domain/core/Cargo.toml
-1
@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Support `ArrowJson` schema output format in QGL API and CLI commands
 - New `kamu system compact <dataset>` command that compacts dataslices for the given dataset
+### Changed
+- Next batch of optimizations of metadata chain traversal through API using Visitors
 
 ## [0.170.0] - 2024-03-29
 ### Added
 
@@ -101,6 +101,8 @@ codegen-odf-serde:
 		| rustfmt > $(ODF_CRATE_DIR)/src/dtos/dtos_generated.rs
 	python $(ODF_SPEC_DIR)/tools/jsonschema_to_rust_traits.py $(ODF_SPEC_DIR)/schemas \
 		| rustfmt > $(ODF_CRATE_DIR)/src/dtos/dtos_dyntraits_generated.rs
+	python $(ODF_SPEC_DIR)/tools/jsonschema_to_rust_dto_enum_flags.py $(ODF_SPEC_DIR)/schemas \
+		| rustfmt > $(ODF_CRATE_DIR)/src/dtos/dtos_enum_flags_generated.rs
 	python $(ODF_SPEC_DIR)/tools/jsonschema_to_rust_serde_yaml.py $(ODF_SPEC_DIR)/schemas \
 		| rustfmt > $(ODF_CRATE_DIR)/src/serde/yaml/derivations_generated.rs
 	python $(ODF_SPEC_DIR)/tools/jsonschema_to_rust_flatbuffers.py $(ODF_SPEC_DIR)/schemas \
@@ -109,6 +111,7 @@ codegen-odf-serde:
 	$(call add_license_header, "$(ODF_CRATE_DIR)/src/serde/flatbuffers/proxies_generated.rs")
 	$(call add_license_header, "$(ODF_CRATE_DIR)/src/dtos/dtos_generated.rs")
 	$(call add_license_header, "$(ODF_CRATE_DIR)/src/dtos/dtos_dyntraits_generated.rs")
+	$(call add_license_header, "$(ODF_CRATE_DIR)/src/dtos/dtos_enum_flags_generated.rs")
 	$(call add_license_header, "$(ODF_CRATE_DIR)/src/serde/yaml/derivations_generated.rs")
 	$(call add_license_header, "$(ODF_CRATE_DIR)/src/serde/flatbuffers/convertors_generated.rs")
 
 
@@ -7,9 +7,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use kamu_core::{self as domain, MetadataChainExt, TryStreamExtExt};
+use kamu_core::{self as domain, MetadataChainExt, SearchSetAttachmentsVisitor};
 use opendatafabric as odf;
-use opendatafabric::{AsTypedBlock, VariantOf};
 
 use super::{CommitResultAppendError, CommitResultSuccess, NoChanges};
 use crate::mutations::MetadataChainMut;
@@ -38,22 +37,6 @@ impl DatasetMetadataMut {
         Ok(dataset)
     }
 
-    #[graphql(skip)]
-    async fn get_last_block_of_type<T: VariantOf<odf::MetadataEvent>>(
-        &self,
-        ctx: &Context<'_>,
-    ) -> Result<Option<odf::MetadataBlockTyped<T>>> {
-        let dataset = self.get_dataset(ctx).await?;
-        let block = dataset
-            .as_metadata_chain()
-            .iter_blocks_ref(&domain::BlockRef::Head)
-            .filter_map_ok(|(_, b)| b.into_typed::<T>())
-            .try_first()
-            .await
-            .int_err()?;
-        Ok(block)
-    }
-
     /// Access to the mutable metadata chain of the dataset
     async fn chain(&self) -> MetadataChainMut {
         MetadataChainMut::new(self.dataset_handle.clone())
@@ -70,11 +53,15 @@ impl DatasetMetadataMut {
 
         let dataset = self.get_dataset(ctx).await?;
 
-        let old_attachments = self
-            .get_last_block_of_type::<odf::SetAttachments>(ctx)
-            .await?
-            .map(|b| {
-                let odf::Attachments::Embedded(at) = b.event.attachments;
+        let old_attachments = dataset
+            .as_metadata_chain()
+            .accept_one(SearchSetAttachmentsVisitor::new())
+            .await
+            .int_err()?
+            .into_event()
+            .map(|e| {
+                let odf::Attachments::Embedded(at) = e.attachments;
+
                 at
             });
 
 
@@ -8,8 +8,7 @@
 // by the Apache License, Version 2.0.
 
 use chrono::prelude::*;
-use futures::TryStreamExt;
-use kamu_core::{self as domain, MetadataChainExt, ServerUrlConfig, TryStreamExtExt};
+use kamu_core::{self as domain, MetadataChainExt, SearchSeedVisitor, ServerUrlConfig};
 use opendatafabric as odf;
 
 use crate::prelude::*;
@@ -108,29 +107,26 @@ impl Dataset {
     /// Creation time of the first metadata block in the chain
     async fn created_at(&self, ctx: &Context<'_>) -> Result<DateTime<Utc>> {
         let dataset = self.get_dataset(ctx).await?;
-        let seed = dataset
+
+        Ok(dataset
             .as_metadata_chain()
-            .iter_blocks_ref(&domain::BlockRef::Head)
-            .map_ok(|(_, b)| b)
-            .try_last()
+            .accept_one(SearchSeedVisitor::new())
             .await
             .int_err()?
-            .expect("Dataset without blocks");
-        Ok(seed.system_time)
+            .into_block()
+            .expect("Dataset without blocks")
+            .system_time)
     }
 
     /// Creation time of the most recent metadata block in the chain
     async fn last_updated_at(&self, ctx: &Context<'_>) -> Result<DateTime<Utc>> {
         let dataset = self.get_dataset(ctx).await?;
-        let head = dataset
+
+        Ok(dataset
             .as_metadata_chain()
-            .iter_blocks_ref(&domain::BlockRef::Head)
-            .map_ok(|(_, b)| b)
-            .try_first()
-            .await
-            .int_err()?
-            .expect("Dataset without blocks");
-        Ok(head.system_time)
+            .get_block_by_ref(&domain::BlockRef::Head)
+            .await?
+            .system_time)
     }
 
     /// Permissions of the current user
 
@@ -8,9 +8,15 @@
 // by the Apache License, Version 2.0.
 
 use chrono::prelude::*;
-use kamu_core::{self as domain, MetadataChainExt, TryStreamExtExt};
+use kamu_core::{
+    self as domain,
+    MetadataChainExt,
+    SearchSetAttachmentsVisitor,
+    SearchSetInfoVisitor,
+    SearchSetLicenseVisitor,
+    SearchSetVocabVisitor,
+};
 use opendatafabric as odf;
-use opendatafabric::{AsTypedBlock, VariantOf};
 
 use crate::prelude::*;
 use crate::queries::*;
@@ -36,38 +42,22 @@ impl DatasetMetadata {
         Ok(dataset)
     }
 
-    #[graphql(skip)]
-    async fn get_last_block_of_type<T: VariantOf<odf::MetadataEvent>>(
-        &self,
-        ctx: &Context<'_>,
-    ) -> Result<Option<odf::MetadataBlockTyped<T>>> {
-        let dataset = self.get_dataset(ctx).await?;
-        let block = dataset
-            .as_metadata_chain()
-            .iter_blocks_ref(&domain::BlockRef::Head)
-            .filter_map_ok(|(_, b)| b.into_typed::<T>())
-            .try_first()
-            .await
-            .int_err()?;
-        Ok(block)
-    }
-
     /// Access to the temporal metadata chain of the dataset
     async fn chain(&self) -> MetadataChain {
         MetadataChain::new(self.dataset_handle.clone())
     }
 
     /// Last recorded watermark
     async fn current_watermark(&self, ctx: &Context<'_>) -> Result<Option<DateTime<Utc>>> {
-        let ds = self.get_dataset(ctx).await?;
-        Ok(ds
+        let dataset = self.get_dataset(ctx).await?;
+
+        Ok(dataset
             .as_metadata_chain()
-            .iter_blocks_ref(&domain::BlockRef::Head)
-            .filter_data_stream_blocks()
-            .filter_map_ok(|(_, b)| b.event.new_watermark)
-            .try_first()
+            .last_data_block()
             .await
-            .int_err()?)
+            .int_err()?
+            .into_block()
+            .and_then(|b| b.event.new_watermark))
     }
 
     /// Latest data schema
@@ -197,51 +187,68 @@ impl DatasetMetadata {
 
     /// Current descriptive information about the dataset
     async fn current_info(&self, ctx: &Context<'_>) -> Result<SetInfo> {
-        Ok(self
-            .get_last_block_of_type::<odf::SetInfo>(ctx)
-            .await?
+        let dataset = self.get_dataset(ctx).await?;
+
+        Ok(dataset
+            .as_metadata_chain()
+            .accept_one(SearchSetInfoVisitor::new())
+            .await
+            .int_err()?
+            .into_event()
             .map_or(
                 SetInfo {
                     description: None,
                     keywords: None,
                 },
-                |b| b.event.into(),
+                Into::into,
             ))
     }
 
     /// Current readme file as discovered from attachments associated with the
     /// dataset
     async fn current_readme(&self, ctx: &Context<'_>) -> Result<Option<String>> {
-        if let Some(attachments) = self
-            .get_last_block_of_type::<odf::SetAttachments>(ctx)
-            .await?
-        {
-            match attachments.event.attachments {
-                odf::Attachments::Embedded(embedded) => Ok(embedded
-                    .items
+        let dataset = self.get_dataset(ctx).await?;
+
+        Ok(dataset
+            .as_metadata_chain()
+            .accept_one(SearchSetAttachmentsVisitor::new())
+            .await
+            .int_err()?
+            .into_event()
+            .and_then(|e| {
+                let odf::Attachments::Embedded(at) = e.attachments;
+
+                at.items
                     .into_iter()
                     .filter(|i| i.path == "README.md")
                     .map(|i| i.content)
-                    .next()),
-            }
-        } else {
-            Ok(None)
-        }
+                    .next()
+            }))
     }
 
     /// Current license associated with the dataset
     async fn current_license(&self, ctx: &Context<'_>) -> Result<Option<SetLicense>> {
-        Ok(self
-            .get_last_block_of_type::<odf::SetLicense>(ctx)
-            .await?
-            .map(|b| b.event.into()))
+        let dataset = self.get_dataset(ctx).await?;
+
+        Ok(dataset
+            .as_metadata_chain()
+            .accept_one(SearchSetLicenseVisitor::new())
+            .await
+            .int_err()?
+            .into_event()
+            .map(Into::into))
     }
 
     /// Current vocabulary associated with the dataset
     async fn current_vocab(&self, ctx: &Context<'_>) -> Result<Option<SetVocab>> {
-        Ok(self
-            .get_last_block_of_type::<odf::SetVocab>(ctx)
-            .await?
-            .map(|b| b.event.into()))
+        let dataset = self.get_dataset(ctx).await?;
+
+        Ok(dataset
+            .as_metadata_chain()
+            .accept_one(SearchSetVocabVisitor::new())
+            .await
+            .int_err()?
+            .into_event()
+            .map(Into::into))
     }
 }
@@ -258,24 +258,25 @@ impl Command for ListCommand {
             size.push(summary.data_size);
 
             if self.detail_level > 0 {
-                // TODO: Should be precomputed
-                let mut num_blocks = 0;
-                let mut last_watermark: Option<DateTime<Utc>> = None;
-                let mut blocks_stream = dataset.as_metadata_chain().iter_blocks();
-                while let Some((_, b)) = blocks_stream.try_next().await? {
-                    if num_blocks == 0 {
-                        num_blocks = b.sequence_number + 1;
-                    }
-                    if let Some(b) = b.as_data_stream_block() {
-                        last_watermark = b.event.new_watermark.copied();
-                        break;
-                    }
-                }
+                let num_blocks = dataset
+                    .as_metadata_chain()
+                    .get_block(&current_head)
+                    .await
+                    .int_err()?
+                    .sequence_number
+                    + 1;
+                let last_watermark = dataset
+                    .as_metadata_chain()
+                    .last_data_block()
+                    .await
+                    .int_err()?
+                    .into_event()
+                    .and_then(|event| event.new_watermark.map(|t| t.timestamp_micros()));
 
                 id.push(hdl.id.as_did_str().to_string());
                 head.push(current_head.as_multibase().to_string());
                 blocks.push(num_blocks);
-                watermark.push(last_watermark.map(|t| t.timestamp_micros()));
+                watermark.push(last_watermark);
             }
         }
 
 
@@ -29,7 +29,6 @@ opendatafabric = { workspace = true }
 async-stream = { version = "0.3", default-features = false }
 async-trait = { version = "0.1", default-features = false }
 bytes = { version = "1", default-features = false }
-bitflags = { version = "2.4", default-features = false }
 chrono = { version = "0.4", default-features = false }
 dill = { version = "0.8", default-features = false }
 futures = { version = "0.3", default-features = false }