feat: data utils

implements utilities for inferring schemas and converting CSVs and JSONs
to parquet files

Author: jpopesculian

Cargo.lock

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["archiver", "config", "runner", "template"]
+members = ["archiver", "config", "runner", "template", "data-utils"]
 
 [package]
 name = "aqora"
@@ -29,6 +29,8 @@ aqora-archiver = { path = "archiver", features = [
 aqora-config = { path = "config" }
 aqora-runner = { path = "runner", features = ["clap"] }
 aqora-template = { path = "template" }
+aqora-data-utils = { path = "data-utils" }
+
 async-tempfile = "0.6"
 axum = "0.7"
 base32 = "0.5"
@@ -71,10 +73,7 @@ sentry = { version = "0.32", default-features = false, features = [
   "tracing",
 ] }
 serde = { version = "1.0", features = ["derive"] }
-serde_json = { version = "1.0", features = [
-  "preserve_order",
-  "arbitrary_precision",
-] }
+serde_json = { version = "1.0", features = ["preserve_order"] }
 supports-color = "3.0"
 tempfile = "3.9"
 thiserror = "1.0"
@@ -106,10 +105,17 @@ graphql-ws-client = { version = "0.11", features = [
 ] }
 tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
 qrcode = { version = "0.14.1", default-features = false }
+regex = "1.11.1"
+comfy-table = "7.1.4"
+ron = { version = "0.10", features = ["indexmap"] }
+parquet = { version = "54.3", features = ["snap"] }
 
 [build-dependencies]
 toml_edit = "0.22"
 sentry-types = "0.32"
 
 [dev-dependencies]
 pretty_assertions = "1.4.0"
+
+[patch.crates-io]
+serde_arrow = { git = "https://github.com/aqora-io/serde_arrow.git", branch = "feat-allow-to-strings" }

Cargo.toml

@@ -0,0 +1 @@
+pkg

data-utils/.gitignore

@@ -0,0 +1,56 @@
+[package]
+name = "aqora-data-utils"
+version = "0.13.0"
+edition = "2021"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "rlib"]
+
+[features]
+default = ["all-formats"]
+default-wasm = ["wasm", "console_error_panic_hook", "all-formats"]
+all-formats = ["json", "csv"]
+csv = ["dep:csv", "csv-core"]
+json = ["serde_json"]
+wasm = ["wasm-bindgen", "wasm-bindgen-futures", "web-sys", "js-sys"]
+fs = ["tokio/fs"]
+
+[dependencies]
+wasm-bindgen = { version = "0.2", optional = true }
+wasm-bindgen-futures = { version = "0.4", optional = true, features = [
+  "futures-core-03-stream",
+] }
+js-sys = { version = "0.3", optional = true }
+web-sys = { version = "0.3", optional = true, features = [
+  "console",
+  "Blob",
+  "ReadableStream",
+] }
+console_error_panic_hook = { version = "0.1", optional = true }
+wasm-bindgen-test = { version = "0.3", optional = true }
+
+arrow = { version = "54", default-features = false }
+serde_arrow = { version = "0.13", features = ["arrow-54"] }
+parquet = { version = "54", default-features = false, features = [
+  "arrow",
+  "async",
+] }
+thiserror = "2"
+tokio = { version = "1", default-features = false, features = [
+  "io-util",
+  "rt",
+  "sync",
+  "macros",
+] }
+futures = { version = "0.3", default-features = false, features = ["std"] }
+serde_json = { version = "1.0", optional = true, features = ["preserve_order"] }
+csv = { version = "1.3", optional = true }
+csv-core = { version = "0.1", optional = true }
+bytes = "1.10"
+serde = "1.0"
+pin-project-lite = "0.2"
+regex = "1.11"
+indexmap = { version = "2.9", features = ["serde"] }
+ron = { version = "0.10.1", features = ["indexmap"] }
+chrono = "0.4"

data-utils/Cargo.toml

@@ -0,0 +1,27 @@
+BUILDFLAGS?=
+RUSTFLAGS?=
+
+ifeq ($(RELEASE),)
+	BUILDFLAGS+= --dev
+else
+	BUILDFLAGS+= --release
+	RUSTFLAGS+= -C opt-level=s
+endif
+
+BROWSER?=chrome
+
+.PHONY: wasm
+wasm:
+	RUSTFLAGS="${RUSTFLAGS}" wasm-pack build --target web --out-name index  ${BUILDFLAGS} --features default-wasm
+
+.PHONY: watch
+watch:
+	cargo watch -s "make wasm"
+
+.PHONY: test
+test:
+	wasm-pack test --${BROWSER} --features "default-wasm wasm-bindgen-test"
+
+.PHONY: clean
+clean:
+	rm -rf pkg

data-utils/Makefile

@@ -0,0 +1,202 @@
+pub mod reader;
+pub use reader::{CsvProcessor, CsvReadStream};
+
+use csv_core::{Reader, ReaderBuilder, Terminator};
+use futures::prelude::*;
+use regex::Regex;
+use tokio::io::{self, AsyncRead, AsyncSeek, AsyncSeekExt};
+
+use crate::value::{DateParseOptions, Value, ValueExt};
+use crate::Format;
+
+#[derive(Debug, Clone)]
+pub struct CsvFormat {
+    pub has_headers: bool,
+    pub delimiter: u8,
+    /// None is CLRF
+    pub terminator: Option<u8>,
+    pub quote: u8,
+    pub escape: Option<u8>,
+    pub comment: Option<u8>,
+    pub null_regex: Option<Regex>,
+    pub true_regex: Option<Regex>,
+    pub false_regex: Option<Regex>,
+    pub date_parse: DateParseOptions,
+}
+
+impl CsvFormat {
+    #[inline]
+    pub(crate) fn check_null(&self, s: &str) -> bool {
+        match self.null_regex.as_ref() {
+            Some(r) => r.is_match(s),
+            None => s.is_empty(),
+        }
+    }
+    #[inline]
+    pub(crate) fn check_true(&self, s: &str) -> bool {
+        match self.null_regex.as_ref() {
+            Some(r) => r.is_match(s),
+            None => matches!(s, "true" | "True" | "TRUE"),
+        }
+    }
+    #[inline]
+    pub(crate) fn check_false(&self, s: &str) -> bool {
+        match self.null_regex.as_ref() {
+            Some(r) => r.is_match(s),
+            None => matches!(s, "false" | "False" | "FALSE"),
+        }
+    }
+}
+
+impl Default for CsvFormat {
+    fn default() -> Self {
+        Self {
+            has_headers: false,
+            terminator: None,
+            delimiter: b',',
+            quote: b'"',
+            escape: None,
+            comment: None,
+            null_regex: None,
+            true_regex: None,
+            false_regex: None,
+            date_parse: DateParseOptions::default(),
+        }
+    }
+}
+
+impl From<&CsvFormat> for Reader {
+    fn from(value: &CsvFormat) -> Self {
+        let mut reader = ReaderBuilder::default();
+        reader
+            .delimiter(value.delimiter)
+            .quote(value.quote)
+            .escape(value.escape)
+            .comment(value.comment)
+            .terminator(
+                value
+                    .terminator
+                    .map(Terminator::Any)
+                    .unwrap_or(Terminator::CRLF),
+            );
+        reader.build()
+    }
+}
+
+impl From<CsvFormat> for Format {
+    fn from(value: CsvFormat) -> Self {
+        Self::Csv(value)
+    }
+}
+
+pub async fn read<'a, R>(
+    mut reader: R,
+    options: CsvFormat,
+) -> io::Result<impl Stream<Item = io::Result<Value>> + 'a>
+where
+    R: AsyncRead + AsyncSeek + Unpin + 'a,
+{
+    let headers = if options.has_headers {
+        let headers =
+            CsvReadStream::<_, Vec<String>>::new(&mut reader, CsvProcessor::new(options.clone()))
+                .try_next()
+                .await?;
+        reader.rewind().await?;
+        headers
+    } else {
+        None
+    };
+    let has_headers = headers.is_some();
+    let mut stream =
+        CsvReadStream::<_, Vec<Value>>::new(reader, CsvProcessor::new(options.clone()))
+            .map(move |seq| Ok(Value::Seq(seq?).with_headers(headers.clone())?))
+            .map_ok(move |v| {
+                v.map_values(|v| match v {
+                    Value::String(s) => {
+                        if options.check_null(&s) {
+                            Value::Unit
+                        } else if options.check_true(&s) {
+                            Value::Bool(true)
+                        } else if options.check_false(&s) {
+                            Value::Bool(false)
+                        } else {
+                            Value::String(options.date_parse.normalize(s))
+                        }
+                    }
+                    _ => v,
+                })
+            })
+            .boxed_local();
+    if has_headers {
+        stream = stream.skip(1).boxed_local();
+    }
+    Ok(stream)
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::{format::AsyncFileReader, infer, FormatReader};
+
+    pub fn load_csv(path: &str, format: CsvFormat) -> FormatReader<impl AsyncFileReader> {
+        FormatReader::new(
+            std::io::Cursor::new(std::fs::read(format!("tests/data/csv/{path}.csv")).unwrap()),
+            format.into(),
+        )
+    }
+
+    fn infer_options() -> infer::Options {
+        infer::Options::new().coerce_numbers(true)
+    }
+
+    #[tokio::test]
+    async fn basic_example() {
+        println!(
+            "basic_example: {:#?}",
+            load_csv(
+                "example",
+                CsvFormat {
+                    has_headers: true,
+                    ..Default::default()
+                }
+            )
+            .infer_schema(infer_options(), None)
+            .await
+            .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn example_no_headers() {
+        println!(
+            "example_no_headers: {:#?}",
+            load_csv(
+                "example_no_headers",
+                CsvFormat {
+                    has_headers: false,
+                    ..Default::default()
+                }
+            )
+            .infer_schema(infer_options(), None)
+            .await
+            .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn null_test() {
+        println!(
+            "null_test: {:#?}",
+            load_csv(
+                "null_test",
+                CsvFormat {
+                    has_headers: true,
+                    ..Default::default()
+                }
+            )
+            .infer_schema(infer_options(), None)
+            .await
+            .unwrap()
+        );
+    }
+}