make tonbo build under wasm32 arch

.github/workflows/ci.yml

@@ -84,7 +84,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: run
-          args: --example declare --all-features
+          args: --example declare --features bytes,tokio
 
   benchmark:
     name: Rust benchmark
@@ -109,7 +109,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: bench
-          args: --all-features
+          args: --features bench
 
       - name: Comment on PR using GitHub CLI
         env:

.github/workflows/ci_wasm.yml

@@ -0,0 +1,62 @@
+name: WASM CI
+
+on:
+  push:
+  pull_request:
+  workflow_dispatch:
+
+env:
+  CARGO_TERM_COLOR: always
+  CARGO_REGISTRIES_MY_REGISTRY_INDEX: https://github.com/rust-lang/crates.io-index
+
+
+jobs:
+  check:
+    name: Rust project wasm check
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os:
+          - ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+          components: rustfmt, clippy
+
+      - name: Run cargo clippy
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+
+      - name: Setup for wasm32
+        run: |
+          rustup target add wasm32-unknown-unknown
+
+      - name: Run cargo build
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --target wasm32-unknown-unknown --no-default-features --features aws,bytes,opfs
+
+      - name: Install Chrome Environment
+        run: |
+          mkdir -p /tmp/chrome
+          wget $(curl https://googlechromelabs.github.io/chrome-for-testing/known-good-versions-with-downloads.json | jq -r '.versions | sort_by(.version) | reverse | .[0] | .downloads.chrome | .[] | select(.platform == "linux64") | .url')
+          wget $(curl https://googlechromelabs.github.io/chrome-for-testing/known-good-versions-with-downloads.json | jq -r '.versions | sort_by(.version) | reverse | .[0] | .downloads.chromedriver | .[] | select(.platform == "linux64") | .url')
+          unzip chromedriver-linux64.zip
+          unzip chrome-linux64.zip
+          cp -r chrome-linux64/ /tmp/chrome/
+          cp -r chromedriver-linux64 /tmp/chrome/chromedriver
+
+      - name: Setup wasm-pack
+        run: |
+          cargo install wasm-pack
+
+      - name: Run wasm-pack test
+        run: |
+          export PATH=$PATH:/tmp/chrome/chrome-linux64/:/tmp/chrome/chromedriver-linux64/
+          wasm-pack test --chrome --headless --test wasm --no-default-features --features aws,bytes,opfs

Cargo.toml

@@ -14,15 +14,31 @@ version = "0.2.0"
 msrv = "1.79.0"
 
 [features]
+aws = ["fusio-dispatch/aws", "fusio/aws"]
 bench = ["redb", "rocksdb", "sled"]
 bytes = ["dep:bytes"]
 datafusion = ["dep:async-trait", "dep:datafusion"]
-default = ["bytes", "tokio"]
+default = ["aws", "bytes", "tokio", "tokio-http"]
 load_tbl = []
+object-store = ["fusio/object_store"]
+opfs = [
+    "dep:wasm-bindgen-futures",
+    "fusio-dispatch/opfs",
+    "fusio-parquet/opfs",
+    "fusio/opfs",
+]
 redb = ["dep:redb"]
 rocksdb = ["dep:rocksdb"]
 sled = ["dep:sled"]
-tokio = ["tokio/fs"]
+tokio = [
+    "fusio-dispatch/tokio",
+    "fusio-parquet/tokio",
+    "fusio/tokio",
+    "parquet/default",
+    "tokio/fs",
+]
+tokio-http = ["fusio/tokio-http"]
+wasm = ["aws", "bytes", "opfs"]
 
 [[example]]
 name = "declare"
@@ -58,25 +74,25 @@ crc32fast = "1"
 crossbeam-skiplist = "0.1"
 datafusion = { version = "42", optional = true }
 flume = { version = "0.11", features = ["async"] }
-fusio = { package = "fusio", version = "0.3.3", features = [
-    "aws",
+fusio = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993675591f87dd65b88ffdade31dc0a254b7", package = "fusio", version = "0.3.3", features = [
     "dyn",
     "fs",
-    "object_store",
-    "tokio",
-    "tokio-http",
 ] }
-fusio-dispatch = { package = "fusio-dispatch", version = "0.2.1", features = [
-    "aws",
-    "tokio",
-] }
-fusio-parquet = { package = "fusio-parquet", version = "0.2.1" }
+fusio-dispatch = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993675591f87dd65b88ffdade31dc0a254b7", package = "fusio-dispatch", version = "0.2.1" }
+fusio-parquet = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993675591f87dd65b88ffdade31dc0a254b7", package = "fusio-parquet", version = "0.2.1" }
 futures-core = "0.3"
 futures-io = "0.3"
 futures-util = "0.3"
 lockable = "0.0.8"
 once_cell = "1"
-parquet = { version = "53", features = ["async"] }
+parquet = { version = "53", default-features = false, features = [
+    "async",
+    "base64",
+    "brotli",
+    "flate2",
+    "lz4",
+    "snap",
+] }
 pin-project-lite = "0.2"
 regex = "1"
 thiserror = "1"
@@ -92,18 +108,28 @@ redb = { version = "2", optional = true }
 rocksdb = { version = "0.22", optional = true }
 sled = { version = "0.34", optional = true }
 
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-bindgen = "0.2.95"
+wasm-bindgen-futures = { version = "0.4.45", optional = true }
+
 [dev-dependencies]
 bincode = "1"
-comfy-table = "7"
-criterion = { version = "0.5", features = ["async_tokio", "html_reports"] }
 fastrand = "2"
 futures = { version = "0.3" }
-mimalloc = "0.1"
 serde = "1"
 tempfile = "3"
-tokio = { version = "1", features = ["full"] }
 trybuild = "1.0"
 
+[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
+comfy-table = "7"
+criterion = { version = "0.5", features = ["async_tokio", "html_reports"] }
+mimalloc = "0.1"
+tokio = { version = "1", features = ["full"] }
+
+[target.'cfg(target_arch = "wasm32")'.dev-dependencies]
+wasm-bindgen = "0.2.95"
+wasm-bindgen-test = "0.3.9"
+
 [target.'cfg(unix)'.dev-dependencies]
 pprof = { version = "0.13", features = ["criterion", "flamegraph"] }
 

bindings/python/Cargo.toml

@@ -9,8 +9,11 @@ crate-type = ["cdylib"]
 [workspace]
 
 [dependencies]
-fusio = { package = "fusio", version = "0.3.1", features = ["aws", "tokio"] }
-fusio-dispatch = { package = "fusio-dispatch", version = "0.2.0", features = [
+fusio = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993675591f87dd65b88ffdade31dc0a254b7", package = "fusio", version = "0.3.1", features = [
+    "aws",
+    "tokio",
+] }
+fusio-dispatch = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993675591f87dd65b88ffdade31dc0a254b7", package = "fusio-dispatch", version = "0.2.0", features = [
     "aws",
     "tokio",
 ] }

src/compaction/mod.rs

@@ -503,7 +503,7 @@ where
     EmptyLevel,
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 pub(crate) mod tests {
     use std::sync::{atomic::AtomicU32, Arc};
 

src/executor.rs

@@ -1,15 +1,18 @@
 use std::future::Future;
 
+use fusio::MaybeSend;
+
 pub trait Executor {
     fn spawn<F>(&self, future: F)
     where
-        F: Future<Output = ()> + Send + 'static;
+        F: Future<Output = ()> + MaybeSend + 'static;
 }
 
-#[cfg(any(test, feature = "tokio"))]
+#[cfg(feature = "tokio")]
 pub mod tokio {
     use std::future::Future;
 
+    use fusio::MaybeSend;
     use tokio::runtime::Handle;
 
     use super::Executor;
@@ -36,9 +39,43 @@ pub mod tokio {
     impl Executor for TokioExecutor {
         fn spawn<F>(&self, future: F)
         where
-            F: Future<Output = ()> + Send + 'static,
+            F: Future<Output = ()> + MaybeSend + 'static,
         {
             self.handle.spawn(future);
         }
     }
 }
+
+#[cfg(all(feature = "opfs", target_arch = "wasm32"))]
+pub mod opfs {
+    use std::future::Future;
+
+    use fusio::MaybeSend;
+    use wasm_bindgen::prelude::*;
+
+    use super::Executor;
+
+    #[wasm_bindgen]
+    pub struct OpfsExecutor();
+
+    impl Default for OpfsExecutor {
+        fn default() -> Self {
+            Self {}
+        }
+    }
+
+    impl OpfsExecutor {
+        pub fn new() -> Self {
+            Self {}
+        }
+    }
+
+    impl Executor for OpfsExecutor {
+        fn spawn<F>(&self, future: F)
+        where
+            F: Future<Output = ()> + MaybeSend + 'static,
+        {
+            wasm_bindgen_futures::spawn_local(future);
+        }
+    }
+}

src/inmem/immutable.rs

@@ -212,7 +212,7 @@ where
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 pub(crate) mod tests {
     use std::{mem, sync::Arc};
 

src/inmem/mutable.rs

@@ -204,7 +204,7 @@ where
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 mod tests {
     use std::{ops::Bound, sync::Arc};
 

src/lib.rs

@@ -372,11 +372,10 @@ where
     ) -> impl Stream<Item = Result<T, CommitError<R>>> + 'scan {
         stream! {
             let schema = self.schema.read().await;
-            let manager = &self.manager;
             let current = self.version_set.current().await;
             let mut scan = Scan::new(
                 &schema,
-                manager,
+                &self.manager,
                 range,
                 self.version_set.load_ts(),
                 &*current,
@@ -834,7 +833,7 @@ pub enum Projection {
     Parts(Vec<usize>),
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 pub(crate) mod tests {
     use std::{
         collections::{BTreeMap, Bound},

src/ondisk/sstable.rs

@@ -97,7 +97,7 @@ where
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 pub(crate) mod tests {
     use std::{borrow::Borrow, fs::File, ops::Bound, sync::Arc};
 

src/record/runtime/record.rs

@@ -284,6 +284,7 @@ pub(crate) mod test {
     use super::DynRecord;
     use crate::record::{Column, ColumnDesc, Datatype};
 
+    #[allow(unused)]
     pub(crate) fn test_dyn_item_schema() -> (Vec<ColumnDesc>, usize) {
         let descs = vec![
             ColumnDesc::new("id".to_string(), Datatype::Int64, false),
@@ -298,6 +299,7 @@ pub(crate) mod test {
         (descs, 0)
     }
 
+    #[allow(unused)]
     pub(crate) fn test_dyn_items() -> Vec<DynRecord> {
         let mut items = vec![];
         for i in 0..50 {

src/serdes/arc.rs

@@ -26,7 +26,7 @@ where
 
     async fn encode<W>(&self, writer: &mut W) -> Result<(), Self::Error>
     where
-        W: Write + Send,
+        W: Write,
     {
         self.as_ref().encode(writer).await
     }

src/serdes/mod.rs

@@ -9,12 +9,15 @@ mod string;
 
 use std::future::Future;
 
-use fusio::{SeqRead, Write};
+use fusio::{MaybeSend, SeqRead, Write};
 
 pub trait Encode {
     type Error: From<fusio::Error> + std::error::Error + Send + Sync + 'static;
 
-    fn encode<W>(&self, writer: &mut W) -> impl Future<Output = Result<(), Self::Error>> + Send
+    fn encode<W>(
+        &self,
+        writer: &mut W,
+    ) -> impl Future<Output = Result<(), Self::Error>> + MaybeSend
     where
         W: Write;
 
@@ -62,7 +65,7 @@ mod tests {
 
             async fn encode<W>(&self, writer: &mut W) -> Result<(), Self::Error>
             where
-                W: Write + Send,
+                W: Write,
             {
                 self.0.encode(writer).await?;
 

src/snapshot.rs

@@ -96,7 +96,7 @@ impl<'s, R: Record> Snapshot<'s, R> {
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "tokio"))]
 mod tests {
     use std::{collections::Bound, sync::Arc};