feat: replace std Instant with wasm-compatible wrapper

Cargo.toml

@@ -17,7 +17,7 @@
 
 [workspace]
 exclude = ["datafusion-cli"]
-members = ["datafusion/common", "datafusion/core", "datafusion/expr", "datafusion/execution", "datafusion/functions", "datafusion/functions-array",  "datafusion/optimizer", "datafusion/physical-expr", "datafusion/physical-plan", "datafusion/proto", "datafusion/proto/gen", "datafusion/sql", "datafusion/sqllogictest", "datafusion/substrait", "datafusion/wasmtest", "datafusion-examples", "docs", "test-utils", "benchmarks",
+members = ["datafusion/common", "datafusion/core", "datafusion/expr", "datafusion/execution", "datafusion/functions", "datafusion/functions-array", "datafusion/optimizer", "datafusion/physical-expr", "datafusion/physical-plan", "datafusion/proto", "datafusion/proto/gen", "datafusion/sql", "datafusion/sqllogictest", "datafusion/substrait", "datafusion/wasmtest", "datafusion-examples", "docs", "test-utils", "benchmarks",
 ]
 resolver = "2"
 

datafusion/common/Cargo.toml

@@ -60,5 +60,8 @@ parquet = { workspace = true, optional = true, default-features = true }
 pyo3 = { version = "0.20.0", optional = true }
 sqlparser = { workspace = true }
 
+[target.'cfg(target_family = "wasm")'.dependencies]
+instant = { version = "0.1", features = ["wasm-bindgen"] }
+
 [dev-dependencies]
 rand = "0.8.4"

datafusion/common/src/lib.rs

@@ -41,6 +41,7 @@ pub mod stats;
 pub mod test_util;
 pub mod tree_node;
 pub mod utils;
+pub mod wasm;
 
 /// Reexport arrow crate
 pub use arrow;

datafusion/common/src/wasm.rs

@@ -0,0 +1,27 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! WASM-related utilities
+
+#[cfg(target_family = "wasm")]
+/// DataFusion wrapper around [`std::time::Instant`]. Uses [`instant::Instant`]
+/// under `wasm` feature gate. It provides the same API as [`std::time::Instant`].
+pub type Instant = instant::Instant;
+
+#[cfg(not(target_family = "wasm"))]
+/// DataFusion wrapper around [`std::time::Instant`]. This is only a type alias.
+pub type Instant = std::time::Instant;

datafusion/core/Cargo.toml

@@ -121,6 +121,7 @@ test-utils = { path = "../../test-utils" }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot", "fs"] }
 tokio-postgres = "0.7.7"
+
 [target.'cfg(not(target_os = "windows"))'.dev-dependencies]
 nix = { version = "0.27.1", features = ["fs"] }
 

datafusion/core/benches/parquet_query_sql.rs

@@ -25,6 +25,7 @@ use arrow::datatypes::{
 use arrow::record_batch::RecordBatch;
 use criterion::{criterion_group, criterion_main, Criterion};
 use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_common::wasm::Instant;
 use futures::stream::StreamExt;
 use parquet::arrow::ArrowWriter;
 use parquet::file::properties::{WriterProperties, WriterVersion};
@@ -36,7 +37,6 @@ use std::io::Read;
 use std::ops::Range;
 use std::path::Path;
 use std::sync::Arc;
-use std::time::Instant;
 use tempfile::NamedTempFile;
 
 /// The number of batches to write

datafusion/core/src/datasource/physical_plan/file_stream.rs

@@ -25,7 +25,6 @@ use std::collections::VecDeque;
 use std::mem;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::time::Instant;
 
 use crate::datasource::listing::PartitionedFile;
 use crate::datasource::physical_plan::{
@@ -40,6 +39,7 @@ use crate::physical_plan::RecordBatchStream;
 use arrow::datatypes::SchemaRef;
 use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
+use datafusion_common::wasm::Instant;
 use datafusion_common::ScalarValue;
 
 use futures::future::BoxFuture;

datafusion/optimizer/src/analyzer/mod.rs

@@ -29,14 +29,14 @@ use crate::analyzer::type_coercion::TypeCoercion;
 use crate::utils::log_plan;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{TreeNode, VisitRecursion};
+use datafusion_common::wasm::Instant;
 use datafusion_common::{DataFusionError, Result};
 use datafusion_expr::expr::Exists;
 use datafusion_expr::expr::InSubquery;
 use datafusion_expr::utils::inspect_expr_pre;
 use datafusion_expr::{Expr, LogicalPlan};
 use log::debug;
 use std::sync::Arc;
-use std::time::Instant;
 
 use self::rewrite_expr::OperatorToFunction;
 

datafusion/optimizer/src/optimizer.rs

@@ -19,7 +19,6 @@
 
 use std::collections::HashSet;
 use std::sync::Arc;
-use std::time::Instant;
 
 use crate::common_subexpr_eliminate::CommonSubexprEliminate;
 use crate::decorrelate_predicate_subquery::DecorrelatePredicateSubquery;
@@ -48,6 +47,7 @@ use crate::utils::log_plan;
 
 use datafusion_common::alias::AliasGenerator;
 use datafusion_common::config::ConfigOptions;
+use datafusion_common::wasm::Instant;
 use datafusion_common::{DataFusionError, Result};
 use datafusion_expr::logical_plan::LogicalPlan;
 

datafusion/physical-plan/src/analyze.rs

@@ -17,8 +17,8 @@
 
 //! Defines the ANALYZE operator
 
+use std::any::Any;
 use std::sync::Arc;
-use std::{any::Any, time::Instant};
 
 use super::expressions::PhysicalSortExpr;
 use super::stream::{RecordBatchReceiverStream, RecordBatchStreamAdapter};
@@ -28,6 +28,7 @@ use crate::display::DisplayableExecutionPlan;
 use crate::{DisplayFormatType, ExecutionPlan, Partitioning};
 
 use arrow::{array::StringBuilder, datatypes::SchemaRef, record_batch::RecordBatch};
+use datafusion_common::wasm::Instant;
 use datafusion_common::{internal_err, DataFusionError, Result};
 use datafusion_execution::TaskContext;
 

datafusion/physical-plan/src/metrics/value.rs

@@ -24,12 +24,12 @@ use std::{
         atomic::{AtomicUsize, Ordering},
         Arc,
     },
-    time::{Duration, Instant},
+    time::Duration,
 };
 
-use parking_lot::Mutex;
-
 use chrono::{DateTime, Utc};
+use datafusion_common::wasm::Instant;
+use parking_lot::Mutex;
 
 /// A counter to record things such as number of input or output rows
 ///

datafusion/wasmtest/Cargo.toml

@@ -31,6 +31,8 @@ rust-version = { workspace = true }
 crate-type = ["cdylib", "rlib"]
 
 [dependencies]
+# chrono must be compiled with wasmbind feature
+chrono = { version = "0.4", features = ["wasmbind"] }
 
 # The `console_error_panic_hook` crate provides better debugging of panics by
 # logging them with `console.error`. This is great for development, but requires
@@ -49,5 +51,10 @@ datafusion-sql = { workspace = true }
 
 # getrandom must be compiled with js feature
 getrandom = { version = "0.2.8", features = ["js"] }
+
 parquet = { workspace = true }
 wasm-bindgen = "0.2.87"
+wasm-bindgen-futures = "0.4.40"
+
+[dev-dependencies]
+wasm-bindgen-test = "0.3"

datafusion/wasmtest/README.md

@@ -36,7 +36,7 @@ First, [install wasm-pack](https://rustwasm.github.io/wasm-pack/installer/)
 
 Then use wasm-pack to compile the crate from within this directory
 
-```
+```shell
 wasm-pack build
 ```
 
@@ -46,14 +46,34 @@ The `datafusion-wasm-app` directory contains a simple app (created with [`create
 
 From within the `datafusion/wasmtest/datafusion-wasm-app` directory:
 
-```
+```shell
 npm install
 npm run start
 ```
 
 Then open http://localhost:8080/ in a web browser and check the console to see the results of using various DataFusion crates.
 
-**Note:** In GitHub Actions we test the compilation with `wasm-build`, but we don't currently invoke `datafusion-wasm-app`. In the future we may want to test the behavior of the WASM build using [`wasm-pack test`](https://rustwasm.github.io/wasm-pack/book/tutorials/npm-browser-packages/testing-your-project.html).
+## Test
+
+This crate uses `wasm-pack test` to run tests. Try it out with one of the following commands depending on your browser of choice:
+
+```shell
+wasm-pack test --firefox
+wasm-pack test --chrome
+wasm-pack test --safari
+```
+
+To run the tests in headless mode, add the `--headless` flag:
+
+```shell
+wasm-pack test --headless --firefox
+wasm-pack test --headless --chrome
+wasm-pack test --headless --safari
+```
+
+**Note:** In GitHub Actions we test the compilation with `wasm-build`, but we don't currently invoke `wasm-pack test`. This is because the headless mode is not yet working. Document of adding a GitHub Action job: https://rustwasm.github.io/docs/wasm-bindgen/wasm-bindgen-test/continuous-integration.html#github-actions.
+
+To tweak timeout setting, use `WASM_BINDGEN_TEST_TIMEOUT` environment variable. E.g., `WASM_BINDGEN_TEST_TIMEOUT=300 wasm-pack test --firefox --headless`.
 
 ## Compatibility
 

datafusion/wasmtest/datafusion-wasm-app/index.js

@@ -17,4 +17,5 @@
 
 import * as wasm from "datafusion-wasmtest";
 
-wasm.try_datafusion();
+wasm.basic_exprs();
+wasm.basic_parse();

datafusion/wasmtest/src/lib.rs

@@ -44,7 +44,7 @@ extern "C" {
 }
 
 #[wasm_bindgen]
-pub fn try_datafusion() {
+pub fn basic_exprs() {
     set_panic_hook();
     // Create a scalar value (from datafusion-common)
     let scalar = ScalarValue::from("Hello, World!");
@@ -61,10 +61,67 @@ pub fn try_datafusion() {
         ExprSimplifier::new(SimplifyContext::new(&execution_props).with_schema(schema));
     let simplified_expr = simplifier.simplify(expr).unwrap();
     log(&format!("Simplified Expr: {simplified_expr:?}"));
+}
 
+#[wasm_bindgen]
+pub fn basic_parse() {
     // Parse SQL (using datafusion-sql)
     let sql = "SELECT 2 + 37";
     let dialect = GenericDialect {}; // or AnsiDialect, or your own dialect ...
     let ast = Parser::parse_sql(&dialect, sql).unwrap();
     log(&format!("Parsed SQL: {ast:?}"));
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use datafusion::execution::context::SessionContext;
+    use datafusion_execution::{
+        config::SessionConfig,
+        disk_manager::DiskManagerConfig,
+        runtime_env::{RuntimeConfig, RuntimeEnv},
+    };
+    use datafusion_physical_plan::collect;
+    use datafusion_sql::parser::DFParser;
+    use wasm_bindgen_test::wasm_bindgen_test;
+
+    wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
+
+    #[wasm_bindgen_test]
+    fn datafusion_test() {
+        basic_exprs();
+        basic_parse();
+    }
+
+    #[wasm_bindgen_test]
+    async fn basic_execute() {
+        let sql = "SELECT 2 + 2;";
+
+        // Execute SQL (using datafusion)
+        let rt = Arc::new(
+            RuntimeEnv::new(
+                RuntimeConfig::new().with_disk_manager(DiskManagerConfig::Disabled),
+            )
+            .unwrap(),
+        );
+        let session_config = SessionConfig::new().with_target_partitions(1);
+        let session_context =
+            Arc::new(SessionContext::new_with_config_rt(session_config, rt));
+
+        let statement = DFParser::parse_sql(&sql).unwrap().pop_back().unwrap();
+
+        let logical_plan = session_context
+            .state()
+            .statement_to_plan(statement)
+            .await
+            .unwrap();
+        let data_frame = session_context
+            .execute_logical_plan(logical_plan)
+            .await
+            .unwrap();
+        let physical_plan = data_frame.create_physical_plan().await.unwrap();
+
+        let task_ctx = session_context.task_ctx();
+        let _ = collect(physical_plan, task_ctx).await.unwrap();
+    }
+}