Stable metadata hashing

CHANGELOG.md

@@ -4,6 +4,12 @@ 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).
 
+## [0.32.0] - 2020-10-31
+### Added
+- Support for `flatbuffers` metadata encoding.
+- Using `flatbuffers` format for achieving true stable hashes of metadata.
+- Aligned metadata with ODF `v0.16.0`.
+
 ## [0.31.0] - 2020-10-08
 ### Added
 - `pull` command now supports `--force-uncacheable` flag for refreshing uncacheable datasets.

kamu-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "kamu-cli"
-version = "0.31.0"
+version = "0.32.0"
 description = "Decentralized data management tool"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 license = "MPL-2.0"
@@ -13,6 +13,7 @@ categories = ["command-line-utilities"]
 
 
 [dependencies]
+opendatafabric = { path = "../opendatafabric" }
 kamu = { path = "../kamu-core" }
 
 # UI

kamu-cli/examples/pull-root-multi.rs

@@ -1,6 +1,7 @@
 use kamu::domain::*;
 use kamu_cli::commands::*;
 use kamu_cli::output::OutputConfig;
+use opendatafabric::*;
 
 use chrono::{DateTime, Utc};
 use std::cell::RefCell;
@@ -21,17 +22,11 @@ fn main() {
     cmd.run().unwrap();
 }
 
