Merge branch 'main' into crypto_benchmark

# Conflicts:
#	datafusion/functions/Cargo.toml

Author: getChan

.github/workflows/audit.yml

@@ -42,7 +42,7 @@ jobs:
     steps:
       - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8  # v6.0.1
       - name: Install cargo-audit
-        uses: taiki-e/install-action@50cee16bd6b97b2579572f83cfa1c0a721b1e336  # v2.65.2
+        uses: taiki-e/install-action@de7896b7cd1c7d181266425abbe571b5a8c757bc  # v2.65.3
         with:
           tool: cargo-audit
       - name: Run audit check

.github/workflows/rust.yml

@@ -421,7 +421,7 @@ jobs:
           sudo apt-get update -qq
           sudo apt-get install -y -qq clang
       - name: Setup wasm-pack
-        uses: taiki-e/install-action@50cee16bd6b97b2579572f83cfa1c0a721b1e336  # v2.65.2
+        uses: taiki-e/install-action@de7896b7cd1c7d181266425abbe571b5a8c757bc  # v2.65.3
         with:
           tool: wasm-pack
       - name: Run tests with headless mode
@@ -709,6 +709,23 @@ jobs:
           ./dev/update_function_docs.sh
           git diff --exit-code
 
+  examples-docs-check:
+    name: check example README is up-to-date
+    needs: linux-build-lib
+    runs-on: ubuntu-latest
+    container:
+      image: amd64/rust
+
+    steps:
+      - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+        with:
+          submodules: true
+          fetch-depth: 1
+
+      - name: Run examples docs check script
+        run: |
+          bash ci/scripts/check_examples_docs.sh
+
   # Verify MSRV for the crates which are directly used by other projects:
   # - datafusion
   # - datafusion-substrait
@@ -724,7 +741,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: ./.github/actions/setup-builder
       - name: Install cargo-msrv
-        uses: taiki-e/install-action@50cee16bd6b97b2579572f83cfa1c0a721b1e336  # v2.65.2
+        uses: taiki-e/install-action@de7896b7cd1c7d181266425abbe571b5a8c757bc  # v2.65.3
         with:
           tool: cargo-msrv
 

ci/scripts/check_examples_docs.sh

@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+set -euo pipefail
+
+EXAMPLES_DIR="datafusion-examples/examples"
+README="datafusion-examples/README.md"
+
+# ffi examples are skipped because they were not part of the recent example
+# consolidation work and do not follow the new grouping and execution pattern.
+# They are not documented in the README using the new structure, so including
+# them here would cause false CI failures.
+SKIP_LIST=("ffi")
+
+missing=0
+
+skip() {
+    local value="$1"
+    for item in "${SKIP_LIST[@]}"; do
+        if [[ "$item" == "$value" ]]; then
+            return 0
+        fi
+    done
+    return 1
+}
+
+# collect folder names
+folders=$(find "$EXAMPLES_DIR" -mindepth 1 -maxdepth 1 -type d -exec basename {} \;)
+
+# collect group names from README headers
+groups=$(grep "^### Group:" "$README" | sed -E 's/^### Group: `([^`]+)`.*/\1/')
+
+for folder in $folders; do
+    if skip "$folder"; then
+        echo "Skipped group: $folder"
+        continue
+    fi
+
+    if ! echo "$groups" | grep -qx "$folder"; then
+        echo "Missing README entry for example group: $folder"
+        missing=1
+    fi
+done
+
+if [[ $missing -eq 1 ]]; then
+    echo "README is out of sync with examples"
+    exit 1
+fi

datafusion/catalog/src/cte_worktable.rs

@@ -17,20 +17,18 @@
 
 //! CteWorkTable implementation used for recursive queries
 
+use std::any::Any;
+use std::borrow::Cow;
 use std::sync::Arc;
-use std::{any::Any, borrow::Cow};
 
-use crate::Session;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
-use datafusion_physical_plan::work_table::WorkTableExec;
-
-use datafusion_physical_plan::ExecutionPlan;
-
 use datafusion_common::error::Result;
 use datafusion_expr::{Expr, LogicalPlan, TableProviderFilterPushDown, TableType};
+use datafusion_physical_plan::ExecutionPlan;
+use datafusion_physical_plan::work_table::WorkTableExec;
 
-use crate::TableProvider;
+use crate::{ScanArgs, ScanResult, Session, TableProvider};
 
 /// The temporary working table where the previous iteration of a recursive query is stored
 /// Naming is based on PostgreSQL's implementation.
