Feature/306 use arrow schema instead of parquet schema in gql and cli (#555)

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: draft

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: formatting

* Use Arrow schema instead of Parquet schema in GQL and CLI #306

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: updated Cargo.lock

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: updated Cargo.lock + corrections

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: removed unused dependency

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: changes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: changes after running clippy and tests

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes for tests

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: updated test group

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixing cargo files

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixing cargo files part 2

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixing cargo files part 3

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixing cargo files part 4

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixing cargo files part 5

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Use Arrow schema instead of Parquet schema in GQL and CLI #306: fixes after code review

* Final tweaks

* Restore schema default format to parquet-json, speed up block iteration

---------

Co-authored-by: Sergii Mikhtoniuk <mikhtoniuk@gmail.com>

Author: Wizzy-wooz

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+### Added
+- Support `ArrowJson` schema output format in QGL API and CLI commands
+
 ## [0.170.0] - 2024-03-29
 ### Added
 - Added GrapqQL Dataset Endpoints object

Cargo.lock

@@ -392,8 +392,9 @@ Shows the dataset schema
 
 * `-o`, `--output-format <FMT>` — Format of an output
 
-  Possible values: `ddl`, `parquet`, `json`
+  Possible values: `ddl`, `parquet`, `parquet-json`, `arrow-json`
 
+* `--from-data-file`
 
 Displays the schema of the dataset. Note that dataset schemas can evolve over time and by default the latest schema will be shown.
 

resources/cli-reference.md

@@ -228,6 +228,7 @@ type DataSchema {
 enum DataSchemaFormat {
 	PARQUET
 	PARQUET_JSON
+	ARROW_JSON
 }
 
 type DataSlice {

resources/schema.gql

@@ -36,10 +36,10 @@ async-graphql = { version = "6", features = ["chrono", "url", "apollo_tracing"]
 async-trait = { version = "0.1", default-features = false }
 cron = { version = "0.12.0", default-features = false }
 chrono = "0.4"
-datafusion = { version = "36", default-features = false } # TODO: Currently needed for type conversions but ideally should be encapsulated by kamu-core
+datafusion = { version = "36", default-features = false, features = ["serde"] } # TODO: Currently needed for type conversions but ideally should be encapsulated by kamu-core
 dill = "0.8"
 futures = "0.3"
-serde = "1"
+serde = {version = "1", default-features = false }
 serde_json = "1"
 tokio = { version = "1", default-features = false, features = [] }
 tokio-stream = { version = "0.1", default-features = false }

src/adapter/graphql/Cargo.toml

@@ -18,7 +18,7 @@ pub(crate) mod mutations;
 pub(crate) mod prelude;
 pub(crate) mod queries;
 mod root;
-pub(crate) mod scalars;
+pub mod scalars;
 pub(crate) mod utils;
 
 pub use root::*;

src/adapter/graphql/src/lib.rs

@@ -76,17 +76,22 @@ impl DatasetMetadata {
         ctx: &Context<'_>,
         format: Option<DataSchemaFormat>,
     ) -> Result<Option<DataSchema>> {
+        // TODO: Default to Arrow eventually
         let format = format.unwrap_or(DataSchemaFormat::Parquet);
 
         let query_svc = from_catalog::<dyn domain::QueryService>(ctx).unwrap();
-        let res_schema = query_svc
+
+        if let Some(schema) = query_svc
             .get_schema(&self.dataset_handle.as_local_ref())
             .await
-            .int_err()?;
-
-        match res_schema {
-            Some(schema) => Ok(Some(DataSchema::from_parquet_schema(&schema, format)?)),
-            None => Ok(None),
+            .int_err()?
+        {
+            Ok(Option::from(DataSchema::from_arrow_schema(
+                schema.as_ref(),
+                format,
+            )))
+        } else {
+            Ok(None)
         }
     }
 

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

@@ -7,7 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use futures::TryStreamExt;
+use futures::{StreamExt, TryStreamExt};
 use kamu_core::{self as domain, MetadataChainExt};
 use opendatafabric as odf;
 
@@ -111,7 +111,6 @@ impl MetadataChain {
 
     // TODO: Add ref parameter (defaulting to "head")
     // TODO: Support before/after style iteration
-    // TODO: PERF: Avoid traversing entire chain
     /// Iterates all metadata blocks in the reverse chronological order
     #[tracing::instrument(level = "info", skip_all)]
     async fn blocks(
@@ -120,32 +119,30 @@ impl MetadataChain {
         page: Option<usize>,
         per_page: Option<usize>,
     ) -> Result<MetadataBlockConnection> {
-        let dataset = self.get_dataset(ctx).await?;
-
         let page = page.unwrap_or(0);
         let per_page = per_page.unwrap_or(Self::DEFAULT_BLOCKS_PER_PAGE);
 
-        let blocks: Vec<_> = dataset
-            .as_metadata_chain()
-            .iter_blocks()
-            .try_collect()
-            .await
-            .int_err()?;
+        let dataset = self.get_dataset(ctx).await?;
+        let chain = dataset.as_metadata_chain();
 
-        let total_count = blocks.len();
+        let head = chain.resolve_ref(&domain::BlockRef::Head).await.int_err()?;
+        let total_count =
+            usize::try_from(chain.get_block(&head).await.int_err()?.sequence_number).unwrap() + 1;
 
-        let nodes: Vec<_> = blocks
-            .into_iter()
+        let mut block_stream = chain
+            .iter_blocks_interval(&head, None, false)
             .skip(page * per_page)
-            .take(per_page)
-            .map(|(hash, block)| {
-                MetadataBlockExtended::new(
-                    hash,
-                    block,
-                    Account::from_dataset_alias(ctx, &self.dataset_handle.alias),
-                )
-            })
-            .collect();
+            .take(per_page);
+
+        let mut nodes = Vec::new();
+        while let Some((hash, block)) = block_stream.try_next().await.int_err()? {
+            let block = MetadataBlockExtended::new(
+                hash,
+                block,
+                Account::from_dataset_alias(ctx, &self.dataset_handle.alias),
+            );
+            nodes.push(block);
+        }
 
         Ok(MetadataBlockConnection::new(
             nodes,