Merge remote-tracking branch 'upstream/main' into feature/make_test_b…

…ench

.github/workflows/rust.yml

@@ -76,6 +76,12 @@ jobs:
 
       - name: Check workspace with all features
         run: cargo check --workspace --benches --features avro,json
+
+      # Ensure that the datafusion crate can be built with only a subset of the function
+      # packages enabled.
+      - name: Check function packages (encoding_expressions)
+        run: cargo check --no-default-features --features=encoding_expressions -p datafusion
+
       - name: Check Cargo.lock for datafusion-cli
         run: |
           # If this test fails, try running `cargo update` in the `datafusion-cli` directory

Cargo.toml

@@ -17,7 +17,7 @@
 
 [workspace]
 exclude = ["datafusion-cli"]
-members = ["datafusion/common", "datafusion/core", "datafusion/expr", "datafusion/execution", "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/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"
 
@@ -49,6 +49,7 @@ datafusion = { path = "datafusion/core", version = "35.0.0" }
 datafusion-common = { path = "datafusion/common", version = "35.0.0" }
 datafusion-execution = { path = "datafusion/execution", version = "35.0.0" }
 datafusion-expr = { path = "datafusion/expr", version = "35.0.0" }
+datafusion-functions = { path = "datafusion/functions", version = "35.0.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "35.0.0" }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "35.0.0" }
 datafusion-physical-plan = { path = "datafusion/physical-plan", version = "35.0.0" }

datafusion-examples/README.md

@@ -44,6 +44,7 @@ cargo run --example csv_sql
 
 - [`avro_sql.rs`](examples/avro_sql.rs): Build and run a query plan from a SQL statement against a local AVRO file
 - [`csv_sql.rs`](examples/csv_sql.rs): Build and run a query plan from a SQL statement against a local CSV file
+- [`csv_sql_streaming.rs`](examples/csv_sql_streaming.rs): Build and run a streaming query plan from a SQL statement against a local CSV file
 - [`catalog.rs`](examples/external_dependency/catalog.rs): Register the table into a custom catalog
 - [`custom_datasource.rs`](examples/custom_datasource.rs): Run queries against a custom datasource (TableProvider)
 - [`dataframe.rs`](examples/dataframe.rs): Run a query using a DataFrame against a local parquet file

datafusion-examples/examples/csv_sql_streaming.rs

@@ -0,0 +1,74 @@
+// 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.
+
+use datafusion::common::test_util::datafusion_test_data;
+use datafusion::error::Result;
+use datafusion::prelude::*;
+
+/// This example demonstrates executing a simple query against an Arrow data source (CSV) and
+/// fetching results with streaming aggregation and streaming window
+#[tokio::main]
+async fn main() -> Result<()> {
+    // create local execution context
+    let ctx = SessionContext::new();
+
+    let testdata = datafusion_test_data();
+
+    // Register a table source and tell DataFusion the file is ordered by `ts ASC`.
+    // Note it is the responsibility of the user to make sure
+    // that file indeed satisfies this condition or else incorrect answers may be produced.
+    let asc = true;
+    let nulls_first = true;
+    let sort_expr = vec![col("ts").sort(asc, nulls_first)];
+    // register csv file with the execution context
+    ctx.register_csv(
+        "ordered_table",
+        &format!("{testdata}/window_1.csv"),
+        CsvReadOptions::new().file_sort_order(vec![sort_expr]),
+    )
+    .await?;
+
+    // execute the query
+    // Following query can be executed with unbounded sources because group by expressions (e.g ts) is
+    // already ordered at the source.
+    //
+    // Unbounded sources means that if the input came from a "never ending" source (such as a FIFO
+    // file on unix) the query could produce results incrementally as data was read.
+    let df = ctx
+        .sql(
+            "SELECT ts, MIN(inc_col), MAX(inc_col) \
+        FROM ordered_table \
+        GROUP BY ts",
+        )
+        .await?;
+
+    df.show().await?;
+
+    // execute the query
+    // Following query can be executed with unbounded sources because window executor can calculate
+    // its result in streaming fashion, because its required ordering is already satisfied at the source.
+    let df = ctx
+        .sql(
+            "SELECT ts, SUM(inc_col) OVER(ORDER BY ts ASC) \
+        FROM ordered_table",
+        )
+        .await?;
+
+    df.show().await?;
+
+    Ok(())
+}

datafusion/common/src/test_util.rs

@@ -144,6 +144,26 @@ macro_rules! assert_not_contains {
     };
 }
 
+/// Returns the datafusion test data directory, which is by default rooted at `datafusion/core/tests/data`.
+///
+/// The default can be overridden by the optional environment
+/// variable `DATAFUSION_TEST_DATA`
+///
+/// panics when the directory can not be found.
+///
+/// Example:
+/// ```
+/// let testdata = datafusion_common::test_util::datafusion_test_data();
+/// let csvdata = format!("{}/window_1.csv", testdata);
+/// assert!(std::path::PathBuf::from(csvdata).exists());
+/// ```
+pub fn datafusion_test_data() -> String {
+    match get_data_dir("DATAFUSION_TEST_DATA", "../../datafusion/core/tests/data") {
+        Ok(pb) => pb.display().to_string(),
+        Err(err) => panic!("failed to get arrow data dir: {err}"),
+    }
+}
+
 /// Returns the arrow test data directory, which is by default stored
 /// in a git submodule rooted at `testing/data`.
 ///

datafusion/core/Cargo.toml

@@ -40,7 +40,7 @@ backtrace = ["datafusion-common/backtrace"]
 compression = ["xz2", "bzip2", "flate2", "zstd", "async-compression", "tokio-util"]
 crypto_expressions = ["datafusion-physical-expr/crypto_expressions", "datafusion-optimizer/crypto_expressions"]
 default = ["crypto_expressions", "encoding_expressions", "regex_expressions", "unicode_expressions", "compression", "parquet"]
-encoding_expressions = ["datafusion-physical-expr/encoding_expressions"]
+encoding_expressions = ["datafusion-functions/encoding_expressions"]
 # Used for testing ONLY: causes all values to hash to the same value (test for collisions)
 force_hash_collisions = []
 parquet = ["datafusion-common/parquet", "dep:parquet"]
@@ -65,6 +65,7 @@ dashmap = { workspace = true }
 datafusion-common = { path = "../common", version = "35.0.0", features = ["object_store"], default-features = false }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
+datafusion-functions = { path = "../functions", version = "35.0.0" }
 datafusion-optimizer = { path = "../optimizer", version = "35.0.0", default-features = false }
 datafusion-physical-expr = { path = "../physical-expr", version = "35.0.0", default-features = false }
 datafusion-physical-plan = { workspace = true }