-fn rand_hash() -> String {
-    use rand::distributions::Standard;
-    use rand::Rng;
-    use std::fmt::Write;
-
-    let mut res = String::with_capacity(64);
-    rand::thread_rng()
-        .sample_iter::<u8, _>(&Standard)
-        .take(32)
-        .for_each(|b| write!(&mut res, "{:02x}", b).unwrap());
-    res
+fn rand_hash() -> Sha3_256 {
+    use rand::RngCore;
+    let mut hash = [0u8; 32];
+    rand::thread_rng().fill_bytes(&mut hash);
+    Sha3_256::new(hash)
 }
 
 pub struct TestPullService;

kamu-cli/examples/pull-root-single.rs

@@ -1,6 +1,7 @@
 use kamu::domain::*;
 use kamu_cli::commands::*;
 use kamu_cli::output::OutputConfig;
+use opendatafabric::*;
 
 use chrono::{DateTime, Utc};
 use std::cell::RefCell;
@@ -66,7 +67,10 @@ impl PullService for TestPullService {
 
         sleep(1000);
 
-        let hash = "a7ffc6f8bf1ed76651c14756a061d662f580ff4de43b49fa82d80a4b80f8434a";
+        let hash =
+            Sha3_256::from_str("a7ffc6f8bf1ed76651c14756a061d662f580ff4de43b49fa82d80a4b80f8434a")
+                .unwrap();
+
         let result = IngestResult::Updated {
             block_hash: hash.to_owned(),
             has_more: false,

kamu-cli/src/commands/add_command.rs

@@ -1,6 +1,6 @@
 use super::{Command, Error};
 use kamu::domain::*;
-use kamu::infra::serde::yaml::*;
+use opendatafabric::*;
 
 use std::cell::RefCell;
 use std::io::Read;

kamu-cli/src/commands/complete_command.rs

@@ -8,7 +8,7 @@ use std::path;
 use std::rc::Rc;
 
 pub struct CompleteCommand {
-    metadata_repo: Rc<RefCell<dyn MetadataRepository>>,
+    metadata_repo: Option<Rc<RefCell<dyn MetadataRepository>>>,
     app: clap::App<'static, 'static>,
     input: String,
     current: usize,
@@ -18,7 +18,7 @@ pub struct CompleteCommand {
 // but we have to do this until clap supports custom completer functions
 impl CompleteCommand {
     pub fn new(
-        metadata_repo: Rc<RefCell<dyn MetadataRepository>>,
+        metadata_repo: Option<Rc<RefCell<dyn MetadataRepository>>>,
         app: clap::App<'static, 'static>,
         input: String,
         current: usize,
@@ -32,9 +32,11 @@ impl CompleteCommand {
     }
 
     fn complete_dataset(&self, prefix: &str) {
-        for dataset_id in self.metadata_repo.borrow().get_all_datasets() {
-            if dataset_id.starts_with(prefix) {
-                println!("{}", dataset_id);
+        if let Some(repo) = self.metadata_repo.as_ref() {
+            for dataset_id in repo.borrow().get_all_datasets() {
+                if dataset_id.starts_with(prefix) {
+                    println!("{}", dataset_id);
+                }
             }
         }
     }

kamu-cli/src/commands/delete_command.rs

@@ -1,5 +1,6 @@
 use super::{Command, Error};
 use kamu::domain::*;
+use opendatafabric::*;
 
 use std::cell::RefCell;
 use std::convert::TryFrom;

kamu-cli/src/commands/depgraph_command.rs

@@ -1,6 +1,6 @@
 use super::{Command, Error};
 use kamu::domain::*;
-use kamu::infra::serde::yaml::DatasetKind;
+use kamu::infra::DatasetKind;
 
 use std::cell::RefCell;
 use std::rc::Rc;

kamu-cli/src/commands/list_command.rs

@@ -1,6 +1,7 @@
 use super::{Command, Error};
 use crate::output::*;
 use kamu::domain::*;
+use opendatafabric::*;
 
 use chrono::{DateTime, Utc};
 use chrono_humanize::HumanTime;

kamu-cli/src/commands/log_command.rs

@@ -1,6 +1,6 @@
 use super::{Command, Error};
 use kamu::domain::*;
-use kamu::infra::serde::yaml::*;
+use opendatafabric::*;
 
 use console::style;
 use std::cell::RefCell;

kamu-cli/src/commands/pull_command.rs

@@ -1,6 +1,7 @@
 use super::{Command, Error};
 use crate::output::OutputConfig;
 use kamu::domain::*;
+use opendatafabric::*;
 
 use std::backtrace::BacktraceStatus;
 use std::cell::RefCell;
@@ -369,7 +370,7 @@ impl IngestListener for PrettyIngestProgress {
             IngestResult::Updated {
                 ref block_hash,
                 has_more: _,
-            } => console::style(format!("Committed new block {}", &block_hash[..8])).green(),
+            } => console::style(format!("Committed new block {}", block_hash.short())).green(),
         };
         self.curr_progress
             .finish_with_message(&Self::spinner_message(
@@ -446,7 +447,7 @@ impl TransformListener for PrettyTransformProgress {
                 console::style("Dataset is up-to-date".to_owned()).yellow()
             }
             TransformResult::Updated { ref block_hash } => {
-                console::style(format!("Committed new block {}", &block_hash[..8])).green()
+                console::style(format!("Committed new block {}", block_hash.short())).green()
             }
         };
         self.curr_progress

kamu-cli/src/commands/set_watermark_command.rs

@@ -1,5 +1,6 @@
 use super::{Command, Error};
 use kamu::domain::*;
+use opendatafabric::*;
 
 use chrono::DateTime;
 use std::cell::RefCell;

kamu-cli/src/main.rs

@@ -1,10 +1,10 @@
 #![feature(backtrace)]
 
-use kamu::domain::*;
 use kamu::infra::*;
 use kamu_cli::cli_parser;
 use kamu_cli::commands::*;
 use kamu_cli::output::*;
+use opendatafabric::DatasetIDBuf;
 
 use clap::value_t_or_exit;
 use console::style;
@@ -67,7 +67,11 @@ fn main() {
             submatches.is_present("recursive"),
         )),
         ("complete", Some(submatches)) => Box::new(CompleteCommand::new(
-            metadata_repo.clone(),
+            if in_workspace(&workspace_layout) {
+                Some(metadata_repo.clone())
+            } else {
+                None
+            },
             cli_parser::cli(BINARY_NAME, VERSION),
             submatches.value_of("input").unwrap().into(),
             submatches.value_of("current").unwrap().parse().unwrap(),
@@ -151,7 +155,7 @@ fn main() {
         _ => unimplemented!(),
     };
 
-    let result = if command.needs_workspace() && !workspace_layout.kamu_root_dir.is_dir() {
+    let result = if command.needs_workspace() && !in_workspace(&workspace_layout) {
         Err(Error::NotInWorkspace)
     } else {
         command.run()
@@ -186,6 +190,10 @@ fn find_workspace_rec(p: &Path) -> Option<WorkspaceLayout> {
     }
 }
 
+fn in_workspace(workspace_layout: &WorkspaceLayout) -> bool {
+    workspace_layout.kamu_root_dir.is_dir()
+}
+
 fn configure_logging(output_config: &OutputConfig, workspace_layout: &WorkspaceLayout) -> Logger {
     let raw_logger = if output_config.verbosity_level > 0 {
         // Log into stderr for verbose output
@@ -226,7 +234,7 @@ fn configure_logging(output_config: &OutputConfig, workspace_layout: &WorkspaceL
 }
 
 fn configure_output_format(matches: &clap::ArgMatches<'_>) -> OutputConfig {
-    let is_tty = console::Term::stdout().is_term();
+    let is_tty = console::Term::stdout().features().is_attended();
 
     let verbosity_level = matches.occurrences_of("v") as u8;
     if verbosity_level > 0 {

kamu-core-test/Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "kamu-test"
-version = "0.31.0"
+version = "0.32.0"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 edition = "2018"
 
 [dependencies]
-kamu = { path = "../kamu-core" }
+opendatafabric = { path = "../opendatafabric" }
 
 chrono = { version = "*", features = ["serde"] }
 url = "*"

kamu-core-test/src/metadata_factory.rs

@@ -1,7 +1,6 @@
-use kamu::domain::*;
-use kamu::infra::serde::yaml::*;
+use opendatafabric::*;
 
-use chrono::{DateTime, SubsecRound, Utc};
+use chrono::{DateTime, Utc};
 use std::convert::TryFrom;
 use std::path::Path;
 
@@ -177,9 +176,9 @@ impl MetadataBlockBuilder {
     fn new() -> Self {
         Self {
             v: MetadataBlock {
-                block_hash: "".to_owned(),
-                prev_block_hash: "".to_owned(),
-                system_time: Utc::now().round_subsecs(3),
+                block_hash: Sha3_256::zero(),
+                prev_block_hash: None,
+                system_time: Utc::now(),
                 output_slice: None,
                 output_watermark: None,
                 input_slices: None,
@@ -188,9 +187,8 @@ impl MetadataBlockBuilder {
         }
     }
 
-    pub fn prev(mut self, prev_block_hash: &str) -> Self {
-        self.v.prev_block_hash.clear();
-        self.v.prev_block_hash.push_str(prev_block_hash);
+    pub fn prev(mut self, prev_block_hash: &Sha3_256) -> Self {
+        self.v.prev_block_hash = Some(*prev_block_hash);
         self
     }
 

kamu-core/Cargo.toml

@@ -1,20 +1,19 @@
 [package]
 name = "kamu"
-version = "0.31.0"
+version = "0.32.0"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 edition = "2018"
 
 [features]
 skip_docker_tests = []
 
 [dependencies]
+opendatafabric = { path = "../opendatafabric" }
+
 # Domain
 chrono = { version = "*", features = ["serde"] }
-intervals-general = "*"
-rust-crypto = "*"  # Data and metadata hashing
 
 # Serialization
-flatbuffers = "*"
 hex = "*"
 serde = { version = "*", features = ["derive"] }
 serde_with = "*"