@@ -85,16 +83,28 @@ impl TableProvider for CteWorkTable {
 
     async fn scan(
         &self,
-        _state: &dyn Session,
-        _projection: Option<&Vec<usize>>,
-        _filters: &[Expr],
-        _limit: Option<usize>,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        filters: &[Expr],
+        limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        // TODO: pushdown filters and limits
-        Ok(Arc::new(WorkTableExec::new(
+        let options = ScanArgs::default()
+            .with_projection(projection.map(|p| p.as_slice()))
+            .with_filters(Some(filters))
+            .with_limit(limit);
+        Ok(self.scan_with_args(state, options).await?.into_inner())
+    }
+
+    async fn scan_with_args<'a>(
+        &self,
+        _state: &dyn Session,
+        args: ScanArgs<'a>,
+    ) -> Result<ScanResult> {
+        Ok(ScanResult::new(Arc::new(WorkTableExec::new(
             self.name.clone(),
             Arc::clone(&self.table_schema),
-        )))
+            args.projection().map(|p| p.to_vec()),
+        )?)))
     }
 
     fn supports_filters_pushdown(

datafusion/common/src/scalar/consts.rs

@@ -17,24 +17,36 @@
 
 // Constants defined for scalar construction.
 
+// Next F16 value above π (upper bound)
+pub(super) const PI_UPPER_F16: half::f16 = half::f16::from_bits(0x4249);
+
 // Next f32 value above π (upper bound)
 pub(super) const PI_UPPER_F32: f32 = std::f32::consts::PI.next_up();
 
 // Next f64 value above π (upper bound)
 pub(super) const PI_UPPER_F64: f64 = std::f64::consts::PI.next_up();
 
+// Next f16 value below -π (lower bound)
+pub(super) const NEGATIVE_PI_LOWER_F16: half::f16 = half::f16::from_bits(0xC249);
+
 // Next f32 value below -π (lower bound)
 pub(super) const NEGATIVE_PI_LOWER_F32: f32 = (-std::f32::consts::PI).next_down();
 
 // Next f64 value below -π (lower bound)
 pub(super) const NEGATIVE_PI_LOWER_F64: f64 = (-std::f64::consts::PI).next_down();
 
+// Next f16 value above π/2 (upper bound)
+pub(super) const FRAC_PI_2_UPPER_F16: half::f16 = half::f16::from_bits(0x3E49);
+
 // Next f32 value above π/2 (upper bound)
 pub(super) const FRAC_PI_2_UPPER_F32: f32 = std::f32::consts::FRAC_PI_2.next_up();
 
 // Next f64 value above π/2 (upper bound)
 pub(super) const FRAC_PI_2_UPPER_F64: f64 = std::f64::consts::FRAC_PI_2.next_up();
 
+// Next f32 value below -π/2 (lower bound)
+pub(super) const NEGATIVE_FRAC_PI_2_LOWER_F16: half::f16 = half::f16::from_bits(0xBE49);
+
 // Next f32 value below -π/2 (lower bound)
 pub(super) const NEGATIVE_FRAC_PI_2_LOWER_F32: f32 =
     (-std::f32::consts::FRAC_PI_2).next_down();

datafusion/common/src/scalar/mod.rs

@@ -1342,8 +1342,7 @@ impl ScalarValue {
     /// Returns a [`ScalarValue`] representing PI's upper bound
     pub fn new_pi_upper(datatype: &DataType) -> Result<ScalarValue> {
         match datatype {
-            // TODO: half::f16 doesn't seem to have equivalent
-            // https://github.com/apache/datafusion/issues/19465
+            DataType::Float16 => Ok(ScalarValue::Float16(Some(consts::PI_UPPER_F16))),
             DataType::Float32 => Ok(ScalarValue::from(consts::PI_UPPER_F32)),
             DataType::Float64 => Ok(ScalarValue::from(consts::PI_UPPER_F64)),
             _ => {
@@ -1355,8 +1354,9 @@ impl ScalarValue {
     /// Returns a [`ScalarValue`] representing -PI's lower bound
     pub fn new_negative_pi_lower(datatype: &DataType) -> Result<ScalarValue> {
         match datatype {
-            // TODO: half::f16 doesn't seem to have equivalent
-            // https://github.com/apache/datafusion/issues/19465
+            DataType::Float16 => {
+                Ok(ScalarValue::Float16(Some(consts::NEGATIVE_PI_LOWER_F16)))
+            }
             DataType::Float32 => Ok(ScalarValue::from(consts::NEGATIVE_PI_LOWER_F32)),
             DataType::Float64 => Ok(ScalarValue::from(consts::NEGATIVE_PI_LOWER_F64)),
             _ => {
@@ -1368,8 +1368,9 @@ impl ScalarValue {
     /// Returns a [`ScalarValue`] representing FRAC_PI_2's upper bound
     pub fn new_frac_pi_2_upper(datatype: &DataType) -> Result<ScalarValue> {
         match datatype {
-            // TODO: half::f16 doesn't seem to have equivalent
-            // https://github.com/apache/datafusion/issues/19465
+            DataType::Float16 => {
+                Ok(ScalarValue::Float16(Some(consts::FRAC_PI_2_UPPER_F16)))
+            }
             DataType::Float32 => Ok(ScalarValue::from(consts::FRAC_PI_2_UPPER_F32)),
             DataType::Float64 => Ok(ScalarValue::from(consts::FRAC_PI_2_UPPER_F64)),
             _ => {
@@ -1381,8 +1382,9 @@ impl ScalarValue {
     // Returns a [`ScalarValue`] representing FRAC_PI_2's lower bound
     pub fn new_neg_frac_pi_2_lower(datatype: &DataType) -> Result<ScalarValue> {
         match datatype {
-            // TODO: half::f16 doesn't seem to have equivalent
-            // https://github.com/apache/datafusion/issues/19465
+            DataType::Float16 => Ok(ScalarValue::Float16(Some(
+                consts::NEGATIVE_FRAC_PI_2_LOWER_F16,
+            ))),
             DataType::Float32 => {
                 Ok(ScalarValue::from(consts::NEGATIVE_FRAC_PI_2_LOWER_F32))
             }

datafusion/core/tests/dataframe/mod.rs

@@ -1112,7 +1112,7 @@ async fn window_using_aggregates() -> Result<()> {
     | -85         | -48      | 6               | -35           | -36    | 83  | -85  | 2  | -43  |
     | -85         | -5       | 4               | -37           | -40    | -5  | -85  | 1  | 83   |
     | -85         | -54      | 15              | -17           | -18    | 83  | -101 | 4  | -38  |
-    | -85         | -56      | 2               | -70           | 57     | -56 | -85  | 1  | -25  |
+    | -85         | -56      | 2               | -70           | -70    | -56 | -85  | 1  | -25  |
     | -85         | -72      | 9               | -43           | -43    | 83  | -85  | 3  | -12  |
     | -85         | -85      | 1               | -85           | -85    | -85 | -85  | 1  | -56  |
     | -85         | 13       | 11              | -17           | -18    | 83  | -85  | 3  | 14   |

datafusion/core/tests/execution/coop.rs

@@ -64,13 +64,14 @@ use std::time::Duration;
 use tokio::runtime::{Handle, Runtime};
 use tokio::select;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 struct RangeBatchGenerator {
     schema: SchemaRef,
     value_range: Range<i64>,
     boundedness: Boundedness,
     batch_size: usize,
     poll_count: usize,
+    original_range: Range<i64>,
 }
 
 impl std::fmt::Display for RangeBatchGenerator {
@@ -110,6 +111,13 @@ impl LazyBatchGenerator for RangeBatchGenerator {
             RecordBatch::try_new(Arc::clone(&self.schema), vec![Arc::new(array)])?;
         Ok(Some(batch))
     }
+
+    fn reset_state(&self) -> Arc<RwLock<dyn LazyBatchGenerator>> {
+        let mut new = self.clone();
+        new.poll_count = 0;
+        new.value_range = new.original_range.clone();
+        Arc::new(RwLock::new(new))
+    }
 }
 
 fn make_lazy_exec(column_name: &str, pretend_infinite: bool) -> LazyMemoryExec {
@@ -139,9 +147,10 @@ fn make_lazy_exec_with_range(
     let batch_gen = RangeBatchGenerator {
         schema: Arc::clone(&schema),
         boundedness,
-        value_range: range,
+        value_range: range.clone(),
         batch_size: 8192,
         poll_count: 0,
+        original_range: range,
     };
 
     // Wrap the generator in a trait object behind Arc<RwLock<_>>