This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/asf-site by this push:
new 670607b676 Publish built docs triggered by
935fb3e2fd8cf3574af6b7ddecfece2c7e515f56
670607b676 is described below
commit 670607b6760eaf9b749d74164e7c0bb639591369
Author: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 30 14:01:51 2025 +0000
Publish built docs triggered by 935fb3e2fd8cf3574af6b7ddecfece2c7e515f56
---
.../1e336e540124589d1e27210fa4f805f6/rust_lint.sh | 8 +
_sources/library-user-guide/catalogs.md.txt | 74 +++
.../custom-table-providers.md.txt | 274 +++++++++-
.../library-user-guide/extending-operators.md.txt | 6 +-
.../functions/adding-udfs.md.txt | 568 ++++++++++++++++++++-
_sources/library-user-guide/query-optimizer.md.txt | 51 ++
_sources/library-user-guide/upgrading.md.txt | 89 +++-
.../library-user-guide/using-the-sql-api.md.txt | 1 +
.../library-user-guide/working-with-exprs.md.txt | 99 ++++
_sources/user-guide/expressions.md.txt | 1 +
_sources/user-guide/sql/prepared_statements.md.txt | 16 +
11 files changed, 1170 insertions(+), 17 deletions(-)
diff --git a/_downloads/1e336e540124589d1e27210fa4f805f6/rust_lint.sh
b/_downloads/1e336e540124589d1e27210fa4f805f6/rust_lint.sh
index af0fce72cc..8fe7220085 100644
--- a/_downloads/1e336e540124589d1e27210fa4f805f6/rust_lint.sh
+++ b/_downloads/1e336e540124589d1e27210fa4f805f6/rust_lint.sh
@@ -20,13 +20,21 @@
# This script runs all the Rust lints locally the same way the
# DataFusion CI does
+# For `.toml` format checking
set -e
if ! command -v taplo &> /dev/null; then
echo "Installing taplo using cargo"
cargo install taplo-cli
fi
+# For Apache licence header checking
+if ! command -v hawkeye &> /dev/null; then
+ echo "Installing hawkeye using cargo"
+ cargo install hawkeye --locked
+fi
+
ci/scripts/rust_fmt.sh
ci/scripts/rust_clippy.sh
ci/scripts/rust_toml_fmt.sh
ci/scripts/rust_docs.sh
+ci/scripts/license_header.sh
\ No newline at end of file
diff --git a/_sources/library-user-guide/catalogs.md.txt
b/_sources/library-user-guide/catalogs.md.txt
index 75ac70b258..d4e6633d40 100644
--- a/_sources/library-user-guide/catalogs.md.txt
+++ b/_sources/library-user-guide/catalogs.md.txt
@@ -58,6 +58,14 @@ pub struct MemorySchemaProvider {
Then we implement the `SchemaProvider` trait for `MemorySchemaProvider`.
```rust
+# use std::sync::Arc;
+# use dashmap::DashMap;
+# use datafusion::catalog::TableProvider;
+#
+# #[derive(Debug)]
+# pub struct MemorySchemaProvider {
+# tables: DashMap<String, Arc<dyn TableProvider>>,
+# }
use std::any::Any;
use datafusion::catalog::SchemaProvider;
@@ -107,6 +115,58 @@ impl SchemaProvider for MemorySchemaProvider {
Without getting into a `CatalogProvider` implementation, we can create a
`MemorySchemaProvider` and register `TableProvider`s with it.
```rust
+# use std::sync::Arc;
+# use dashmap::DashMap;
+# use datafusion::catalog::TableProvider;
+#
+# #[derive(Debug)]
+# pub struct MemorySchemaProvider {
+# tables: DashMap<String, Arc<dyn TableProvider>>,
+# }
+#
+# use std::any::Any;
+# use datafusion::catalog::SchemaProvider;
+# use async_trait::async_trait;
+# use datafusion::common::{Result, exec_err};
+#
+# #[async_trait]
+# impl SchemaProvider for MemorySchemaProvider {
+# fn as_any(&self) -> &dyn Any {
+# self
+# }
+#
+# fn table_names(&self) -> Vec<String> {
+# self.tables
+# .iter()
+# .map(|table| table.key().clone())
+# .collect()
+# }
+#
+# async fn table(&self, name: &str) -> Result<Option<Arc<dyn
TableProvider>>> {
+# Ok(self.tables.get(name).map(|table| table.value().clone()))
+# }
+#
+# fn register_table(
+# &self,
+# name: String,
+# table: Arc<dyn TableProvider>,
+# ) -> Result<Option<Arc<dyn TableProvider>>> {
+# if self.table_exist(name.as_str()) {
+# return exec_err!(
+# "The table {name} already exists"
+# );
+# }
+# Ok(self.tables.insert(name, table))
+# }
+#
+# fn deregister_table(&self, name: &str) -> Result<Option<Arc<dyn
TableProvider>>> {
+# Ok(self.tables.remove(name).map(|(_, table)| table))
+# }
+#
+# fn table_exist(&self, name: &str) -> bool {
+# self.tables.contains_key(name)
+# }
+# }
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
@@ -143,12 +203,25 @@ It's often useful to fetch metadata about which tables
are in a schema, from a r
The trait is roughly the same except for the `table` method, and the addition
of the `#[async_trait]` attribute.
```rust
+# use async_trait::async_trait;
+# use std::sync::Arc;
+# use datafusion::catalog::{TableProvider, SchemaProvider};
+# use datafusion::common::Result;
+#
+# type OriginSchema = arrow::datatypes::Schema;
+#
+# #[derive(Debug)]
+# struct Schema(OriginSchema);
#[async_trait]
impl SchemaProvider for Schema {
async fn table(&self, name: &str) -> Result<Option<Arc<dyn
TableProvider>>> {
+# todo!();
}
+# fn as_any(&self) -> &(dyn std::any::Any + 'static) { todo!() }
+# fn table_names(&self) -> Vec<std::string::String> { todo!() }
+# fn table_exist(&self, _: &str) -> bool { todo!() }
}
```
@@ -211,6 +284,7 @@ Again, this is fairly straightforward, as there's an
underlying data structure t
## Implementing `MemoryCatalogProviderList`
```rust
+
use std::any::Any;
use std::sync::Arc;
use dashmap::DashMap;
diff --git a/_sources/library-user-guide/custom-table-providers.md.txt
b/_sources/library-user-guide/custom-table-providers.md.txt
index 2da4d781e2..695cb16ac8 100644
--- a/_sources/library-user-guide/custom-table-providers.md.txt
+++ b/_sources/library-user-guide/custom-table-providers.md.txt
@@ -172,6 +172,113 @@ The `scan` method of the `TableProvider` returns a
`Result<Arc<dyn ExecutionPlan
```rust
+# use std::any::Any;
+# use std::sync::{Arc, Mutex};
+# use std::collections::{BTreeMap, HashMap};
+# use datafusion::common::Result;
+# use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+# use datafusion::physical_plan::expressions::PhysicalSortExpr;
+# use datafusion::physical_plan::{
+# ExecutionPlan, SendableRecordBatchStream, DisplayAs, DisplayFormatType,
+# Statistics, PlanProperties
+# };
+# use datafusion::execution::context::TaskContext;
+# use datafusion::arrow::array::{UInt64Builder, UInt8Builder};
+# use datafusion::physical_plan::memory::MemoryStream;
+# use datafusion::arrow::record_batch::RecordBatch;
+#
+# /// A User, with an id and a bank account
+# #[derive(Clone, Debug)]
+# struct User {
+# id: u8,
+# bank_account: u64,
+# }
+#
+# /// A custom datasource, used to represent a datastore with a single index
+# #[derive(Clone, Debug)]
+# pub struct CustomDataSource {
+# inner: Arc<Mutex<CustomDataSourceInner>>,
+# }
+#
+# #[derive(Debug)]
+# struct CustomDataSourceInner {
+# data: HashMap<u8, User>,
+# bank_account_index: BTreeMap<u64, u8>,
+# }
+#
+# #[derive(Debug)]
+# struct CustomExec {
+# db: CustomDataSource,
+# projected_schema: SchemaRef,
+# }
+#
+# impl DisplayAs for CustomExec {
+# fn fmt_as(&self, _t: DisplayFormatType, f: &mut std::fmt::Formatter) ->
std::fmt::Result {
+# write!(f, "CustomExec")
+# }
+# }
+#
+# impl ExecutionPlan for CustomExec {
+# fn name(&self) -> &str {
+# "CustomExec"
+# }
+#
+# fn as_any(&self) -> &dyn Any {
+# self
+# }
+#
+# fn schema(&self) -> SchemaRef {
+# self.projected_schema.clone()
+# }
+#
+#
+# fn properties(&self) -> &PlanProperties {
+# unreachable!()
+# }
+#
+# fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+# Vec::new()
+# }
+#
+# fn with_new_children(
+# self: Arc<Self>,
+# _: Vec<Arc<dyn ExecutionPlan>>,
+# ) -> Result<Arc<dyn ExecutionPlan>> {
+# Ok(self)
+# }
+#
+# fn execute(
+# &self,
+# _partition: usize,
+# _context: Arc<TaskContext>,
+# ) -> Result<SendableRecordBatchStream> {
+# let users: Vec<User> = {
+# let db = self.db.inner.lock().unwrap();
+# db.data.values().cloned().collect()
+# };
+#
+# let mut id_array = UInt8Builder::with_capacity(users.len());
+# let mut account_array = UInt64Builder::with_capacity(users.len());
+#
+# for user in users {
+# id_array.append_value(user.id);
+# account_array.append_value(user.bank_account);
+# }
+#
+# Ok(Box::pin(MemoryStream::try_new(
+# vec![RecordBatch::try_new(
+# self.projected_schema.clone(),
+# vec![
+# Arc::new(id_array.finish()),
+# Arc::new(account_array.finish()),
+# ],
+# )?],
+# self.schema(),
+# None,
+# )?))
+# }
+# }
+
use async_trait::async_trait;
use datafusion::logical_expr::expr::Expr;
use datafusion::datasource::{TableProvider, TableType};
@@ -257,7 +364,171 @@ In order to use the custom table provider, we need to
register it with DataFusio
This will allow you to use the custom table provider in DataFusion. For
example, you could use it in a SQL query to get a `DataFrame`.
```rust
-
+# use std::any::Any;
+# use std::sync::{Arc, Mutex};
+# use std::collections::{BTreeMap, HashMap};
+# use datafusion::common::Result;
+# use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+# use datafusion::physical_plan::expressions::PhysicalSortExpr;
+# use datafusion::physical_plan::{
+# ExecutionPlan, SendableRecordBatchStream, DisplayAs, DisplayFormatType,
+# Statistics, PlanProperties
+# };
+# use datafusion::execution::context::TaskContext;
+# use datafusion::arrow::array::{UInt64Builder, UInt8Builder};
+# use datafusion::physical_plan::memory::MemoryStream;
+# use datafusion::arrow::record_batch::RecordBatch;
+#
+# /// A User, with an id and a bank account
+# #[derive(Clone, Debug)]
+# struct User {
+# id: u8,
+# bank_account: u64,
+# }
+#
+# /// A custom datasource, used to represent a datastore with a single index
+# #[derive(Clone, Debug)]
+# pub struct CustomDataSource {
+# inner: Arc<Mutex<CustomDataSourceInner>>,
+# }
+#
+# #[derive(Debug)]
+# struct CustomDataSourceInner {
+# data: HashMap<u8, User>,
+# bank_account_index: BTreeMap<u64, u8>,
+# }
+#
+# #[derive(Debug)]
+# struct CustomExec {
+# db: CustomDataSource,
+# projected_schema: SchemaRef,
+# }
+#
+# impl DisplayAs for CustomExec {
+# fn fmt_as(&self, _t: DisplayFormatType, f: &mut std::fmt::Formatter) ->
std::fmt::Result {
+# write!(f, "CustomExec")
+# }
+# }
+#
+# impl ExecutionPlan for CustomExec {
+# fn name(&self) -> &str {
+# "CustomExec"
+# }
+#
+# fn as_any(&self) -> &dyn Any {
+# self
+# }
+#
+# fn schema(&self) -> SchemaRef {
+# self.projected_schema.clone()
+# }
+#
+#
+# fn properties(&self) -> &PlanProperties {
+# unreachable!()
+# }
+#
+# fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+# Vec::new()
+# }
+#
+# fn with_new_children(
+# self: Arc<Self>,
+# _: Vec<Arc<dyn ExecutionPlan>>,
+# ) -> Result<Arc<dyn ExecutionPlan>> {
+# Ok(self)
+# }
+#
+# fn execute(
+# &self,
+# _partition: usize,
+# _context: Arc<TaskContext>,
+# ) -> Result<SendableRecordBatchStream> {
+# let users: Vec<User> = {
+# let db = self.db.inner.lock().unwrap();
+# db.data.values().cloned().collect()
+# };
+#
+# let mut id_array = UInt8Builder::with_capacity(users.len());
+# let mut account_array = UInt64Builder::with_capacity(users.len());
+#
+# for user in users {
+# id_array.append_value(user.id);
+# account_array.append_value(user.bank_account);
+# }
+#
+# Ok(Box::pin(MemoryStream::try_new(
+# vec![RecordBatch::try_new(
+# self.projected_schema.clone(),
+# vec![
+# Arc::new(id_array.finish()),
+# Arc::new(account_array.finish()),
+# ],
+# )?],
+# self.schema(),
+# None,
+# )?))
+# }
+# }
+
+# use async_trait::async_trait;
+# use datafusion::logical_expr::expr::Expr;
+# use datafusion::datasource::{TableProvider, TableType};
+# use datafusion::physical_plan::project_schema;
+# use datafusion::catalog::Session;
+#
+# impl CustomExec {
+# fn new(
+# projections: Option<&Vec<usize>>,
+# schema: SchemaRef,
+# db: CustomDataSource,
+# ) -> Self {
+# let projected_schema = project_schema(&schema, projections).unwrap();
+# Self {
+# db,
+# projected_schema,
+# }
+# }
+# }
+#
+# impl CustomDataSource {
+# pub(crate) async fn create_physical_plan(
+# &self,
+# projections: Option<&Vec<usize>>,
+# schema: SchemaRef,
+# ) -> Result<Arc<dyn ExecutionPlan>> {
+# Ok(Arc::new(CustomExec::new(projections, schema, self.clone())))
+# }
+# }
+#
+# #[async_trait]
+# impl TableProvider for CustomDataSource {
+# fn as_any(&self) -> &dyn Any {
+# self
+# }
+#
+# fn schema(&self) -> SchemaRef {
+# SchemaRef::new(Schema::new(vec![
+# Field::new("id", DataType::UInt8, false),
+# Field::new("bank_account", DataType::UInt64, true),
+# ]))
+# }
+#
+# fn table_type(&self) -> TableType {
+# TableType::Base
+# }
+#
+# async fn scan(
+# &self,
+# _state: &dyn Session,
+# projection: Option<&Vec<usize>>,
+# // filters and limit can be used here to inject some push-down
operations if needed
+# _filters: &[Expr],
+# _limit: Option<usize>,
+# ) -> Result<Arc<dyn ExecutionPlan>> {
+# return self.create_physical_plan(projection, self.schema()).await;
+# }
+# }
use datafusion::execution::context::SessionContext;
@@ -277,6 +548,7 @@ async fn main() -> Result<()> {
Ok(())
}
+
```
## Recap
diff --git a/_sources/library-user-guide/extending-operators.md.txt
b/_sources/library-user-guide/extending-operators.md.txt
index c7b9cb2b10..5c28d1e670 100644
--- a/_sources/library-user-guide/extending-operators.md.txt
+++ b/_sources/library-user-guide/extending-operators.md.txt
@@ -29,8 +29,7 @@ DataFusion supports extension of operators by transforming
logical plan and exec
The `rewrite` function transforms logical plans by identifying temporal
patterns and aggregation functions that match the stored wheel indices. When
match is found, it queries the corresponding index to retrieve pre-computed
aggregate values, stores these results in a
[MemTable](https://docs.rs/datafusion/latest/datafusion/datasource/memory/struct.MemTable.html),
and returns as a new `LogicalPlan::TableScan`. If no match is found, the
original plan proceeds unchanged through DataFusion's [...]
-```rust
-,ignore
+```rust,ignore
fn rewrite(
&self,
plan: LogicalPlan,
@@ -46,8 +45,7 @@ fn rewrite(
}
```
-```rust
-,ignore
+```rust,ignore
// Converts a uwheel aggregate result to a TableScan with a MemTable as source
fn agg_to_table_scan(result: f64, schema: SchemaRef) -> Result<LogicalPlan> {
let data = Float64Array::from(vec![result]);
diff --git a/_sources/library-user-guide/functions/adding-udfs.md.txt
b/_sources/library-user-guide/functions/adding-udfs.md.txt
index 78477aeb95..2335105882 100644
--- a/_sources/library-user-guide/functions/adding-udfs.md.txt
+++ b/_sources/library-user-guide/functions/adding-udfs.md.txt
@@ -128,6 +128,64 @@ impl ScalarUDFImpl for AddOne {
We now need to register the function with DataFusion so that it can be used in
the context of a query.
```rust
+# use std::sync::Arc;
+# use std::any::Any;
+# use std::sync::LazyLock;
+# use arrow::datatypes::DataType;
+# use datafusion_common::cast::as_int64_array;
+# use datafusion_common::{DataFusionError, plan_err, Result};
+# use datafusion_expr::{col, ColumnarValue, ScalarFunctionArgs, Signature,
Volatility};
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion_expr::{ScalarUDFImpl, ScalarUDF};
+# use datafusion_macros::user_doc;
+# use datafusion_doc::Documentation;
+#
+# /// This struct for a simple UDF that adds one to an int32
+# #[user_doc(
+# doc_section(label = "Math Functions"),
+# description = "Add one udf",
+# syntax_example = "add_one(1)"
+# )]
+# #[derive(Debug, PartialEq, Eq, Hash)]
+# struct AddOne {
+# signature: Signature,
+# }
+#
+# impl AddOne {
+# fn new() -> Self {
+# Self {
+# signature: Signature::uniform(1, vec![DataType::Int32],
Volatility::Immutable),
+# }
+# }
+# }
+#
+# /// Implement the ScalarUDFImpl trait for AddOne
+# impl ScalarUDFImpl for AddOne {
+# fn as_any(&self) -> &dyn Any { self }
+# fn name(&self) -> &str { "add_one" }
+# fn signature(&self) -> &Signature { &self.signature }
+# fn return_type(&self, args: &[DataType]) -> Result<DataType> {
+# if !matches!(args.get(0), Some(&DataType::Int32)) {
+# return plan_err!("add_one only accepts Int32 arguments");
+# }
+# Ok(DataType::Int32)
+# }
+# // The actual implementation would add one to the argument
+# fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
+# let args = ColumnarValue::values_to_arrays(&args.args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
+# fn documentation(&self) -> Option<&Documentation> {
+# self.doc()
+# }
+# }
use datafusion::execution::context::SessionContext;
// Create a new ScalarUDF from the implementation
@@ -172,6 +230,24 @@ This "works" in isolation, i.e. if you have a slice of
`ArrayRef`s, you can call
`ArrayRef` with 1 added to each value.
```rust
+# use std::sync::Arc;
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::common::cast::as_int64_array;
+# use datafusion::common::Result;
+# use datafusion::logical_expr::ColumnarValue;
+#
+# pub fn add_one(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+# // Error handling omitted for brevity
+# let args = ColumnarValue::values_to_arrays(args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
let input = vec![Some(1), None, Some(3)];
let input = ColumnarValue::from(Arc::new(Int64Array::from(input)) as ArrayRef);
@@ -192,6 +268,24 @@ with the `SessionContext`.
DataFusion provides the [`create_udf`] and helper functions to make this
easier.
```rust
+# use std::sync::Arc;
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::common::cast::as_int64_array;
+# use datafusion::common::Result;
+# use datafusion::logical_expr::ColumnarValue;
+#
+# pub fn add_one(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+# // Error handling omitted for brevity
+# let args = ColumnarValue::values_to_arrays(args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
use datafusion::logical_expr::{Volatility, create_udf};
use datafusion::arrow::datatypes::DataType;
@@ -219,6 +313,24 @@ A few things to note on `create_udf`:
That gives us a `ScalarUDF` that we can register with the `SessionContext`:
```rust
+# use std::sync::Arc;
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::common::cast::as_int64_array;
+# use datafusion::common::Result;
+# use datafusion::logical_expr::ColumnarValue;
+#
+# pub fn add_one(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+# // Error handling omitted for brevity
+# let args = ColumnarValue::values_to_arrays(args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
use datafusion::logical_expr::{Volatility, create_udf};
use datafusion::arrow::datatypes::DataType;
use datafusion::execution::context::SessionContext;
@@ -256,6 +368,21 @@ To add a Scalar Async UDF, you need to:
### Adding by `impl AsyncScalarUDFImpl`
```rust
+# use arrow::array::{ArrayIter, ArrayRef, AsArray, StringArray};
+# use arrow_schema::DataType;
+# use async_trait::async_trait;
+# use datafusion::common::error::Result;
+# use datafusion::common::{internal_err, not_impl_err};
+# use datafusion::common::types::logical_string;
+# use datafusion::config::ConfigOptions;
+# use datafusion_expr::ScalarUDFImpl;
+# use datafusion::logical_expr::async_udf::AsyncScalarUDFImpl;
+# use datafusion::logical_expr::{
+# ColumnarValue, Signature, TypeSignature, TypeSignatureClass, Volatility,
ScalarFunctionArgs
+# };
+# use datafusion::logical_expr_common::signature::Coercion;
+# use std::any::Any;
+# use std::sync::Arc;
#[derive(Debug, PartialEq, Eq, Hash)]
pub struct AsyncUpper {
@@ -344,6 +471,99 @@ impl AsyncScalarUDFImpl for AsyncUpper {
We can now transfer the async UDF into the normal scalar using
`into_scalar_udf` to register the function with DataFusion so that it can be
used in the context of a query.
```rust
+# use arrow::array::{ArrayIter, ArrayRef, AsArray, StringArray};
+# use arrow_schema::DataType;
+# use async_trait::async_trait;
+# use datafusion::common::error::Result;
+# use datafusion::common::{internal_err, not_impl_err};
+# use datafusion::common::types::logical_string;
+# use datafusion::config::ConfigOptions;
+# use datafusion_expr::ScalarUDFImpl;
+# use datafusion::logical_expr::async_udf::AsyncScalarUDFImpl;
+# use datafusion::logical_expr::{
+# ColumnarValue, Signature, TypeSignature, TypeSignatureClass, Volatility,
ScalarFunctionArgs
+# };
+# use datafusion::logical_expr_common::signature::Coercion;
+# use log::trace;
+# use std::any::Any;
+# use std::sync::Arc;
+#
+# #[derive(Debug, PartialEq, Eq, Hash)]
+# pub struct AsyncUpper {
+# signature: Signature,
+# }
+#
+# impl Default for AsyncUpper {
+# fn default() -> Self {
+# Self::new()
+# }
+# }
+#
+# impl AsyncUpper {
+# pub fn new() -> Self {
+# Self {
+# signature: Signature::new(
+# TypeSignature::Coercible(vec![Coercion::Exact {
+# desired_type:
TypeSignatureClass::Native(logical_string()),
+# }]),
+# Volatility::Volatile,
+# ),
+# }
+# }
+# }
+#
+# #[async_trait]
+# impl ScalarUDFImpl for AsyncUpper {
+# fn as_any(&self) -> &dyn Any {
+# self
+# }
+#
+# fn name(&self) -> &str {
+# "async_upper"
+# }
+#
+# fn signature(&self) -> &Signature {
+# &self.signature
+# }
+#
+# fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+# Ok(DataType::Utf8)
+# }
+#
+# fn invoke_with_args(
+# &self,
+# _args: ScalarFunctionArgs,
+# ) -> Result<ColumnarValue> {
+# not_impl_err!("AsyncUpper can only be called from async contexts")
+# }
+# }
+#
+# #[async_trait]
+# impl AsyncScalarUDFImpl for AsyncUpper {
+# fn ideal_batch_size(&self) -> Option<usize> {
+# Some(10)
+# }
+#
+# async fn invoke_async_with_args(
+# &self,
+# args: ScalarFunctionArgs,
+# ) -> Result<ColumnarValue> {
+# trace!("Invoking async_upper with args: {:?}", args);
+# let value = &args.args[0];
+# let result = match value {
+# ColumnarValue::Array(array) => {
+# let string_array = array.as_string::<i32>();
+# let iter = ArrayIter::new(string_array);
+# let result = iter
+# .map(|string| string.map(|s| s.to_uppercase()))
+# .collect::<StringArray>();
+# Arc::new(result) as ArrayRef
+# }
+# _ => return internal_err!("Expected a string argument, got
{:?}", value),
+# };
+# Ok(ColumnarValue::from(result))
+# }
+# }
use datafusion::execution::context::SessionContext;
use datafusion::logical_expr::async_udf::AsyncScalarUDF;
@@ -447,6 +667,50 @@ with the `SessionContext`. DataFusion provides the
[`create_udwf`] helper functi
There is a lower level API with more functionality but is more complex, that
is documented in [`advanced_udwf.rs`].
```rust
+# use datafusion::arrow::{array::{ArrayRef, Float64Array, AsArray},
datatypes::Float64Type};
+# use datafusion::logical_expr::{PartitionEvaluator};
+# use datafusion::common::ScalarValue;
+# use datafusion::error::Result;
+#
+# #[derive(Clone, Debug)]
+# struct MyPartitionEvaluator {}
+#
+# impl MyPartitionEvaluator {
+# fn new() -> Self {
+# Self {}
+# }
+# }
+#
+# impl PartitionEvaluator for MyPartitionEvaluator {
+# fn uses_window_frame(&self) -> bool {
+# true
+# }
+#
+# fn evaluate(
+# &mut self,
+# values: &[ArrayRef],
+# range: &std::ops::Range<usize>,
+# ) -> Result<ScalarValue> {
+# // Again, the input argument is an array of floating
+# // point numbers to calculate a moving average
+# let arr: &Float64Array =
values[0].as_ref().as_primitive::<Float64Type>();
+#
+# let range_len = range.end - range.start;
+#
+# // our smoothing function will average all the values in the
+# let output = if range_len > 0 {
+# let sum: f64 =
arr.values().iter().skip(range.start).take(range_len).sum();
+# Some(sum / range_len as f64)
+# } else {
+# None
+# };
+#
+# Ok(ScalarValue::Float64(output))
+# }
+# }
+# fn make_partition_evaluator() -> Result<Box<dyn PartitionEvaluator>> {
+# Ok(Box::new(MyPartitionEvaluator::new()))
+# }
use datafusion::logical_expr::{Volatility, create_udwf};
use datafusion::arrow::datatypes::DataType;
use std::sync::Arc;
@@ -480,6 +744,62 @@ The `create_udwf` has five arguments to check:
That gives us a `WindowUDF` that we can register with the `SessionContext`:
```rust
+# use datafusion::arrow::{array::{ArrayRef, Float64Array, AsArray},
datatypes::Float64Type};
+# use datafusion::logical_expr::{PartitionEvaluator};
+# use datafusion::common::ScalarValue;
+# use datafusion::error::Result;
+#
+# #[derive(Clone, Debug)]
+# struct MyPartitionEvaluator {}
+#
+# impl MyPartitionEvaluator {
+# fn new() -> Self {
+# Self {}
+# }
+# }
+#
+# impl PartitionEvaluator for MyPartitionEvaluator {
+# fn uses_window_frame(&self) -> bool {
+# true
+# }
+#
+# fn evaluate(
+# &mut self,
+# values: &[ArrayRef],
+# range: &std::ops::Range<usize>,
+# ) -> Result<ScalarValue> {
+# // Again, the input argument is an array of floating
+# // point numbers to calculate a moving average
+# let arr: &Float64Array =
values[0].as_ref().as_primitive::<Float64Type>();
+#
+# let range_len = range.end - range.start;
+#
+# // our smoothing function will average all the values in the
+# let output = if range_len > 0 {
+# let sum: f64 =
arr.values().iter().skip(range.start).take(range_len).sum();
+# Some(sum / range_len as f64)
+# } else {
+# None
+# };
+#
+# Ok(ScalarValue::Float64(output))
+# }
+# }
+# fn make_partition_evaluator() -> Result<Box<dyn PartitionEvaluator>> {
+# Ok(Box::new(MyPartitionEvaluator::new()))
+# }
+# use datafusion::logical_expr::{Volatility, create_udwf};
+# use datafusion::arrow::datatypes::DataType;
+# use std::sync::Arc;
+#
+# // here is where we define the UDWF. We also declare its signature:
+# let smooth_it = create_udwf(
+# "smooth_it",
+# DataType::Float64,
+# Arc::new(DataType::Float64),
+# Volatility::Immutable,
+# Arc::new(make_partition_evaluator),
+# );
use datafusion::execution::context::SessionContext;
let ctx = SessionContext::new();
@@ -503,6 +823,54 @@ green,10.3,1996-04-12T12:05:04.000000000
Then, we can query like below:
```rust
+# use datafusion::arrow::{array::{ArrayRef, Float64Array, AsArray},
datatypes::Float64Type};
+# use datafusion::logical_expr::{PartitionEvaluator};
+# use datafusion::common::ScalarValue;
+# use datafusion::error::Result;
+#
+# #[derive(Clone, Debug)]
+# struct MyPartitionEvaluator {}
+#
+# impl MyPartitionEvaluator {
+# fn new() -> Self {
+# Self {}
+# }
+# }
+#
+# impl PartitionEvaluator for MyPartitionEvaluator {
+# fn uses_window_frame(&self) -> bool {
+# true
+# }
+#
+# fn evaluate(
+# &mut self,
+# values: &[ArrayRef],
+# range: &std::ops::Range<usize>,
+# ) -> Result<ScalarValue> {
+# // Again, the input argument is an array of floating
+# // point numbers to calculate a moving average
+# let arr: &Float64Array =
values[0].as_ref().as_primitive::<Float64Type>();
+#
+# let range_len = range.end - range.start;
+#
+# // our smoothing function will average all the values in the
+# let output = if range_len > 0 {
+# let sum: f64 =
arr.values().iter().skip(range.start).take(range_len).sum();
+# Some(sum / range_len as f64)
+# } else {
+# None
+# };
+#
+# Ok(ScalarValue::Float64(output))
+# }
+# }
+# fn make_partition_evaluator() -> Result<Box<dyn PartitionEvaluator>> {
+# Ok(Box::new(MyPartitionEvaluator::new()))
+# }
+# use datafusion::logical_expr::{Volatility, create_udwf};
+# use datafusion::arrow::datatypes::DataType;
+# use std::sync::Arc;
+# use datafusion::execution::context::SessionContext;
use datafusion::datasource::file_format::options::CsvReadOptions;
@@ -574,6 +942,7 @@ Aggregate UDFs are functions that take a group of rows and
return a single value
For example, we will declare a single-type, single return type UDAF that
computes the geometric mean.
```rust
+
use datafusion::arrow::array::ArrayRef;
use datafusion::scalar::ScalarValue;
use datafusion::{error::Result, physical_plan::Accumulator};
@@ -669,6 +1038,78 @@ it with the `SessionContext`. DataFusion provides the
[`create_udaf`] helper fun
There is a lower level API with more functionality but is more complex, that
is documented in [`advanced_udaf.rs`].
```rust
+# use datafusion::arrow::array::ArrayRef;
+# use datafusion::scalar::ScalarValue;
+# use datafusion::{error::Result, physical_plan::Accumulator};
+#
+# #[derive(Debug)]
+# struct GeometricMean {
+# n: u32,
+# prod: f64,
+# }
+#
+# impl GeometricMean {
+# pub fn new() -> Self {
+# GeometricMean { n: 0, prod: 1.0 }
+# }
+# }
+#
+# impl Accumulator for GeometricMean {
+# fn state(&mut self) -> Result<Vec<ScalarValue>> {
+# Ok(vec![
+# ScalarValue::from(self.prod),
+# ScalarValue::from(self.n),
+# ])
+# }
+#
+# fn evaluate(&mut self) -> Result<ScalarValue> {
+# let value = self.prod.powf(1.0 / self.n as f64);
+# Ok(ScalarValue::from(value))
+# }
+#
+# fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+# if values.is_empty() {
+# return Ok(());
+# }
+# let arr = &values[0];
+# (0..arr.len()).try_for_each(|index| {
+# let v = ScalarValue::try_from_array(arr, index)?;
+#
+# if let ScalarValue::Float64(Some(value)) = v {
+# self.prod *= value;
+# self.n += 1;
+# } else {
+# unreachable!("")
+# }
+# Ok(())
+# })
+# }
+#
+# fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+# if states.is_empty() {
+# return Ok(());
+# }
+# let arr = &states[0];
+# (0..arr.len()).try_for_each(|index| {
+# let v = states
+# .iter()
+# .map(|array| ScalarValue::try_from_array(array, index))
+# .collect::<Result<Vec<_>>>()?;
+# if let (ScalarValue::Float64(Some(prod)),
ScalarValue::UInt32(Some(n))) = (&v[0], &v[1])
+# {
+# self.prod *= prod;
+# self.n += n;
+# } else {
+# unreachable!("")
+# }
+# Ok(())
+# })
+# }
+#
+# fn size(&self) -> usize {
+# std::mem::size_of_val(self)
+# }
+# }
use datafusion::logical_expr::{Volatility, create_udaf};
use datafusion::arrow::datatypes::DataType;
@@ -705,6 +1146,79 @@ The `create_udaf` has six arguments to check:
```rust
+# use datafusion::arrow::array::ArrayRef;
+# use datafusion::scalar::ScalarValue;
+# use datafusion::{error::Result, physical_plan::Accumulator};
+#
+# #[derive(Debug)]
+# struct GeometricMean {
+# n: u32,
+# prod: f64,
+# }
+#
+# impl GeometricMean {
+# pub fn new() -> Self {
+# GeometricMean { n: 0, prod: 1.0 }
+# }
+# }
+#
+# impl Accumulator for GeometricMean {
+# fn state(&mut self) -> Result<Vec<ScalarValue>> {
+# Ok(vec![
+# ScalarValue::from(self.prod),
+# ScalarValue::from(self.n),
+# ])
+# }
+#
+# fn evaluate(&mut self) -> Result<ScalarValue> {
+# let value = self.prod.powf(1.0 / self.n as f64);
+# Ok(ScalarValue::from(value))
+# }
+#
+# fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+# if values.is_empty() {
+# return Ok(());
+# }
+# let arr = &values[0];
+# (0..arr.len()).try_for_each(|index| {
+# let v = ScalarValue::try_from_array(arr, index)?;
+#
+# if let ScalarValue::Float64(Some(value)) = v {
+# self.prod *= value;
+# self.n += 1;
+# } else {
+# unreachable!("")
+# }
+# Ok(())
+# })
+# }
+#
+# fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+# if states.is_empty() {
+# return Ok(());
+# }
+# let arr = &states[0];
+# (0..arr.len()).try_for_each(|index| {
+# let v = states
+# .iter()
+# .map(|array| ScalarValue::try_from_array(array, index))
+# .collect::<Result<Vec<_>>>()?;
+# if let (ScalarValue::Float64(Some(prod)),
ScalarValue::UInt32(Some(n))) = (&v[0], &v[1])
+# {
+# self.prod *= prod;
+# self.n += n;
+# } else {
+# unreachable!("")
+# }
+# Ok(())
+# })
+# }
+#
+# fn size(&self) -> usize {
+# std::mem::size_of_val(self)
+# }
+# }
+
use datafusion::logical_expr::{Volatility, create_udaf};
use datafusion::arrow::datatypes::DataType;
use std::sync::Arc;
@@ -736,6 +1250,7 @@ async fn main() -> Result<()> {
let df = ctx.sql("SELECT geo_mean(speed) FROM cars").await?;
Ok(())
}
+
```
[`aggregateudf`]:
https://docs.rs/datafusion/latest/datafusion/logical_expr/struct.AggregateUDF.html
@@ -824,6 +1339,40 @@ impl TableFunctionImpl for EchoFunction {
With the UDTF implemented, you can register it with the `SessionContext`:
```rust
+# use std::sync::Arc;
+# use datafusion::common::{plan_err, ScalarValue, Result};
+# use datafusion::catalog::{TableFunctionImpl, TableProvider};
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::datasource::memory::MemTable;
+# use arrow::record_batch::RecordBatch;
+# use arrow::datatypes::{DataType, Field, Schema};
+# use datafusion_expr::Expr;
+#
+# /// A table function that returns a table provider with the value as a
single column
+# #[derive(Debug, Default)]
+# pub struct EchoFunction {}
+#
+# impl TableFunctionImpl for EchoFunction {
+# fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
+# let Some(Expr::Literal(ScalarValue::Int64(Some(value)), _)) =
exprs.get(0) else {
+# return plan_err!("First argument must be an integer");
+# };
+#
+# // Create the schema for the table
+# let schema = Arc::new(Schema::new(vec![Field::new("a",
DataType::Int64, false)]));
+#
+# // Create a single RecordBatch with the value as a single column
+# let batch = RecordBatch::try_new(
+# schema.clone(),
+# vec![Arc::new(Int64Array::from(vec![*value]))],
+# )?;
+#
+# // Create a MemTable plan that returns the RecordBatch
+# let provider = MemTable::try_new(schema, vec![vec![batch]])?;
+#
+# Ok(Arc::new(provider))
+# }
+# }
use datafusion::execution::context::SessionContext;
use datafusion::arrow::util::pretty;
@@ -867,8 +1416,23 @@ To extend DataFusion with support for custom operators
not natively available, y
See example below:
```rust
-
-
+# use arrow::array::RecordBatch;
+# use std::sync::Arc;
+
+# use datafusion::common::{assert_batches_eq, DFSchema};
+# use datafusion::error::Result;
+# use datafusion::execution::FunctionRegistry;
+# use datafusion::logical_expr::Operator;
+# use datafusion::prelude::*;
+# use datafusion::sql::sqlparser::ast::BinaryOperator;
+# use datafusion_common::ScalarValue;
+# use datafusion_expr::expr::Alias;
+# use datafusion_expr::planner::{ExprPlanner, PlannerResult, RawBinaryExpr};
+# use datafusion_expr::BinaryExpr;
+
+# #[derive(Debug)]
+# // Define the custom planner
+# struct MyCustomPlanner;
// Implement ExprPlanner to add support for the `->` custom operator
impl ExprPlanner for MyCustomPlanner {
diff --git a/_sources/library-user-guide/query-optimizer.md.txt
b/_sources/library-user-guide/query-optimizer.md.txt
index fac171ff62..877ff8c754 100644
--- a/_sources/library-user-guide/query-optimizer.md.txt
+++ b/_sources/library-user-guide/query-optimizer.md.txt
@@ -34,6 +34,7 @@ The following code demonstrates the basic flow of creating
the optimizer with a
and applying it to a logical plan to produce an optimized logical plan.
```rust
+
use std::sync::Arc;
use datafusion::logical_expr::{col, lit, LogicalPlan, LogicalPlanBuilder};
use datafusion::optimizer::{OptimizerRule, OptimizerContext, Optimizer};
@@ -79,6 +80,11 @@ the optimizer can simply return it as is.
All rules must implement the `OptimizerRule` trait.
```rust
+# use datafusion::common::tree_node::Transformed;
+# use datafusion::common::Result;
+# use datafusion::logical_expr::LogicalPlan;
+# use datafusion::optimizer::{OptimizerConfig, OptimizerRule};
+#
#[derive(Default, Debug)]
struct MyOptimizerRule {}
@@ -103,6 +109,28 @@ impl OptimizerRule for MyOptimizerRule {
The optimizer can be created with a custom set of rules.
```rust
+# use std::sync::Arc;
+# use datafusion::logical_expr::{col, lit, LogicalPlan, LogicalPlanBuilder};
+# use datafusion::optimizer::{OptimizerRule, OptimizerConfig,
OptimizerContext, Optimizer};
+# use datafusion::common::tree_node::Transformed;
+# use datafusion::common::Result;
+#
+# #[derive(Default, Debug)]
+# struct MyOptimizerRule {}
+#
+# impl OptimizerRule for MyOptimizerRule {
+# fn name(&self) -> &str {
+# "my_optimizer_rule"
+# }
+#
+# fn rewrite(
+# &self,
+# plan: LogicalPlan,
+# _config: &dyn OptimizerConfig,
+# ) -> Result<Transformed<LogicalPlan>> {
+# unimplemented!()
+# }
+# }
let optimizer = Optimizer::with_rules(vec occur where the ex
There are currently two ways to create a name for an expression in the logical
plan.
```rust
+# use datafusion::common::Result;
+# struct Expr;
impl Expr {
/// Returns the name of this expression as it should appear in a schema.
This name
@@ -216,6 +246,9 @@ The [TreeNode API] provides a convenient way to recursively
walk an expression o
For example, to find all subquery references in a logical plan, the following
code can be used:
```rust
+# use datafusion::prelude::*;
+# use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
+# use datafusion::common::Result;
// Return all subquery references in an expression
fn extract_subquery_filters(expression: &Expr) -> Result<Vec<&Expr>> {
let mut extracted = vec![];
@@ -232,6 +265,10 @@ fn extract_subquery_filters(expression: &Expr) ->
Result<Vec<&Expr>> {
Likewise you can use the [TreeNode API] to rewrite a `LogicalPlan` or
`ExecutionPlan`
```rust
+# use datafusion::prelude::*;
+# use datafusion::logical_expr::{LogicalPlan, Join};
+# use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
+# use datafusion::common::Result;
// Return all joins in a logical plan
fn find_joins(overall_plan: &LogicalPlan) -> Result<Vec<&Join>> {
let mut extracted = vec![];
@@ -263,6 +300,11 @@ col >= x AND col <= y
you can use the following code:
```rust
+# use datafusion::prelude::*;
+# use datafusion::logical_expr::{Between};
+# use datafusion::logical_expr::expr_fn::*;
+# use datafusion::common::tree_node::{Transformed, TreeNode,
TreeNodeRecursion};
+# use datafusion::common::Result;
// Recursively rewrite all BETWEEN expressions
// returns Transformed::yes if any changes were made
fn rewrite_between(expr: Expr) -> Result<Transformed<Expr>> {
@@ -421,6 +463,15 @@ to infer the selectivity of the expression and the space
of possible values it c
take.
```rust
+# use std::sync::Arc;
+# use datafusion::prelude::*;
+# use datafusion::physical_expr::{analyze, AnalysisContext, ExprBoundaries};
+# use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
+# use datafusion::common::stats::Precision;
+#
+# use datafusion::common::{ColumnStatistics, DFSchema};
+# use datafusion::common::{ScalarValue, ToDFSchema};
+# use datafusion::error::Result;
fn analyze_filter_example() -> Result<()> {
// Create a schema with an 'age' column
let age = Field::new("age", DataType::Int64, false);
diff --git a/_sources/library-user-guide/upgrading.md.txt
b/_sources/library-user-guide/upgrading.md.txt
index 8530dd372f..d70413467a 100644
--- a/_sources/library-user-guide/upgrading.md.txt
+++ b/_sources/library-user-guide/upgrading.md.txt
@@ -115,6 +115,7 @@ returns a `ColumnarValue` instead of a `ArrayRef`.
To upgrade, change the return type of your implementation
```rust
+# /* comment to avoid running
impl AsyncScalarUDFImpl for AskLLM {
async fn invoke_async_with_args(
&self,
@@ -125,11 +126,13 @@ impl AsyncScalarUDFImpl for AskLLM {
return array_ref; // old code
}
}
+# */
```
To return a `ColumnarValue`
```rust
+# /* comment to avoid running
impl AsyncScalarUDFImpl for AskLLM {
async fn invoke_async_with_args(
&self,
@@ -140,6 +143,7 @@ impl AsyncScalarUDFImpl for AskLLM {
return ColumnarValue::from(array_ref); // new code
}
}
+# */
```
See [#16896](https://github.com/apache/datafusion/issues/16896) for more
details.
@@ -150,15 +154,13 @@ See
[#16896](https://github.com/apache/datafusion/issues/16896) for more details
**Before:**
-```rust
-,ignore
+```rust,ignore
pub type ProjectionExpr = (Arc<dyn PhysicalExpr>, String);
```
**After:**
-```rust
-,ignore
+```rust,ignore
#[derive(Debug, Clone)]
pub struct ProjectionExpr {
pub expr: Arc<dyn PhysicalExpr>,
@@ -191,13 +193,17 @@ automatically dereference the `Arc` when needed. However,
in some cases you may
have to change your code to explicitly call `as_ref()` for example, from
```rust
+# /* comment to avoid running
let optimizer_config: &ConfigOptions = state.options();
+# */
```
To
```rust
+# /* comment to avoid running
let optimizer_config: &ConfigOptions = state.options().as_ref();
+# */
```
See PR [#16970](https://github.com/apache/datafusion/pull/16970)
@@ -212,6 +218,7 @@ parameter.
You can change your code like this
```rust
+# /* comment to avoid running
impl AsyncScalarUDFImpl for AskLLM {
async fn invoke_async_with_args(
&self,
@@ -222,11 +229,13 @@ impl AsyncScalarUDFImpl for AskLLM {
}
...
}
+# */
```
To this:
```rust
+# /* comment to avoid running
impl AsyncScalarUDFImpl for AskLLM {
async fn invoke_async_with_args(
@@ -238,6 +247,7 @@ impl AsyncScalarUDFImpl for AskLLM {
}
...
}
+# */
```
### Schema Rewriter Module Moved to New Crate
@@ -289,6 +299,7 @@ See [#16996] for details.
To help with protobuf serialization, the `as_any()` method has been added to
the `LazyBatchGenerator` trait. This means you will need to add `as_any()` to
your implementation of `LazyBatchGenerator`:
```rust
+# /* comment to avoid running
impl LazyBatchGenerator for MyBatchGenerator {
fn as_any(&self) -> &dyn Any {
@@ -298,6 +309,7 @@ impl LazyBatchGenerator for MyBatchGenerator {
...
}
+# */
```
See [#17200](https://github.com/apache/datafusion/pull/17200) for details.
@@ -315,15 +327,13 @@ The `FileOpenFuture` type alias has been updated to use
`DataFusionError` instea
**Before:**
-```rust
-,ignore
+```rust,ignore
pub type FileOpenFuture = BoxFuture<'static, Result<BoxStream<'static,
Result<RecordBatch, ArrowError>>>>;
```
**After:**
-```rust
-,ignore
+```rust,ignore
pub type FileOpenFuture = BoxFuture<'static, Result<BoxStream<'static,
Result<RecordBatch>>>>;
```
@@ -353,8 +363,7 @@ See [#17407] for details.
We added a method to `PhysicalExpr` to mark a `PhysicalExpr` as volatile:
-```rust
-,ignore
+```rust,ignore
impl PhysicalExpr for MyRandomExpr {
fn is_volatile_node(&self) -> bool {
true
@@ -393,21 +402,25 @@ This is a breaking change. Code that constructs or
matches on these variants wil
For example, to create a `SchemaError`, instead of:
```rust
+# /* comment to avoid running
use datafusion_common::{DataFusionError, SchemaError};
DataFusionError::SchemaError(
SchemaError::DuplicateUnqualifiedField { name: "foo".to_string() },
Box::new(None)
)
+# */
```
You now need to `Box` the inner error:
```rust
+# /* comment to avoid running
use datafusion_common::{DataFusionError, SchemaError};
DataFusionError::SchemaError(
Box::new(SchemaError::DuplicateUnqualifiedField { name: "foo".to_string() }),
Box::new(None)
)
+# */
```
[#16652]: https://github.com/apache/datafusion/issues/16652
@@ -422,13 +435,17 @@ is more efficient.
To create `FieldMetadata` from a `Field`:
```rust
+# /* comment to avoid running
let metadata = FieldMetadata::from(&field);
+# */
```
To add metadata to a `Field`, use the `add_to_field` method:
```rust
+# /* comment to avoid running
let updated_field = metadata.add_to_field(field);
+# */
```
See [#16317] for details.
@@ -448,12 +465,14 @@ DataFusion 49.0.0 adds support for compressing spill
files when data is written
**Usage:**
```rust
+# /* comment to avoid running
use datafusion::prelude::*;
use datafusion_common::config::SpillCompression;
let config = SessionConfig::default()
.with_spill_compression(SpillCompression::Zstd);
let ctx = SessionContext::new_with_config(config);
+# */
```
Or via SQL:
@@ -493,12 +512,14 @@ To unify **all** SQL string types (`CHAR`, `VARCHAR`,
`TEXT`, `STRING`) to Arrow
#### Examples
```rust
+# /* comment to avoid running
// Disable Utf8View mapping for all SQL string types
let opts = datafusion::sql::planner::ParserOptions::new()
.with_map_string_types_to_utf8view(false);
// Verify the setting is applied
assert!(!opts.map_string_types_to_utf8view);
+# */
```
---
@@ -536,22 +557,26 @@ options for Parquet files. The `ParquetEncryptionOptions`
implements `Default`
so you can upgrade your existing code like this:
```rust
+# /* comment to avoid running
TableParquetOptions {
global,
column_specific_options,
key_value_metadata,
}
+# */
```
To this:
```rust
+# /* comment to avoid running
TableParquetOptions {
global,
column_specific_options,
key_value_metadata,
crypto: Default::default(), // New crypto field
}
+# */
```
## DataFusion `48.0.1`
@@ -566,9 +591,11 @@ This change also restores the default behavior of
`ListingTable` to its previous
you can maintain the current behavior by overriding the default value in your
code.
```rust
+# /* comment to avoid running
ListingOptions::new(Arc::new(ParquetFormat::default()))
.with_collect_stat(false)
// other options
+# */
```
## DataFusion `48.0.0`
@@ -581,21 +608,25 @@ carrying through Arrow field metadata to support
extension types and other uses.
This means code such as
```rust
+# /* comment to avoid running
match expr {
...
Expr::Literal(scalar) => ...
...
}
+# */
```
Should be updated to:
```rust
+# /* comment to avoid running
match expr {
...
Expr::Literal(scalar, _metadata) => ...
...
}
+# */
```
Likewise constructing `Expr::Literal` requires metadata as well. The [`lit`]
function
@@ -614,6 +645,7 @@ This is a breaking change, so you will need to update your
code if you match
on `Expr::WindowFunction` directly. For example, if you have code like this:
```rust
+# /* comment to avoid running
match expr {
Expr::WindowFunction(WindowFunction {
params:
@@ -629,11 +661,13 @@ match expr {
// other expr
}
}
+# */
```
You will need to change it to:
```rust
+# /* comment to avoid running
match expr {
Expr::WindowFunction(window_fun) => {
let WindowFunction {
@@ -650,6 +684,7 @@ match expr {
// other expr
}
}
+# */
```
[details on #16207]:
https://github.com/apache/datafusion/pull/16207#issuecomment-2922659103
@@ -711,9 +746,11 @@ and relied on the default value of `collect_stat` being
`true`, you will need to
explicitly set it to `true` in your code.
```rust
+# /* comment to avoid running
ListingOptions::new(Arc::new(ParquetFormat::default()))
.with_collect_stat(true)
// other options
+# */
```
### Processing `FieldRef` instead of `DataType` for user defined functions
@@ -808,6 +845,7 @@ Additionally `ObjectStore::list` and
`ObjectStore::list_with_offset` have been c
This requires converting from `usize` to `u64` occasionally as well as changes
to `ObjectStore` implementations such as
```rust
+# /* comment to avoid running
impl Objectstore {
...
// The range is now a u64 instead of usize
@@ -821,6 +859,7 @@ impl Objectstore {
self.inner.list(prefix)
}
}
+# */
```
The `ParquetObjectReader` has been updated to no longer require the object size
@@ -831,16 +870,20 @@ The `ParquetObjectReader` has been updated to no longer
require the object size
Pattern in DataFusion `46.0.0`:
```rust
+# /* comment to avoid running
let meta: ObjectMeta = ...;
let reader = ParquetObjectReader::new(store, meta);
+# */
```
Pattern in DataFusion `47.0.0`:
```rust
+# /* comment to avoid running
let meta: ObjectMeta = ...;
let reader = ParquetObjectReader::new(store, location)
.with_file_size(meta.size);
+# */
```
### `DisplayFormatType::TreeRender`
@@ -874,20 +917,24 @@ DataFusion 47.0.0 this has been changed to use
`FileScanConfigBuilder`. See
Pattern in DataFusion `46.0.0`:
```rust
+# /* comment to avoid running
let plan = FileScanConfig::new(url, schema, Arc::new(file_source))
.with_statistics(stats)
...
.build()
+# */
```
Pattern in DataFusion `47.0.0`:
```rust
+# /* comment to avoid running
let config = FileScanConfigBuilder::new(url, schema, Arc::new(file_source))
.with_statistics(stats)
...
.build();
let scan = DataSourceExec::from_data_source(config);
+# */
```
## DataFusion `46.0.0`
@@ -910,6 +957,7 @@ below. See [PR 14876] for an example.
Given existing code like this:
```rust
+# /* comment to avoid running
impl ScalarUDFImpl for SparkConcat {
...
fn invoke_batch(&self, args: &[ColumnarValue], number_rows: usize) ->
Result<ColumnarValue> {
@@ -923,11 +971,13 @@ impl ScalarUDFImpl for SparkConcat {
}
}
}
+# */
```
To
```rust
+# /* comment to avoid running
impl ScalarUDFImpl for SparkConcat {
...
fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
Result<ColumnarValue> {
@@ -942,6 +992,7 @@ impl ScalarUDFImpl for SparkConcat {
}
}
}
+ # */
```
[`scalarudfimpl::invoke()`]:
https://docs.rs/datafusion/latest/datafusion/logical_expr/trait.ScalarUDFImpl.html#method.invoke
@@ -972,15 +1023,18 @@ Here is more information about
Code that looks for `ParquetExec` like this will no longer work:
```rust
+# /* comment to avoid running
if let Some(parquet_exec) = plan.as_any().downcast_ref::<ParquetExec>() {
// Do something with ParquetExec here
}
+# */
```
Instead, with `DataSourceExec`, the same information is now on
`FileScanConfig` and
`ParquetSource`. The equivalent code is
```rust
+# /* comment to avoid running
if let Some(datasource_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
if let Some(scan_config) =
datasource_exec.data_source().as_any().downcast_ref::<FileScanConfig>() {
// FileGroups, and other information is on the FileScanConfig
@@ -990,6 +1044,7 @@ if let Some(datasource_exec) =
plan.as_any().downcast_ref::<DataSourceExec>() {
// Information on PruningPredicates and parquet options are here
}
}
+# */
```
### Cookbook: Changes to `ParquetExecBuilder`
@@ -998,6 +1053,7 @@ Likewise code that builds `ParquetExec` using the
`ParquetExecBuilder` such as
the following must be changed:
```rust
+# /* comment to avoid running
let mut exec_plan_builder = ParquetExecBuilder::new(
FileScanConfig::new(self.log_store.object_store_url(), file_schema)
.with_projection(self.projection.cloned())
@@ -1013,11 +1069,13 @@ if let Some(predicate) = logical_filter {
exec_plan_builder = exec_plan_builder.with_predicate(predicate);
}
};
+# */
```
New code should use `FileScanConfig` to build the appropriate `DataSourceExec`:
```rust
+# /* comment to avoid running
let mut file_source = ParquetSource::new(parquet_options)
.with_schema_adapter_factory(Arc::new(DeltaSchemaAdapterFactory {}));
@@ -1040,6 +1098,7 @@ let file_scan_config = FileScanConfig::new(
// Build the actual scan like this
parquet_scan: file_scan_config.build(),
+# */
```
### `datafusion-cli` no longer automatically unescapes strings
@@ -1091,6 +1150,8 @@ Each of the old variants can be converted to the new
format as follows:
`TypeSignature::ArraySignature(ArrayFunctionSignature::ArrayAndElement)`:
```rust
+# use datafusion::common::utils::ListCoercion;
+# use datafusion_expr_common::signature::{ArrayFunctionArgument,
ArrayFunctionSignature, TypeSignature};
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
arguments: vec![ArrayFunctionArgument::Array,
ArrayFunctionArgument::Element],
@@ -1101,6 +1162,8 @@
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
`TypeSignature::ArraySignature(ArrayFunctionSignature::ElementAndArray)`:
```rust
+# use datafusion::common::utils::ListCoercion;
+# use datafusion_expr_common::signature::{ArrayFunctionArgument,
ArrayFunctionSignature, TypeSignature};
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
arguments: vec![ArrayFunctionArgument::Element,
ArrayFunctionArgument::Array],
@@ -1111,6 +1174,8 @@
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
`TypeSignature::ArraySignature(ArrayFunctionSignature::ArrayAndIndex)`:
```rust
+# use datafusion::common::utils::ListCoercion;
+# use datafusion_expr_common::signature::{ArrayFunctionArgument,
ArrayFunctionSignature, TypeSignature};
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
arguments: vec![ArrayFunctionArgument::Array,
ArrayFunctionArgument::Index],
@@ -1121,6 +1186,8 @@
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
`TypeSignature::ArraySignature(ArrayFunctionSignature::ArrayAndElementAndOptionalIndex)`:
```rust
+# use datafusion::common::utils::ListCoercion;
+# use datafusion_expr_common::signature::{ArrayFunctionArgument,
ArrayFunctionSignature, TypeSignature};
TypeSignature::OneOf(vec![
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
@@ -1141,6 +1208,8 @@ TypeSignature::OneOf(vec![
`TypeSignature::ArraySignature(ArrayFunctionSignature::Array)`:
```rust
+# use datafusion::common::utils::ListCoercion;
+# use datafusion_expr_common::signature::{ArrayFunctionArgument,
ArrayFunctionSignature, TypeSignature};
TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
arguments: vec![ArrayFunctionArgument::Array],
diff --git a/_sources/library-user-guide/using-the-sql-api.md.txt
b/_sources/library-user-guide/using-the-sql-api.md.txt
index fad5e3e1d6..8b8ba2a371 100644
--- a/_sources/library-user-guide/using-the-sql-api.md.txt
+++ b/_sources/library-user-guide/using-the-sql-api.md.txt
@@ -119,6 +119,7 @@ async fn main() -> Result<()> {
DataFusion can also read Avro files using the `register_avro` method.
```rust
+# #[cfg(feature = "avro")]
{
use datafusion::arrow::util::pretty;
use datafusion::error::Result;
diff --git a/_sources/library-user-guide/working-with-exprs.md.txt
b/_sources/library-user-guide/working-with-exprs.md.txt
index 90019fd58b..634e3fea55 100644
--- a/_sources/library-user-guide/working-with-exprs.md.txt
+++ b/_sources/library-user-guide/working-with-exprs.md.txt
@@ -80,6 +80,24 @@ We'll use a `ScalarUDF` expression as our example. This
necessitates implementin
So assuming you've written that function, you can use it to create an `Expr`:
```rust
+# use std::sync::Arc;
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::common::cast::as_int64_array;
+# use datafusion::common::Result;
+# use datafusion::logical_expr::ColumnarValue;
+#
+# pub fn add_one(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+# // Error handling omitted for brevity
+# let args = ColumnarValue::values_to_arrays(args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
use datafusion::logical_expr::{Volatility, create_udf};
use datafusion::arrow::datatypes::DataType;
use datafusion::logical_expr::{col, lit};
@@ -158,6 +176,19 @@ use datafusion::common::tree_node::{Transformed, TreeNode};
use datafusion::logical_expr::{col, lit, Expr, LogicalPlan,
LogicalPlanBuilder};
use datafusion::optimizer::{OptimizerRule, OptimizerConfig, OptimizerContext,
Optimizer};
+# fn rewrite_add_one(expr: Expr) -> Result<Transformed<Expr>> {
+# expr.transform(&|expr| {
+# Ok(match expr {
+# Expr::ScalarFunction(scalar_func) if
scalar_func.func.inner().name() == "add_one" => {
+# let input_arg = scalar_func.args[0].clone();
+# let new_expression = input_arg + lit(1i64);
+#
+# Transformed::yes(new_expression)
+# }
+# _ => Transformed::no(expr),
+# })
+# })
+# }
#[derive(Default, Debug)]
struct AddOneInliner {}
@@ -200,6 +231,74 @@ We're almost there. Let's just test our rule works
properly.
Testing the rule is fairly simple, we can create a SessionState with our rule
and then create a DataFrame and run a query. The logical plan will be optimized
by our rule.
```rust
+# use std::sync::Arc;
+# use datafusion::common::Result;
+# use datafusion::common::tree_node::{Transformed, TreeNode};
+# use datafusion::logical_expr::{col, lit, Expr, LogicalPlan,
LogicalPlanBuilder};
+# use datafusion::optimizer::{OptimizerRule, OptimizerConfig,
OptimizerContext, Optimizer};
+# use datafusion::arrow::array::{ArrayRef, Int64Array};
+# use datafusion::common::cast::as_int64_array;
+# use datafusion::logical_expr::ColumnarValue;
+# use datafusion::logical_expr::{Volatility, create_udf};
+# use datafusion::arrow::datatypes::DataType;
+#
+# fn rewrite_add_one(expr: Expr) -> Result<Transformed<Expr>> {
+# expr.transform(&|expr| {
+# Ok(match expr {
+# Expr::ScalarFunction(scalar_func) if
scalar_func.func.inner().name() == "add_one" => {
+# let input_arg = scalar_func.args[0].clone();
+# let new_expression = input_arg + lit(1i64);
+#
+# Transformed::yes(new_expression)
+# }
+# _ => Transformed::no(expr),
+# })
+# })
+# }
+#
+# #[derive(Default, Debug)]
+# struct AddOneInliner {}
+#
+# impl OptimizerRule for AddOneInliner {
+# fn name(&self) -> &str {
+# "add_one_inliner"
+# }
+#
+# fn rewrite(
+# &self,
+# plan: LogicalPlan,
+# _config: &dyn OptimizerConfig,
+# ) -> Result<Transformed<LogicalPlan>> {
+# // Map over the expressions and rewrite them
+# let new_expressions: Vec<Expr> = plan
+# .expressions()
+# .into_iter()
+# .map(|expr| rewrite_add_one(expr))
+# .collect::<Result<Vec<_>>>()? // returns Vec<Transformed<Expr>>
+# .into_iter()
+# .map(|transformed| transformed.data)
+# .collect();
+#
+# let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
+#
+# let plan: Result<LogicalPlan> = plan.with_new_exprs(new_expressions,
inputs);
+#
+# plan.map(|p| Transformed::yes(p))
+# }
+# }
+#
+# pub fn add_one(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+# // Error handling omitted for brevity
+# let args = ColumnarValue::values_to_arrays(args)?;
+# let i64s = as_int64_array(&args[0])?;
+#
+# let new_array = i64s
+# .iter()
+# .map(|array_elem| array_elem.map(|value| value + 1))
+# .collect::<Int64Array>();
+#
+# Ok(ColumnarValue::from(Arc::new(new_array) as ArrayRef))
+# }
use datafusion::execution::context::SessionContext;
diff --git a/_sources/user-guide/expressions.md.txt
b/_sources/user-guide/expressions.md.txt
index 199f0cb472..56e4369a9b 100644
--- a/_sources/user-guide/expressions.md.txt
+++ b/_sources/user-guide/expressions.md.txt
@@ -29,6 +29,7 @@ Most functions and methods may receive and return an `Expr`,
which can be chaine
use datafusion::prelude::*;
// create the expression `(a > 6) AND (b < 7)`
col("a").gt(lit(6)).and(col("b").lt(lit(7)));
+
```
:::
diff --git a/_sources/user-guide/sql/prepared_statements.md.txt
b/_sources/user-guide/sql/prepared_statements.md.txt
index f283573311..6677b212fd 100644
--- a/_sources/user-guide/sql/prepared_statements.md.txt
+++ b/_sources/user-guide/sql/prepared_statements.md.txt
@@ -83,6 +83,12 @@ EXECUTE greater_than(20);
**Rust Example**
```rust
+# use datafusion::prelude::*;
+# #[tokio::main]
+# async fn main() -> datafusion::error::Result<()> {
+# let ctx = SessionContext::new();
+# ctx.register_csv("example", "tests/data/example.csv",
CsvReadOptions::new()).await?;
+#
// Create the prepared statement `greater_than`
let prepare_sql = "PREPARE greater_than AS SELECT * FROM example WHERE a >
$1";
ctx.sql(prepare_sql).await?;
@@ -90,6 +96,9 @@ EXECUTE greater_than(20);
// Execute the prepared statement `greater_than`
let execute_sql = "EXECUTE greater_than(20)";
let df = ctx.sql(execute_sql).await?;
+#
+# Ok(())
+# }
```
## Positional Arguments
@@ -113,6 +122,11 @@ EXECUTE greater_than(20, 23.3);
**Rust Example**
```rust
+# use datafusion::prelude::*;
+# #[tokio::main]
+# async fn main() -> datafusion::error::Result<()> {
+# let ctx = SessionContext::new();
+# ctx.register_csv("example", "tests/data/example.csv",
CsvReadOptions::new()).await?;
// Create the prepared statement `greater_than`
let prepare_sql = "PREPARE greater_than(INT, DOUBLE) AS SELECT * FROM
example WHERE a > $1 AND b > $2";
ctx.sql(prepare_sql).await?;
@@ -120,4 +134,6 @@ EXECUTE greater_than(20, 23.3);
// Execute the prepared statement `greater_than`
let execute_sql = "EXECUTE greater_than(20, 23.3)";
let df = ctx.sql(execute_sql).await?;
+# Ok(())
+# }
```
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
