Adding webassembly support

Cargo.toml

@@ -1,79 +1,15 @@
-[package]
-name = "kafka-delta-ingest"
-version = "0.2.0"
-authors = ["R. Tyler Croy <rtyler@brokenco.de>", "Christian Williams <christianw@scribd.com>"]
-edition = "2018"
-
-[dependencies]
-anyhow = "1"
-async-trait = "0.1"
-apache-avro = "^0.14"
-base64 = "0.13"
-bytes = "1"
-chrono = "0.4.31"
-clap = { version = "4", features = ["color"] }
-dipstick = "0.9"
-env_logger = "0"
-futures = "0.3"
-jmespatch = { version = "0.3", features = ["sync"] }
-lazy_static = "1"
-log = "0"
-maplit = "1"
-rdkafka = { version = "0.28", features = ["ssl-vendored"] }
-schema_registry_converter = { version = "3.1.0", features = ["easy", "json", "avro"] }
-serde = { version = "1", features = ["derive"] }
-serde_json = "1"
-strum = "0.20"
-strum_macros = "0.20"
-thiserror = "1"
-tokio = { version = "1", features = ["full"] }
-tokio-stream = { version = "0", features = ["fs"] }
-tokio-util = "0.6.3"
-uuid = { version = "0.8", features = ["serde", "v4"] }
-url = "2.3"
-
-#deltalake = { version = "0.16.5", features = ["arrow", "json", "parquet"], optional = true }
-deltalake-core = { git = "https://github.com/delta-io/delta-rs", branch = "main", features = ["json"]}
-deltalake-aws = { git = "https://github.com/delta-io/delta-rs", branch = "main", optional = true }
-deltalake-azure = { git = "https://github.com/delta-io/delta-rs", branch = "main", optional = true }
-
-# s3 feature enabled
-dynamodb_lock = { version = "0.6.0", optional = true }
-rusoto_core = { version = "0.47", default-features = false, features = ["rustls"], optional = true }
-rusoto_credential = { version = "0.47", optional = true }
-rusoto_s3 = { version = "0.47", default-features = false, features = ["rustls"], optional = true }
-
-# sentry
-sentry = { version = "0.23.0", optional = true }
-
-# azure feature enabled, mostly used for tests
-azure_core = { version = "0.18.0", optional = true }
-azure_storage = { version = "0.18.0", optional = true }
-azure_storage_blobs = { version = "0.18.0", optional = true }
-
-[features]
-default = []
-sentry-ext = ["sentry"]
-dynamic-linking = [ "rdkafka/dynamic-linking" ]
-azure = [
-    "deltalake-azure",
-    "azure_core",
-    "azure_storage",
-    "azure_storage_blobs"
-]
-s3 = [
-    "deltalake-aws",
-    "dynamodb_lock",
-    "rusoto_core",
-    "rusoto_credential",
-    "rusoto_s3",
+[workspace]
+members = [
+    "kafka-delta-ingest",
+    "kafka-delta-ingest-wasm",
+    "kafka-delta-ingest-wasm-sdk",
 ]
 
-[dev-dependencies]
-utime = "0.3"
-serial_test = "*"
-tempfile = "3"
-time = "0.3.20"
+resolver = "2"
+
+[profile.release-with-debug]
+inherits = "release"
+debug = true
 
-[profile.release]
-lto = true
+[workspace.dependencies]
+log = "0"

bin/clean-example-data.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-WEB_REQUESTS_DIR=tests/data/web_requests
+WEB_REQUESTS_DIR=kafka-delta-ingest/tests/data/web_requests
 
 find $WEB_REQUESTS_DIR/_delta_log -type f -not -name '00000000000000000000.json' -exec rm {} +
 find $WEB_REQUESTS_DIR -type d -name 'date=*' -exec rm -rf {} +

bin/consume-example-json.sh

@@ -4,7 +4,7 @@ export AWS_ENDPOINT_URL=http://0.0.0.0:4566
 export AWS_ACCESS_KEY_ID=test
 export AWS_SECRET_ACCESS_KEY=test
 
-RUST_LOG=debug ./target/debug/kafka-delta-ingest ingest web_requests ./tests/data/web_requests \
+RUST_LOG=debug ./target/debug/kafka-delta-ingest ingest web_requests ./tests/tests/data/web_requests \
   -l 60 \
   -a web_requests \
   -K "auto.offset.reset=earliest" \

bin/extract-example-json.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 
-tar -xzvf tests/json/web_requests-100K.json.tar.gz -C tests/json
+tar -xzvf kafka-delta-ingest/tests/json/web_requests-100K.json.tar.gz -C kafka-delta-ingest/tests/json
 

docker-compose.yml

@@ -1,18 +1,18 @@
 ---
-version: '3.9'
+version: "3.9"
 services:
   kafka:
     image: docker.redpanda.com/redpandadata/redpanda:v23.1.13
     command:
       - redpanda
       - start
       - --smp
-      - '1'
+      - "1"
       - --reserve-memory
       - 0M
       - --overprovisioned
       - --node-id
-      - '0'
+      - "0"
       - --kafka-addr
       - PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092
       - --advertise-kafka-addr
@@ -41,7 +41,7 @@ services:
       - "8081:8081"
     environment:
       SCHEMA_REGISTRY_HOST_NAME: schema-registry
-      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'redpanda:29092'
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "redpanda:29092"
       SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
 
   localstack:
@@ -55,7 +55,7 @@ services:
       - DOCKER_HOST=unix:///var/run/docker.sock
       - HOST_TMP_FOLDER=${TMPDIR}
     healthcheck:
-      test: [ "CMD", "curl", "-f", "http://localhost:4566/health" ]
+      test: ["CMD", "curl", "-f", "http://localhost:4566/health"]
   azurite:
     image: mcr.microsoft.com/azure-storage/azurite
     ports:
@@ -75,4 +75,4 @@ services:
       - /localstack-setup_emails.sh
     volumes:
       - ./bin/localstack-setup_emails.sh:/localstack-setup_emails.sh
-      - "./tests/data/emails/:/data/emails"
+      - "./kafka-delta-ingest/tests/data/emails/:/data/emails"

kafka-delta-ingest-wasm-sdk/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "kafka-delta-ingest-wasm-sdk"
+version = "0.2.1"
+authors = ["KyJah Keys <kyjah.keys@gmail.com>"]
+rust-version = "1.61"
+description = "WebAssembly for Proxies"
+readme = "README.md"
+license = "Apache-2.0"
+repository = "https://github.com/delta-io/kafka-delta-ingest"
+edition = "2018"
+build = "build.rs"
+
+[lib]
+crate-type = ["cdylib"]
+bench = false
+
+[dependencies]
+hashbrown = "0.14"
+log = { workspace = true }
+
+[profile.release]
+lto = true
+opt-level = 3
+codegen-units = 1
+panic = "abort"
+strip = "debuginfo"

kafka-delta-ingest-wasm-sdk/build.rs

@@ -0,0 +1,19 @@
+fn main() {
+    println!("cargo:rerun-if-changed=build.rs");
+    println!("cargo:rerun-if-env-changed=RUSTFLAGS");
+
+    if let Some(target_os) = std::env::var_os("CARGO_CFG_TARGET_OS") {
+        if target_os != "wasi" {
+            return;
+        }
+    }
+
+    if let Some(rustflags) = std::env::var_os("CARGO_ENCODED_RUSTFLAGS") {
+        for flag in rustflags.to_string_lossy().split('\x1f') {
+            if flag.ends_with("wasi-exec-model=reactor") {
+                println!("cargo:rustc-cfg=wasi_exec_model_reactor");
+                return;
+            }
+        }
+    }
+}

kafka-delta-ingest-wasm-sdk/src/allocator.rs

@@ -0,0 +1,15 @@
+use std::mem::MaybeUninit;
+
+#[cfg_attr(
+    all(target_arch = "wasm32", target_os = "unknown"),
+    export_name = "malloc"
+)]
+#[no_mangle]
+pub extern "C" fn proxy_on_memory_allocate(size: usize) -> *mut u8 {
+    let mut vec: Vec<MaybeUninit<u8>> = Vec::with_capacity(size);
+    unsafe {
+        vec.set_len(size);
+    }
+    let slice = vec.into_boxed_slice();
+    Box::into_raw(slice) as *mut u8
+}