You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/24 18:49:09 UTC
(datafusion) 01/01: chore: Enable `unused_async` lint, make some functions sync (#23679)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23679-efa84e8309ced6614e62d1607899c47ed0b6b08a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 7269d13e63aa6b67d5db08655134d9c6f9c71ce8
Author: Neil Conway <ne...@gmail.com>
AuthorDate: Fri Jul 24 14:48:43 2026 -0400
chore: Enable `unused_async` lint, make some functions sync (#23679)
## Which issue does this PR close?
- Closes: N/A
## Rationale for this change
There are a reasonable number of places where we have functions marked
as `async` that don't need to be so. Fix this by enabling the
`unused_async` lint, and fixing up the resulting breakage. There are a
handful of spots that need an `expect(clippy::unused_async)` -- mostly
mocks of `async` methods and example code.
## What changes are included in this PR?
* Enable `unused_async` lint
* Remove unnecessary `async` annotations in a bunch of places
* Add `expect(clippy::unused_async)` where necessary
## Are these changes tested?
Yes, covered by existing tests (no behavioral change expected).
## Are there any user-facing changes?
Yes: this PR updates a few public APIs:
- `datafusion_cli::command::OutputFormat::execute`
- `datafusion::test_util::parquet::TestParquetFile::create_scan`
-
`datafusion_datasource_csv::file_format::CsvFormat::read_to_delimited_chunks_from_stream`
- `datafusion_substrait::serializer::deserialize_bytes`
Migration is mostly straightforward (e.g., removing `await` from calling
code).
---
Cargo.toml | 1 +
benchmarks/src/imdb/run.rs | 4 +-
datafusion-cli/src/command.rs | 2 +-
datafusion-cli/src/exec.rs | 2 +-
datafusion-cli/src/object_storage.rs | 12 +++--
.../custom_data_source/custom_datasource.rs | 4 +-
.../examples/data_io/remote_catalog.rs | 3 ++
.../examples/proto/composed_extension_codec.rs | 10 ++--
.../examples/proto/expression_deduplication.rs | 2 +-
datafusion-examples/examples/proto/main.rs | 4 +-
.../examples/query_planning/expr_api.rs | 2 +-
.../examples/query_planning/main.rs | 4 +-
.../examples/query_planning/pruning.rs | 2 +-
datafusion/catalog/src/information_schema.rs | 4 +-
datafusion/core/benches/filter_query_sql.rs | 13 +++--
datafusion/core/benches/struct_query_sql.rs | 5 +-
datafusion/core/benches/topk_aggregate.rs | 42 +++++------------
datafusion/core/src/datasource/file_format/csv.rs | 3 +-
datafusion/core/src/execution/context/mod.rs | 34 ++++++-------
datafusion/core/src/test_util/parquet.rs | 2 +-
datafusion/core/tests/fuzz_cases/pruning.rs | 19 +++-----
datafusion/core/tests/memory_limit/mod.rs | 10 ++--
datafusion/core/tests/parquet/filter_pushdown.rs | 1 -
datafusion/core/tests/parquet/mod.rs | 8 ++--
datafusion/core/tests/parquet/schema_coercion.rs | 6 +--
.../tests/physical_optimizer/enforce_sorting.rs | 14 +++---
.../tests/physical_optimizer/join_selection.rs | 26 +++++-----
datafusion/core/tests/sql/aggregates/dict_nulls.rs | 8 ++--
datafusion/core/tests/sql/aggregates/mod.rs | 45 +++++++++---------
.../user_defined_async_scalar_functions.rs | 1 +
datafusion/datasource-csv/src/file_format.rs | 3 +-
datafusion/datasource-parquet/src/metadata.rs | 6 +--
datafusion/datasource-parquet/src/sink.rs | 16 +++----
datafusion/execution/src/async_stream.rs | 4 ++
datafusion/physical-plan/src/limit.rs | 55 ++++++++++------------
datafusion/physical-plan/src/sorts/sort.rs | 12 ++---
datafusion/sqllogictest/bin/sqllogictests.rs | 12 +++++
datafusion/sqllogictest/src/test_context.rs | 18 +++----
.../logical_plan/consumer/expr/scalar_function.rs | 40 ++++++++--------
datafusion/substrait/src/serializer.rs | 6 +--
.../tests/cases/roundtrip_logical_plan.rs | 11 ++---
docs/source/library-user-guide/upgrading/55.0.0.md | 29 ++++++++++++
42 files changed, 251 insertions(+), 254 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index 6f4c10f8e7..87c23cc456 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -222,6 +222,7 @@ needless_pass_by_value = "warn"
# https://github.com/apache/datafusion/issues/18881
allow_attributes = "warn"
assigning_clones = "warn"
+unused_async = "warn"
[workspace.lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = [
diff --git a/benchmarks/src/imdb/run.rs b/benchmarks/src/imdb/run.rs
index e0e302e466..a8e2028887 100644
--- a/benchmarks/src/imdb/run.rs
+++ b/benchmarks/src/imdb/run.rs
@@ -355,7 +355,7 @@ impl RunOpt {
async fn register_tables(&self, ctx: &SessionContext) -> Result<()> {
for table in IMDB_TABLES {
- let table_provider = { self.get_table(ctx, table).await? };
+ let table_provider = { self.get_table(ctx, table)? };
if self.mem_table {
println!("Loading table '{table}' into memory");
@@ -416,7 +416,7 @@ impl RunOpt {
Ok(result)
}
- async fn get_table(
+ fn get_table(
&self,
ctx: &SessionContext,
table: &str,
diff --git a/datafusion-cli/src/command.rs b/datafusion-cli/src/command.rs
index 8aaa8025d1..e847f7fdb5 100644
--- a/datafusion-cli/src/command.rs
+++ b/datafusion-cli/src/command.rs
@@ -259,7 +259,7 @@ impl FromStr for OutputFormat {
}
impl OutputFormat {
- pub async fn execute(&self, print_options: &mut PrintOptions) -> Result<()> {
+ pub fn execute(&self, print_options: &mut PrintOptions) -> Result<()> {
match self {
Self::ChangeFormat(format) => {
if let Ok(format) = format.parse::<PrintFormat>() {
diff --git a/datafusion-cli/src/exec.rs b/datafusion-cli/src/exec.rs
index bc2c15f48d..fc230d5362 100644
--- a/datafusion-cli/src/exec.rs
+++ b/datafusion-cli/src/exec.rs
@@ -148,7 +148,7 @@ pub async fn exec_from_repl(
Command::OutputFormat(subcommand) => {
if let Some(subcommand) = subcommand {
if let Ok(command) = subcommand.parse::<OutputFormat>() {
- if let Err(e) = command.execute(print_options).await {
+ if let Err(e) = command.execute(print_options) {
eprintln!("{e}")
}
} else {
diff --git a/datafusion-cli/src/object_storage.rs b/datafusion-cli/src/object_storage.rs
index 4293788e0c..e2ba992961 100644
--- a/datafusion-cli/src/object_storage.rs
+++ b/datafusion-cli/src/object_storage.rs
@@ -56,6 +56,10 @@ use object_store::aws::resolve_bucket_region;
// Provide a local mock when running tests so we don't make network calls
#[cfg(test)]
+#[expect(
+ clippy::unused_async,
+ reason = "matches object_store::aws::resolve_bucket_region"
+)]
async fn resolve_bucket_region(
_bucket: &str,
_client_options: &ClientOptions,
@@ -600,7 +604,7 @@ mod tests {
#[tokio::test]
async fn s3_object_store_builder_default() -> Result<()> {
- if let Err(DataFusionError::Execution(e)) = check_aws_envs().await {
+ if let Err(DataFusionError::Execution(e)) = check_aws_envs() {
// Skip test if AWS envs are not set
eprintln!("{e}");
return Ok(());
@@ -765,7 +769,7 @@ mod tests {
#[tokio::test]
async fn s3_object_store_builder_resolves_region_when_none_provided() -> Result<()> {
- if let Err(DataFusionError::Execution(e)) = check_aws_envs().await {
+ if let Err(DataFusionError::Execution(e)) = check_aws_envs() {
// Skip test if AWS envs are not set
eprintln!("{e}");
return Ok(());
@@ -798,7 +802,7 @@ mod tests {
#[tokio::test]
async fn s3_object_store_builder_overrides_region_when_resolve_region_enabled()
-> Result<()> {
- if let Err(DataFusionError::Execution(e)) = check_aws_envs().await {
+ if let Err(DataFusionError::Execution(e)) = check_aws_envs() {
// Skip test if AWS envs are not set
eprintln!("{e}");
return Ok(());
@@ -909,7 +913,7 @@ mod tests {
table_options
}
- async fn check_aws_envs() -> Result<()> {
+ fn check_aws_envs() -> Result<()> {
let aws_envs = [
"AWS_ACCESS_KEY_ID",
"AWS_SECRET_ACCESS_KEY",
diff --git a/datafusion-examples/examples/custom_data_source/custom_datasource.rs b/datafusion-examples/examples/custom_data_source/custom_datasource.rs
index a67738520b..a2d7d76999 100644
--- a/datafusion-examples/examples/custom_data_source/custom_datasource.rs
+++ b/datafusion-examples/examples/custom_data_source/custom_datasource.rs
@@ -145,7 +145,7 @@ impl Debug for CustomDataSource {
}
impl CustomDataSource {
- pub(crate) async fn create_physical_plan(
+ pub(crate) fn create_physical_plan(
&self,
projections: Option<&Vec<usize>>,
schema: SchemaRef,
@@ -207,7 +207,7 @@ impl TableProvider for CustomDataSource {
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
- return self.create_physical_plan(projection, self.schema()).await;
+ self.create_physical_plan(projection, self.schema())
}
}
diff --git a/datafusion-examples/examples/data_io/remote_catalog.rs b/datafusion-examples/examples/data_io/remote_catalog.rs
index 16814752b3..a24ca22381 100644
--- a/datafusion-examples/examples/data_io/remote_catalog.rs
+++ b/datafusion-examples/examples/data_io/remote_catalog.rs
@@ -130,6 +130,7 @@ struct RemoteCatalogInterface {}
impl RemoteCatalogInterface {
/// Establish a connection to the remote catalog
+ #[expect(clippy::unused_async)]
pub async fn connect() -> Result<Self> {
// In a real implementation this method might connect to a remote
// catalog, validate credentials, cache basic information, etc
@@ -137,6 +138,7 @@ impl RemoteCatalogInterface {
}
/// Fetches information for a specific table
+ #[expect(clippy::unused_async)]
pub async fn table_info(&self, name: &str) -> Result<Option<SchemaRef>> {
if name != "remote_table" {
return Ok(None);
@@ -155,6 +157,7 @@ impl RemoteCatalogInterface {
}
/// Fetches data for a table from a remote data source
+ #[expect(clippy::unused_async)]
pub async fn read_data(&self, name: &str) -> Result<SendableRecordBatchStream> {
if name != "remote_table" {
return plan_err!("Remote table not found: {}", name);
diff --git a/datafusion-examples/examples/proto/composed_extension_codec.rs b/datafusion-examples/examples/proto/composed_extension_codec.rs
index 2581f4a2ce..6077a982c3 100644
--- a/datafusion-examples/examples/proto/composed_extension_codec.rs
+++ b/datafusion-examples/examples/proto/composed_extension_codec.rs
@@ -47,8 +47,8 @@ use datafusion_proto::physical_plan::{
use datafusion_proto::protobuf;
/// Example of using multiple extension codecs for serialization / deserialization
-pub async fn composed_extension_codec() -> Result<()> {
- // build execution plan that has both types of nodes
+pub fn composed_extension_codec() -> Result<()> {
+ // Build execution plan that has both types of nodes
//
// Note each node requires a different `PhysicalExtensionCodec` to decode
let exec_plan = Arc::new(ParentExec {
@@ -63,18 +63,18 @@ pub async fn composed_extension_codec() -> Result<()> {
Arc::new(ChildPhysicalExtensionCodec {}),
]);
- // serialize execution plan to proto
+ // Serialize execution plan to proto
let proto: protobuf::PhysicalPlanNode =
protobuf::PhysicalPlanNode::try_from_physical_plan(
exec_plan.clone(),
&composed_codec,
)?;
- // deserialize proto back to execution plan
+ // Deserialize proto back to execution plan
let result_exec_plan: Arc<dyn ExecutionPlan> =
proto.try_into_physical_plan(&ctx.task_ctx(), &composed_codec)?;
- // assert that the original and deserialized execution plans are equal
+ // Assert that the original and deserialized execution plans are equal
assert_eq!(format!("{exec_plan:?}"), format!("{result_exec_plan:?}"));
Ok(())
diff --git a/datafusion-examples/examples/proto/expression_deduplication.rs b/datafusion-examples/examples/proto/expression_deduplication.rs
index 31bb234e28..8ee59fa14d 100644
--- a/datafusion-examples/examples/proto/expression_deduplication.rs
+++ b/datafusion-examples/examples/proto/expression_deduplication.rs
@@ -72,7 +72,7 @@ use prost::Message;
/// In real scenarios, expressions can be much more complex, e.g. a large InList
/// expression could be megabytes in size, so deduplication can save significant memory
/// in addition to more correctly representing the original plan structure.
-pub async fn expression_deduplication() -> Result<()> {
+pub fn expression_deduplication() -> Result<()> {
println!("=== Expression Deduplication Example ===\n");
// Create a schema for our test expressions
diff --git a/datafusion-examples/examples/proto/main.rs b/datafusion-examples/examples/proto/main.rs
index 3f525b5d46..d534eda24b 100644
--- a/datafusion-examples/examples/proto/main.rs
+++ b/datafusion-examples/examples/proto/main.rs
@@ -64,10 +64,10 @@ impl ExampleKind {
}
}
ExampleKind::ComposedExtensionCodec => {
- composed_extension_codec::composed_extension_codec().await?
+ composed_extension_codec::composed_extension_codec()?
}
ExampleKind::ExpressionDeduplication => {
- expression_deduplication::expression_deduplication().await?
+ expression_deduplication::expression_deduplication()?
}
}
Ok(())
diff --git a/datafusion-examples/examples/query_planning/expr_api.rs b/datafusion-examples/examples/query_planning/expr_api.rs
index dd5145def3..08efff7777 100644
--- a/datafusion-examples/examples/query_planning/expr_api.rs
+++ b/datafusion-examples/examples/query_planning/expr_api.rs
@@ -58,7 +58,7 @@ use datafusion::prelude::*;
/// 5. Analyze predicates for boundary ranges: [`range_analysis_demo`]
/// 6. Get the types of the expressions: [`expression_type_demo`]
/// 7. Apply type coercion to expressions: [`type_coercion_demo`]
-pub async fn expr_api() -> Result<()> {
+pub fn expr_api() -> Result<()> {
// The easiest way to do create expressions is to use the
// "fluent"-style API:
let expr = col("a") + lit(5);
diff --git a/datafusion-examples/examples/query_planning/main.rs b/datafusion-examples/examples/query_planning/main.rs
index d3f99aedce..2e4310082c 100644
--- a/datafusion-examples/examples/query_planning/main.rs
+++ b/datafusion-examples/examples/query_planning/main.rs
@@ -94,12 +94,12 @@ impl ExampleKind {
}
}
ExampleKind::AnalyzerRule => analyzer_rule::analyzer_rule().await?,
- ExampleKind::ExprApi => expr_api::expr_api().await?,
+ ExampleKind::ExprApi => expr_api::expr_api()?,
ExampleKind::OptimizerRule => optimizer_rule::optimizer_rule().await?,
ExampleKind::ParseSqlExpr => parse_sql_expr::parse_sql_expr().await?,
ExampleKind::PlanToSql => plan_to_sql::plan_to_sql_examples().await?,
ExampleKind::PlannerApi => planner_api::planner_api().await?,
- ExampleKind::Pruning => pruning::pruning().await?,
+ ExampleKind::Pruning => pruning::pruning()?,
ExampleKind::ThreadPools => thread_pools::thread_pools().await?,
}
Ok(())
diff --git a/datafusion-examples/examples/query_planning/pruning.rs b/datafusion-examples/examples/query_planning/pruning.rs
index df26aa57b6..dad57cd261 100644
--- a/datafusion-examples/examples/query_planning/pruning.rs
+++ b/datafusion-examples/examples/query_planning/pruning.rs
@@ -44,7 +44,7 @@ use datafusion::prelude::*;
/// one might do as part of a higher level storage engine. See
/// `parquet_index.rs` for an example that uses pruning in the context of an
/// individual query.
-pub async fn pruning() -> Result<()> {
+pub fn pruning() -> Result<()> {
// In this example, we'll use the PruningPredicate to determine if
// the expression `x = 5 AND y = 10` can never be true based on statistics
diff --git a/datafusion/catalog/src/information_schema.rs b/datafusion/catalog/src/information_schema.rs
index ca5060896f..d9ad7791af 100644
--- a/datafusion/catalog/src/information_schema.rs
+++ b/datafusion/catalog/src/information_schema.rs
@@ -151,7 +151,7 @@ impl InformationSchemaConfig {
Ok(())
}
- async fn make_schemata(&self, builder: &mut InformationSchemataBuilder) {
+ fn make_schemata(&self, builder: &mut InformationSchemataBuilder) {
for catalog_name in self.catalog_list.catalog_names() {
let catalog = self.catalog_list.catalog(&catalog_name).unwrap();
@@ -1152,7 +1152,7 @@ impl PartitionStream for InformationSchemata {
Arc::clone(&self.schema),
// TODO: Stream this
futures::stream::once(async move {
- config.make_schemata(&mut builder).await;
+ config.make_schemata(&mut builder);
builder.finish()
}),
))
diff --git a/datafusion/core/benches/filter_query_sql.rs b/datafusion/core/benches/filter_query_sql.rs
index 3b80518d32..6ddf6fa318 100644
--- a/datafusion/core/benches/filter_query_sql.rs
+++ b/datafusion/core/benches/filter_query_sql.rs
@@ -23,12 +23,11 @@ use arrow::{
use criterion::{Criterion, criterion_group, criterion_main};
use datafusion::prelude::SessionContext;
use datafusion::{datasource::MemTable, error::Result};
-use futures::executor::block_on;
use std::hint::black_box;
use std::sync::Arc;
use tokio::runtime::Runtime;
-async fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) {
+fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) {
// execute the query
let df = rt.block_on(ctx.sql(sql)).unwrap();
black_box(rt.block_on(df.collect()).unwrap());
@@ -71,28 +70,28 @@ fn criterion_benchmark(c: &mut Criterion) {
c.bench_function("filter_array", |b| {
let ctx = create_context(array_len, batch_size).unwrap();
- b.iter(|| block_on(query(&ctx, &rt, "select f32, f64 from t where f32 >= f64")))
+ b.iter(|| query(&ctx, &rt, "select f32, f64 from t where f32 >= f64"))
});
c.bench_function("filter_scalar", |b| {
let ctx = create_context(array_len, batch_size).unwrap();
b.iter(|| {
- block_on(query(
+ query(
&ctx,
&rt,
"select f32, f64 from t where f32 >= 250 and f64 > 250",
- ))
+ )
})
});
c.bench_function("filter_scalar in list", |b| {
let ctx = create_context(array_len, batch_size).unwrap();
b.iter(|| {
- block_on(query(
+ query(
&ctx,
&rt,
"select f32, f64 from t where f32 in (10, 20, 30, 40)",
- ))
+ )
})
});
}
diff --git a/datafusion/core/benches/struct_query_sql.rs b/datafusion/core/benches/struct_query_sql.rs
index 96434fc379..848d5a3c3e 100644
--- a/datafusion/core/benches/struct_query_sql.rs
+++ b/datafusion/core/benches/struct_query_sql.rs
@@ -23,12 +23,11 @@ use arrow::{
use criterion::{Criterion, criterion_group, criterion_main};
use datafusion::prelude::SessionContext;
use datafusion::{datasource::MemTable, error::Result};
-use futures::executor::block_on;
use std::hint::black_box;
use std::sync::Arc;
use tokio::runtime::Runtime;
-async fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) {
+fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) {
// execute the query
let df = rt.block_on(ctx.sql(sql)).unwrap();
black_box(rt.block_on(df.collect()).unwrap());
@@ -71,7 +70,7 @@ fn criterion_benchmark(c: &mut Criterion) {
let rt = Runtime::new().unwrap();
c.bench_function("struct", |b| {
- b.iter(|| block_on(query(&ctx, &rt, "select struct(f32, f64) from t")))
+ b.iter(|| query(&ctx, &rt, "select struct(f32, f64) from t"))
});
}
diff --git a/datafusion/core/benches/topk_aggregate.rs b/datafusion/core/benches/topk_aggregate.rs
index c78b1ea494..d8ca0d58b8 100644
--- a/datafusion/core/benches/topk_aggregate.rs
+++ b/datafusion/core/benches/topk_aggregate.rs
@@ -74,7 +74,7 @@ fn test_distinct_schema() -> SchemaRef {
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]))
}
-async fn create_context(
+fn create_context(
partition_cnt: i32,
sample_cnt: i32,
asc: bool,
@@ -94,7 +94,7 @@ async fn create_context(
Ok(ctx)
}
-async fn create_context_distinct(
+fn create_context_distinct(
partition_cnt: i32,
sample_cnt: i32,
use_topk: bool,
@@ -306,12 +306,8 @@ fn assert_utf8_utf8view_match(
asc: bool,
use_topk: bool,
) {
- let ctx_utf8 = rt
- .block_on(create_context(partitions, samples, asc, use_topk, false))
- .unwrap();
- let ctx_view = rt
- .block_on(create_context(partitions, samples, asc, use_topk, true))
- .unwrap();
+ let ctx_utf8 = create_context(partitions, samples, asc, use_topk, false).unwrap();
+ let ctx_view = create_context(partitions, samples, asc, use_topk, true).unwrap();
let batches_utf8 = rt
.block_on(aggregate_string(ctx_utf8, limit, use_topk))
.unwrap();
@@ -390,15 +386,9 @@ fn criterion_benchmark(c: &mut Criterion) {
.name_tpl
.replace("{rows}", &total_rows.to_string())
.replace("{limit}", &limit.to_string());
- let ctx = rt
- .block_on(create_context(
- partitions,
- samples,
- case.asc,
- case.use_topk,
- case.use_view,
- ))
- .unwrap();
+ let ctx =
+ create_context(partitions, samples, case.asc, case.use_topk, case.use_view)
+ .unwrap();
c.bench_function(&name, |b| {
b.iter(|| run(&rt, ctx.clone(), limit, case.use_topk, case.asc))
});
@@ -462,15 +452,9 @@ fn criterion_benchmark(c: &mut Criterion) {
} else {
format!("string aggregate {total_rows} {scenario} rows [{type_label}]")
};
- let ctx = rt
- .block_on(create_context(
- partitions,
- samples,
- case.asc,
- case.use_topk,
- case.use_view,
- ))
- .unwrap();
+ let ctx =
+ create_context(partitions, samples, case.asc, case.use_topk, case.use_view)
+ .unwrap();
c.bench_function(&name, |b| {
b.iter(|| run_string(&rt, ctx.clone(), limit, case.use_topk))
});
@@ -478,11 +462,7 @@ fn criterion_benchmark(c: &mut Criterion) {
// DISTINCT benchmarks
for use_topk in [false, true] {
- let ctx = rt.block_on(async {
- create_context_distinct(partitions, samples, use_topk)
- .await
- .unwrap()
- });
+ let ctx = create_context_distinct(partitions, samples, use_topk).unwrap();
let topk_label = if use_topk { "TopK" } else { "no TopK" };
for asc in [false, true] {
let dir = if asc { "asc" } else { "desc" };
diff --git a/datafusion/core/src/datasource/file_format/csv.rs b/datafusion/core/src/datasource/file_format/csv.rs
index 651a15d776..90d7eb3b41 100644
--- a/datafusion/core/src/datasource/file_format/csv.rs
+++ b/datafusion/core/src/datasource/file_format/csv.rs
@@ -591,8 +591,7 @@ mod tests {
//convert compressed_stream to decoded_stream
let decoded_stream = compressed_csv
- .read_to_delimited_chunks_from_stream(compressed_stream.unwrap())
- .await;
+ .read_to_delimited_chunks_from_stream(compressed_stream.unwrap());
let (schema, records_read) = compressed_csv
.infer_schema_from_stream(&session_state, records_to_read, decoded_stream)
.await?;
diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs
index 281cb4dd79..cd30193e30 100644
--- a/datafusion/core/src/execution/context/mod.rs
+++ b/datafusion/core/src/execution/context/mod.rs
@@ -687,8 +687,8 @@ impl SessionContext {
pub async fn execute_logical_plan(&self, plan: LogicalPlan) -> Result<DataFrame> {
match plan {
LogicalPlan::Ddl(ddl) => {
- // Box::pin avoids allocating the stack space within this function's frame
- // for every one of these individual async functions, decreasing the risk of
+ // Box async DDL handlers to avoid reserving space for all of their
+ // futures in this function's state machine, decreasing the risk of
// stack overflows.
match ddl {
DdlStatement::CreateExternalTable(cmd) => {
@@ -703,32 +703,26 @@ impl SessionContext {
Box::pin(self.create_view(cmd)).await
}
DdlStatement::CreateCatalogSchema(cmd) => {
- Box::pin(self.create_catalog_schema(cmd)).await
- }
- DdlStatement::CreateCatalog(cmd) => {
- Box::pin(self.create_catalog(cmd)).await
+ self.create_catalog_schema(cmd)
}
+ DdlStatement::CreateCatalog(cmd) => self.create_catalog(cmd),
DdlStatement::DropTable(cmd) => Box::pin(self.drop_table(cmd)).await,
DdlStatement::DropView(cmd) => Box::pin(self.drop_view(cmd)).await,
- DdlStatement::DropCatalogSchema(cmd) => {
- Box::pin(self.drop_schema(cmd)).await
- }
+ DdlStatement::DropCatalogSchema(cmd) => self.drop_schema(cmd),
DdlStatement::CreateFunction(cmd) => {
Box::pin(self.create_function(*cmd)).await
}
- DdlStatement::DropFunction(cmd) => {
- Box::pin(self.drop_function(cmd)).await
- }
+ DdlStatement::DropFunction(cmd) => self.drop_function(&cmd),
ddl => Ok(DataFrame::new(self.state(), LogicalPlan::Ddl(ddl))),
}
}
// TODO what about the other statements (like TransactionStart and TransactionEnd)
LogicalPlan::Statement(Statement::SetVariable(stmt)) => {
- self.set_variable(stmt).await?;
+ self.set_variable(stmt)?;
self.return_empty_dataframe()
}
LogicalPlan::Statement(Statement::ResetVariable(stmt)) => {
- self.reset_variable(stmt).await?;
+ self.reset_variable(stmt)?;
self.return_empty_dataframe()
}
LogicalPlan::Statement(Statement::Prepare(Prepare {
@@ -987,7 +981,7 @@ impl SessionContext {
Ok(())
}
- async fn create_catalog_schema(&self, cmd: CreateCatalogSchema) -> Result<DataFrame> {
+ fn create_catalog_schema(&self, cmd: CreateCatalogSchema) -> Result<DataFrame> {
let CreateCatalogSchema {
schema_name,
if_not_exists,
@@ -1028,7 +1022,7 @@ impl SessionContext {
}
}
- async fn create_catalog(&self, cmd: CreateCatalog) -> Result<DataFrame> {
+ fn create_catalog(&self, cmd: CreateCatalog) -> Result<DataFrame> {
let CreateCatalog {
catalog_name,
if_not_exists,
@@ -1078,7 +1072,7 @@ impl SessionContext {
}
}
- async fn drop_schema(&self, cmd: DropCatalogSchema) -> Result<DataFrame> {
+ fn drop_schema(&self, cmd: DropCatalogSchema) -> Result<DataFrame> {
let DropCatalogSchema {
name,
if_exists: allow_missing,
@@ -1113,7 +1107,7 @@ impl SessionContext {
exec_err!("Schema '{schema_ref}' doesn't exist.")
}
- async fn set_variable(&self, stmt: SetVariable) -> Result<()> {
+ fn set_variable(&self, stmt: SetVariable) -> Result<()> {
let SetVariable {
variable, value, ..
} = stmt;
@@ -1148,7 +1142,7 @@ impl SessionContext {
Ok(())
}
- async fn reset_variable(&self, stmt: ResetVariable) -> Result<()> {
+ fn reset_variable(&self, stmt: ResetVariable) -> Result<()> {
let variable = stmt.variable;
if variable.starts_with("datafusion.runtime.") {
return self.reset_runtime_variable(&variable);
@@ -1531,7 +1525,7 @@ impl SessionContext {
self.return_empty_dataframe()
}
- async fn drop_function(&self, stmt: DropFunction) -> Result<DataFrame> {
+ fn drop_function(&self, stmt: &DropFunction) -> Result<DataFrame> {
// we don't know function type at this point
// decision has been made to drop all functions
let mut dropped = false;
diff --git a/datafusion/core/src/test_util/parquet.rs b/datafusion/core/src/test_util/parquet.rs
index d1018f3fb0..e25fe74669 100644
--- a/datafusion/core/src/test_util/parquet.rs
+++ b/datafusion/core/src/test_util/parquet.rs
@@ -150,7 +150,7 @@ impl TestParquetFile {
/// ```
///
/// Otherwise if `maybe_filter` is None, return just a `DataSourceExec`
- pub async fn create_scan(
+ pub fn create_scan(
&self,
ctx: &SessionContext,
maybe_filter: Option<Expr>,
diff --git a/datafusion/core/tests/fuzz_cases/pruning.rs b/datafusion/core/tests/fuzz_cases/pruning.rs
index 8ce5207f91..7624c97cf4 100644
--- a/datafusion/core/tests/fuzz_cases/pruning.rs
+++ b/datafusion/core/tests/fuzz_cases/pruning.rs
@@ -249,12 +249,7 @@ impl Utf8Test {
for (idx, truncation_length) in [Some(1), Some(2), None].iter().enumerate() {
// parquet files only support 32767 row groups per file, so chunk up into multiple files so we don't error if running on a large number of row groups
for (rg_idx, row_groups) in row_groups.chunks(32766).enumerate() {
- let buf = write_parquet_file(
- *truncation_length,
- Arc::clone(&schema),
- row_groups.to_vec(),
- )
- .await;
+ let buf = write_parquet_file(*truncation_length, &schema, row_groups);
let filename = format!("test_fuzz_utf8_{idx}_{rg_idx}.parquet");
let size = buf.len();
let path = Path::from(filename);
@@ -314,10 +309,10 @@ async fn execute_with_predicate(
values
}
-async fn write_parquet_file(
+fn write_parquet_file(
truncation_length: Option<usize>,
- schema: Arc<Schema>,
- row_groups: Vec<Vec<String>>,
+ schema: &Arc<Schema>,
+ row_groups: &[Vec<String>],
) -> Bytes {
let mut buf = BytesMut::new().writer();
let props = WriterProperties::builder()
@@ -326,11 +321,11 @@ async fn write_parquet_file(
let props = props.build();
{
let mut writer =
- ArrowWriter::try_new(&mut buf, schema.clone(), Some(props)).unwrap();
- for rg_values in row_groups.iter() {
+ ArrowWriter::try_new(&mut buf, Arc::clone(schema), Some(props)).unwrap();
+ for rg_values in row_groups {
let arr = StringArray::from_iter_values(rg_values.iter());
let batch =
- RecordBatch::try_new(schema.clone(), vec![Arc::new(arr)]).unwrap();
+ RecordBatch::try_new(Arc::clone(schema), vec![Arc::new(arr)]).unwrap();
writer.write(&batch).unwrap();
writer.flush().unwrap(); // finishes the current row group and starts a new one
}
diff --git a/datafusion/core/tests/memory_limit/mod.rs b/datafusion/core/tests/memory_limit/mod.rs
index ebbe4312b1..d6e38b5d01 100644
--- a/datafusion/core/tests/memory_limit/mod.rs
+++ b/datafusion/core/tests/memory_limit/mod.rs
@@ -614,7 +614,7 @@ async fn test_sort_skewed_batches_spill() {
// ------------------------------------------------------------------
// Create a new `SessionContext` with specified disk limit, memory pool limit, and spill compression codec
-async fn setup_context(
+fn setup_context(
disk_limit: u64,
memory_pool_limit: usize,
spill_compression: SpillCompression,
@@ -655,7 +655,7 @@ async fn setup_context(
#[tokio::test]
async fn test_disk_spill_limit_reached() -> Result<()> {
let spill_compression = SpillCompression::Uncompressed;
- let ctx = setup_context(1024 * 1024, 1024 * 1024, spill_compression).await?; // 1MB disk limit, 1MB memory limit
+ let ctx = setup_context(1024 * 1024, 1024 * 1024, spill_compression)?; // 1MB disk limit, 1MB memory limit
let df = ctx
.sql("select * from generate_series(1, 1000000000000) as t1(v1) order by v1 desc")
@@ -683,7 +683,7 @@ async fn test_disk_spill_limit_reached() -> Result<()> {
async fn test_disk_spill_limit_not_reached() -> Result<()> {
let disk_spill_limit = 1024 * 1024; // 1MB
let spill_compression = SpillCompression::Uncompressed;
- let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression).await?; // 1MB disk limit, 128KB memory limit
+ let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression)?; // 1MB disk limit, 128KB memory limit
let df = ctx
.sql("select * from generate_series(1, 10000) as t1(v1) order by v1 desc")
@@ -719,7 +719,7 @@ async fn test_disk_spill_limit_not_reached() -> Result<()> {
async fn test_spill_file_compressed_with_zstd() -> Result<()> {
let disk_spill_limit = 1024 * 1024; // 1MB
let spill_compression = SpillCompression::Zstd;
- let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression).await?; // 1MB disk limit, 128KB memory limit, zstd
+ let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression)?; // 1MB disk limit, 128KB memory limit, zstd
let df = ctx
.sql("select * from generate_series(1, 100000) as t1(v1) order by v1 desc")
@@ -755,7 +755,7 @@ async fn test_spill_file_compressed_with_zstd() -> Result<()> {
async fn test_spill_file_compressed_with_lz4_frame() -> Result<()> {
let disk_spill_limit = 1024 * 1024; // 1MB
let spill_compression = SpillCompression::Lz4Frame;
- let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression).await?; // 1MB disk limit, 128KB memory limit, lz4_frame
+ let ctx = setup_context(disk_spill_limit, 128 * 1024, spill_compression)?; // 1MB disk limit, 128KB memory limit, lz4_frame
let df = ctx
.sql("select * from generate_series(1, 100000) as t1(v1) order by v1 desc")
diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs
index 5dfcd50c01..dabb2f35b2 100644
--- a/datafusion/core/tests/parquet/filter_pushdown.rs
+++ b/datafusion/core/tests/parquet/filter_pushdown.rs
@@ -515,7 +515,6 @@ impl<'a> TestCase<'a> {
let exec = self
.test_parquet_file
.create_scan(&ctx, Some(filter.clone()))
- .await
.unwrap();
let result = collect(exec.clone(), ctx.task_ctx()).await.unwrap();
diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs
index 1cc4bb32d9..7066a4147c 100644
--- a/datafusion/core/tests/parquet/mod.rs
+++ b/datafusion/core/tests/parquet/mod.rs
@@ -330,11 +330,10 @@ impl ContextWithParquet {
custom_schema,
custom_batches,
)
- .await
}
Unit::Page(row_per_page) => {
config = config.with_parquet_page_index_pruning(true);
- make_test_file_page(scenario, row_per_page).await
+ make_test_file_page(scenario, row_per_page)
}
Unit::RowGroupAndPage(row_per_group, row_per_page) => {
config = config.with_parquet_bloom_filter_pruning(true);
@@ -347,7 +346,6 @@ impl ContextWithParquet {
custom_schema,
custom_batches,
)
- .await
}
};
let parquet_path = file.path().to_string_lossy();
@@ -1173,7 +1171,7 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
}
/// Create a test parquet file with various data types
-async fn make_test_file_rg(
+fn make_test_file_rg(
scenario: Scenario,
row_per_group: usize,
row_per_page: Option<usize>,
@@ -1219,7 +1217,7 @@ async fn make_test_file_rg(
output_file
}
-async fn make_test_file_page(scenario: Scenario, row_per_page: usize) -> NamedTempFile {
+fn make_test_file_page(scenario: Scenario, row_per_page: usize) -> NamedTempFile {
let mut output_file = tempfile::Builder::new()
.prefix("parquet_page_pruning")
.suffix(".parquet")
diff --git a/datafusion/core/tests/parquet/schema_coercion.rs b/datafusion/core/tests/parquet/schema_coercion.rs
index 6f7e2e328d..be45ab38da 100644
--- a/datafusion/core/tests/parquet/schema_coercion.rs
+++ b/datafusion/core/tests/parquet/schema_coercion.rs
@@ -53,7 +53,7 @@ async fn multi_parquet_coercion() {
// batch2: c2(int64), c3(float32)
let batch2 = RecordBatch::try_from_iter(vec![("c2", c2), ("c3", c3)]).unwrap();
- let (meta, _files) = store_parquet(vec![batch1, batch2]).await.unwrap();
+ let (meta, _files) = store_parquet(vec![batch1, batch2]).unwrap();
let file_group = meta.into_iter().map(Into::into).collect();
// cast c1 to utf8, c2 to int32, c3 to float64
@@ -107,7 +107,7 @@ async fn multi_parquet_coercion_projection() {
let batch2 =
RecordBatch::try_from_iter(vec![("c2", c2), ("c1", c1s), ("c3", c3)]).unwrap();
- let (meta, _files) = store_parquet(vec![batch1, batch2]).await.unwrap();
+ let (meta, _files) = store_parquet(vec![batch1, batch2]).unwrap();
let file_group = meta.into_iter().map(Into::into).collect();
// cast c1 to utf8, c2 to int32, c3 to float64
@@ -146,7 +146,7 @@ async fn multi_parquet_coercion_projection() {
}
/// Writes `batches` to a temporary parquet file
-pub async fn store_parquet(
+pub fn store_parquet(
batches: Vec<RecordBatch>,
) -> Result<(Vec<ObjectMeta>, Vec<NamedTempFile>)> {
// Each batch writes to their own file
diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
index e9ad978b2e..9338fcc0bc 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
@@ -425,12 +425,12 @@ async fn test_union_inputs_different_sorted2() -> Result<()> {
Ok(())
}
-#[tokio::test]
+#[test]
// Test with `repartition_sorts` enabled to preserve pre-sorted partitions and avoid resorting
-async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_repartition_sorts_true()
+fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_repartition_sorts_true()
-> Result<()> {
assert_snapshot!(
- union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(true).await?,
+ union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(true)?,
@r"
Input Plan:
OutputRequirementExec: order_by=[(nullable_col@0, asc)], dist_by=SinglePartition
@@ -451,12 +451,12 @@ async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_reparti
Ok(())
}
-#[tokio::test]
+#[test]
// Test with `repartition_sorts` disabled, causing a full resort of the data
-async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_repartition_sorts_false()
+fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_repartition_sorts_false()
-> Result<()> {
assert_snapshot!(
- union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(false).await?,
+ union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(false)?,
@r"
Input Plan:
OutputRequirementExec: order_by=[(nullable_col@0, asc)], dist_by=SinglePartition
@@ -477,7 +477,7 @@ async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_reparti
Ok(())
}
-async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(
+fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(
repartition_sorts: bool,
) -> Result<String> {
let schema = create_test_schema()?;
diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs
index cca54909a1..3827e6e98b 100644
--- a/datafusion/core/tests/physical_optimizer/join_selection.rs
+++ b/datafusion/core/tests/physical_optimizer/join_selection.rs
@@ -1271,8 +1271,8 @@ struct TestCase {
expecting_swap: bool,
}
-#[tokio::test]
-async fn test_join_with_swap_full() -> Result<()> {
+#[test]
+fn test_join_with_swap_full() -> Result<()> {
// NOTE: Currently, some initial conditions are not viable after join order selection.
// For example, full join always comes in partitioned mode. See the warning in
// function "swap". If this changes in the future, we should update these tests.
@@ -1319,13 +1319,13 @@ async fn test_join_with_swap_full() -> Result<()> {
},
];
for case in cases.into_iter() {
- test_join_with_maybe_swap_unbounded_case(case).await?
+ test_join_with_maybe_swap_unbounded_case(case)?
}
Ok(())
}
-#[tokio::test]
-async fn test_cases_without_collect_left_check() -> Result<()> {
+#[test]
+fn test_cases_without_collect_left_check() -> Result<()> {
let mut cases = vec![];
let join_types = vec![JoinType::LeftSemi, JoinType::Inner];
for join_type in join_types {
@@ -1412,13 +1412,13 @@ async fn test_cases_without_collect_left_check() -> Result<()> {
}
for case in cases.into_iter() {
- test_join_with_maybe_swap_unbounded_case(case).await?
+ test_join_with_maybe_swap_unbounded_case(case)?
}
Ok(())
}
-#[tokio::test]
-async fn test_not_support_collect_left() -> Result<()> {
+#[test]
+fn test_not_support_collect_left() -> Result<()> {
let mut cases = vec![];
// After [JoinSelection] optimization, these join types cannot run in CollectLeft mode except
// [JoinType::LeftSemi]
@@ -1467,13 +1467,13 @@ async fn test_not_support_collect_left() -> Result<()> {
}
for case in cases.into_iter() {
- test_join_with_maybe_swap_unbounded_case(case).await?
+ test_join_with_maybe_swap_unbounded_case(case)?
}
Ok(())
}
-#[tokio::test]
-async fn test_not_supporting_swaps_possible_collect_left() -> Result<()> {
+#[test]
+fn test_not_supporting_swaps_possible_collect_left() -> Result<()> {
let mut cases = vec![];
let the_ones_not_support_collect_left =
vec![JoinType::Right, JoinType::RightAnti, JoinType::RightSemi];
@@ -1567,12 +1567,12 @@ async fn test_not_supporting_swaps_possible_collect_left() -> Result<()> {
}
for case in cases.into_iter() {
- test_join_with_maybe_swap_unbounded_case(case).await?
+ test_join_with_maybe_swap_unbounded_case(case)?
}
Ok(())
}
-async fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
+fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
let left_unbounded = t.initial_sources_unbounded.0 == SourceType::Unbounded;
let right_unbounded = t.initial_sources_unbounded.1 == SourceType::Unbounded;
let left_exec = Arc::new(UnboundedExec::new(
diff --git a/datafusion/core/tests/sql/aggregates/dict_nulls.rs b/datafusion/core/tests/sql/aggregates/dict_nulls.rs
index 8733b9e87b..c6c3f02829 100644
--- a/datafusion/core/tests/sql/aggregates/dict_nulls.rs
+++ b/datafusion/core/tests/sql/aggregates/dict_nulls.rs
@@ -292,7 +292,7 @@ async fn test_first_last_value_group_by_dict_nulls() -> Result<()> {
/// Test MAX with dictionary columns containing null keys and values as specified in the SQL query
#[tokio::test]
async fn test_max_with_fuzz_table_dict_nulls() -> Result<()> {
- let (ctx_single, ctx_multi) = setup_fuzz_test_contexts().await?;
+ let (ctx_single, ctx_multi) = setup_fuzz_test_contexts()?;
// Execute the SQL query with MAX aggregations
let sql = "SELECT
@@ -333,7 +333,7 @@ async fn test_max_with_fuzz_table_dict_nulls() -> Result<()> {
/// Test MIN with fuzz table containing dictionary columns with null keys and values and timestamp data (single and multiple partitions)
#[tokio::test]
async fn test_min_timestamp_with_fuzz_table_dict_nulls() -> Result<()> {
- let (ctx_single, ctx_multi) = setup_fuzz_timestamp_test_contexts().await?;
+ let (ctx_single, ctx_multi) = setup_fuzz_timestamp_test_contexts()?;
// Execute the SQL query with MIN aggregation on timestamp
let sql = "SELECT
@@ -373,7 +373,7 @@ async fn test_min_timestamp_with_fuzz_table_dict_nulls() -> Result<()> {
/// Test COUNT and COUNT DISTINCT with fuzz table containing dictionary columns with null keys and values (single and multiple partitions)
#[tokio::test]
async fn test_count_distinct_with_fuzz_table_dict_nulls() -> Result<()> {
- let (ctx_single, ctx_multi) = setup_fuzz_count_test_contexts().await?;
+ let (ctx_single, ctx_multi) = setup_fuzz_count_test_contexts()?;
// Execute the SQL query with COUNT and COUNT DISTINCT aggregations
let sql = "SELECT
@@ -414,7 +414,7 @@ async fn test_count_distinct_with_fuzz_table_dict_nulls() -> Result<()> {
/// Test MEDIAN and MEDIAN DISTINCT with fuzz table containing various numeric types and dictionary columns with null keys and values (single and multiple partitions)
#[tokio::test]
async fn test_median_distinct_with_fuzz_table_dict_nulls() -> Result<()> {
- let (ctx_single, ctx_multi) = setup_fuzz_median_test_contexts().await?;
+ let (ctx_single, ctx_multi) = setup_fuzz_median_test_contexts()?;
// Execute the SQL query with MEDIAN and MEDIAN DISTINCT aggregations
let sql = "SELECT
diff --git a/datafusion/core/tests/sql/aggregates/mod.rs b/datafusion/core/tests/sql/aggregates/mod.rs
index ede40d5c4c..b209e91cc8 100644
--- a/datafusion/core/tests/sql/aggregates/mod.rs
+++ b/datafusion/core/tests/sql/aggregates/mod.rs
@@ -259,20 +259,20 @@ impl TestData {
}
/// Sets up test contexts for TestData with both single and multiple partitions
-pub async fn setup_test_contexts(
+pub fn setup_test_contexts(
test_data: &TestData,
) -> Result<(SessionContext, SessionContext)> {
// Single partition context
- let ctx_single = create_context_with_partitions(test_data, 1).await?;
+ let ctx_single = create_context_with_partitions(test_data, 1)?;
// Multiple partition context
- let ctx_multi = create_context_with_partitions(test_data, 3).await?;
+ let ctx_multi = create_context_with_partitions(test_data, 3)?;
Ok((ctx_single, ctx_multi))
}
/// Creates a session context with the specified number of partitions and registers test data
-pub async fn create_context_with_partitions(
+pub fn create_context_with_partitions(
test_data: &TestData,
num_partitions: usize,
) -> Result<SessionContext> {
@@ -348,7 +348,7 @@ pub async fn run_snapshot_test(
test_data: &TestData,
sql: &str,
) -> Result<Vec<RecordBatch>> {
- let (ctx_single, ctx_multi) = setup_test_contexts(test_data).await?;
+ let (ctx_single, ctx_multi) = setup_test_contexts(test_data)?;
let results = test_query_consistency(&ctx_single, &ctx_multi, sql).await?;
Ok(results)
}
@@ -430,20 +430,20 @@ impl FuzzTestData {
}
/// Sets up test contexts for fuzz table with both single and multiple partitions
-pub async fn setup_fuzz_test_contexts() -> Result<(SessionContext, SessionContext)> {
+pub fn setup_fuzz_test_contexts() -> Result<(SessionContext, SessionContext)> {
let test_data = FuzzTestData::new();
// Single partition context
- let ctx_single = create_fuzz_context_with_partitions(&test_data, 1).await?;
+ let ctx_single = create_fuzz_context_with_partitions(&test_data, 1)?;
// Multiple partition context
- let ctx_multi = create_fuzz_context_with_partitions(&test_data, 3).await?;
+ let ctx_multi = create_fuzz_context_with_partitions(&test_data, 3)?;
Ok((ctx_single, ctx_multi))
}
/// Creates a session context with fuzz table partitioned into specified number of partitions
-pub async fn create_fuzz_context_with_partitions(
+pub fn create_fuzz_context_with_partitions(
test_data: &FuzzTestData,
num_partitions: usize,
) -> Result<SessionContext> {
@@ -604,21 +604,20 @@ impl FuzzCountTestData {
}
/// Sets up test contexts for fuzz table with duration/binary columns and both single and multiple partitions
-pub async fn setup_fuzz_count_test_contexts() -> Result<(SessionContext, SessionContext)>
-{
+pub fn setup_fuzz_count_test_contexts() -> Result<(SessionContext, SessionContext)> {
let test_data = FuzzCountTestData::new();
// Single partition context
- let ctx_single = create_fuzz_count_context_with_partitions(&test_data, 1).await?;
+ let ctx_single = create_fuzz_count_context_with_partitions(&test_data, 1)?;
// Multiple partition context
- let ctx_multi = create_fuzz_count_context_with_partitions(&test_data, 3).await?;
+ let ctx_multi = create_fuzz_count_context_with_partitions(&test_data, 3)?;
Ok((ctx_single, ctx_multi))
}
/// Creates a session context with fuzz count table partitioned into specified number of partitions
-pub async fn create_fuzz_count_context_with_partitions(
+pub fn create_fuzz_count_context_with_partitions(
test_data: &FuzzCountTestData,
num_partitions: usize,
) -> Result<SessionContext> {
@@ -808,21 +807,20 @@ impl FuzzMedianTestData {
}
/// Sets up test contexts for fuzz table with numeric types for median testing and both single and multiple partitions
-pub async fn setup_fuzz_median_test_contexts() -> Result<(SessionContext, SessionContext)>
-{
+pub fn setup_fuzz_median_test_contexts() -> Result<(SessionContext, SessionContext)> {
let test_data = FuzzMedianTestData::new();
// Single partition context
- let ctx_single = create_fuzz_median_context_with_partitions(&test_data, 1).await?;
+ let ctx_single = create_fuzz_median_context_with_partitions(&test_data, 1)?;
// Multiple partition context
- let ctx_multi = create_fuzz_median_context_with_partitions(&test_data, 3).await?;
+ let ctx_multi = create_fuzz_median_context_with_partitions(&test_data, 3)?;
Ok((ctx_single, ctx_multi))
}
/// Creates a session context with fuzz median table partitioned into specified number of partitions
-pub async fn create_fuzz_median_context_with_partitions(
+pub fn create_fuzz_median_context_with_partitions(
test_data: &FuzzMedianTestData,
num_partitions: usize,
) -> Result<SessionContext> {
@@ -959,21 +957,20 @@ impl FuzzTimestampTestData {
}
/// Sets up test contexts for fuzz table with timestamps and both single and multiple partitions
-pub async fn setup_fuzz_timestamp_test_contexts()
--> Result<(SessionContext, SessionContext)> {
+pub fn setup_fuzz_timestamp_test_contexts() -> Result<(SessionContext, SessionContext)> {
let test_data = FuzzTimestampTestData::new();
// Single partition context
- let ctx_single = create_fuzz_timestamp_context_with_partitions(&test_data, 1).await?;
+ let ctx_single = create_fuzz_timestamp_context_with_partitions(&test_data, 1)?;
// Multiple partition context
- let ctx_multi = create_fuzz_timestamp_context_with_partitions(&test_data, 3).await?;
+ let ctx_multi = create_fuzz_timestamp_context_with_partitions(&test_data, 3)?;
Ok((ctx_single, ctx_multi))
}
/// Creates a session context with fuzz timestamp table partitioned into specified number of partitions
-pub async fn create_fuzz_timestamp_context_with_partitions(
+pub fn create_fuzz_timestamp_context_with_partitions(
test_data: &FuzzTimestampTestData,
num_partitions: usize,
) -> Result<SessionContext> {
diff --git a/datafusion/core/tests/user_defined/user_defined_async_scalar_functions.rs b/datafusion/core/tests/user_defined/user_defined_async_scalar_functions.rs
index dd91267d58..5b552e5369 100644
--- a/datafusion/core/tests/user_defined/user_defined_async_scalar_functions.rs
+++ b/datafusion/core/tests/user_defined/user_defined_async_scalar_functions.rs
@@ -267,6 +267,7 @@ impl AsyncScalarUDFImpl for TestAsyncUDFImpl {
}
/// Simulates calling an async external service
+#[expect(clippy::unused_async)]
async fn call_external_service(arg1: ColumnarValue) -> Result<ColumnarValue> {
Ok(arg1)
}
diff --git a/datafusion/datasource-csv/src/file_format.rs b/datafusion/datasource-csv/src/file_format.rs
index 89c3d374e6..a7f01f6ffe 100644
--- a/datafusion/datasource-csv/src/file_format.rs
+++ b/datafusion/datasource-csv/src/file_format.rs
@@ -158,7 +158,6 @@ impl CsvFormat {
.map_err(|e| DataFusionError::ObjectStore(Box::new(e)))
.boxed(),
)
- .await
.map_err(DataFusionError::from)
.left_stream(),
Err(e) => {
@@ -170,7 +169,7 @@ impl CsvFormat {
/// Convert a stream of bytes into a stream of [`Bytes`] containing newline
/// delimited CSV records, while accounting for `\` and `"`.
- pub async fn read_to_delimited_chunks_from_stream<'a>(
+ pub fn read_to_delimited_chunks_from_stream<'a>(
&self,
stream: BoxStream<'a, Result<Bytes>>,
) -> BoxStream<'a, Result<Bytes>> {
diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs
index ad1caa59b8..56abf52144 100644
--- a/datafusion/datasource-parquet/src/metadata.rs
+++ b/datafusion/datasource-parquet/src/metadata.rs
@@ -181,14 +181,14 @@ impl<'a> DFParquetMetadata<'a> {
Self::load_page_index(self.store, self.object_meta, cached_metadata)
.await?;
if cache_metadata {
- self.cache_metadata(Arc::clone(&metadata)).await?;
+ self.cache_metadata(Arc::clone(&metadata))?;
}
return Ok(metadata);
}
let metadata = self.fetch_metadata_from_store(page_index_policy).await?;
if cache_metadata {
- self.cache_metadata(Arc::clone(&metadata)).await?;
+ self.cache_metadata(Arc::clone(&metadata))?;
}
Ok(metadata)
}
@@ -207,7 +207,7 @@ impl<'a> DFParquetMetadata<'a> {
metadata.column_index().is_some() && metadata.offset_index().is_some()
}
- async fn cache_metadata(&self, metadata: Arc<ParquetMetaData>) -> Result<()> {
+ fn cache_metadata(&self, metadata: Arc<ParquetMetaData>) -> Result<()> {
if let Some(file_metadata_cache) = &self.file_metadata_cache {
file_metadata_cache.put(
&self.object_meta.location,
diff --git a/datafusion/datasource-parquet/src/sink.rs b/datafusion/datasource-parquet/src/sink.rs
index f15f67aab0..df2f17c6be 100644
--- a/datafusion/datasource-parquet/src/sink.rs
+++ b/datafusion/datasource-parquet/src/sink.rs
@@ -171,7 +171,7 @@ impl ParquetSink {
/// Creates an AsyncArrowWriter which serializes a parquet file to an ObjectStore
/// AsyncArrowWriters are used when individual parquet file serialization is not parallelized
- async fn create_async_arrow_writer(
+ fn create_async_arrow_writer(
&self,
location: &Path,
object_store: Arc<dyn ObjectStore>,
@@ -296,14 +296,12 @@ impl FileSink for ParquetSink {
if !parquet_opts.global.allow_single_file_parallelism
|| parquet_opts.global.content_defined_chunking.enabled
{
- let mut writer = self
- .create_async_arrow_writer(
- &path,
- Arc::clone(&object_store),
- context,
- parquet_props.clone(),
- )
- .await?;
+ let mut writer = self.create_async_arrow_writer(
+ &path,
+ Arc::clone(&object_store),
+ context,
+ parquet_props.clone(),
+ )?;
let reservation = MemoryConsumer::new(format!("ParquetSink[{path}]"))
.register(context.memory_pool());
file_write_tasks.spawn(
diff --git a/datafusion/execution/src/async_stream.rs b/datafusion/execution/src/async_stream.rs
index a84984d192..e271145e03 100644
--- a/datafusion/execution/src/async_stream.rs
+++ b/datafusion/execution/src/async_stream.rs
@@ -388,6 +388,7 @@ mod test {
async fn unit_emit_in_select() {
use tokio::select;
+ #[expect(clippy::unused_async)]
async fn do_stuff_async() {}
let s = async_stream(|mut emitter| async move {
@@ -405,7 +406,9 @@ mod test {
async fn emit_with_select() {
use tokio::select;
+ #[expect(clippy::unused_async)]
async fn do_stuff_async() {}
+ #[expect(clippy::unused_async)]
async fn more_async_work() {}
let s = async_stream(|mut emitter| async move {
@@ -556,6 +559,7 @@ mod test {
fn inner_try_stream() {
use tokio::select;
+ #[expect(clippy::unused_async)]
async fn do_stuff_async() {}
let _ = async_stream(|mut emitter| async move {
diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs
index a1f6074cb9..ddce680fc1 100644
--- a/datafusion/physical-plan/src/limit.rs
+++ b/datafusion/physical-plan/src/limit.rs
@@ -839,80 +839,73 @@ mod tests {
Ok(())
}
- #[tokio::test]
- async fn test_row_number_statistics_for_global_limit() -> Result<()> {
- let row_count = row_number_statistics_for_global_limit(0, Some(10)).await?;
+ #[test]
+ fn test_row_number_statistics_for_global_limit() -> Result<()> {
+ let row_count = row_number_statistics_for_global_limit(0, Some(10))?;
assert_eq!(row_count, Precision::Exact(10));
- let row_count = row_number_statistics_for_global_limit(5, Some(10)).await?;
+ let row_count = row_number_statistics_for_global_limit(5, Some(10))?;
assert_eq!(row_count, Precision::Exact(10));
- let row_count = row_number_statistics_for_global_limit(400, Some(10)).await?;
+ let row_count = row_number_statistics_for_global_limit(400, Some(10))?;
assert_eq!(row_count, Precision::Exact(0));
- let row_count = row_number_statistics_for_global_limit(398, Some(10)).await?;
+ let row_count = row_number_statistics_for_global_limit(398, Some(10))?;
assert_eq!(row_count, Precision::Exact(2));
- let row_count = row_number_statistics_for_global_limit(398, Some(1)).await?;
+ let row_count = row_number_statistics_for_global_limit(398, Some(1))?;
assert_eq!(row_count, Precision::Exact(1));
- let row_count = row_number_statistics_for_global_limit(398, None).await?;
+ let row_count = row_number_statistics_for_global_limit(398, None)?;
assert_eq!(row_count, Precision::Exact(2));
- let row_count =
- row_number_statistics_for_global_limit(0, Some(usize::MAX)).await?;
+ let row_count = row_number_statistics_for_global_limit(0, Some(usize::MAX))?;
assert_eq!(row_count, Precision::Exact(400));
- let row_count =
- row_number_statistics_for_global_limit(398, Some(usize::MAX)).await?;
+ let row_count = row_number_statistics_for_global_limit(398, Some(usize::MAX))?;
assert_eq!(row_count, Precision::Exact(2));
- let row_count =
- row_number_inexact_statistics_for_global_limit(0, Some(10)).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(0, Some(10))?;
assert_eq!(row_count, Precision::Inexact(10));
- let row_count =
- row_number_inexact_statistics_for_global_limit(5, Some(10)).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(5, Some(10))?;
assert_eq!(row_count, Precision::Inexact(10));
// Input was Inexact, so an `nr <= skip` outcome must remain Inexact:
// the inexact estimate could be wrong, so we cannot promote 0 to
// Exact.
- let row_count =
- row_number_inexact_statistics_for_global_limit(400, Some(10)).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(400, Some(10))?;
assert_eq!(row_count, Precision::Inexact(0));
- let row_count =
- row_number_inexact_statistics_for_global_limit(398, Some(10)).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(398, Some(10))?;
assert_eq!(row_count, Precision::Inexact(2));
- let row_count =
- row_number_inexact_statistics_for_global_limit(398, Some(1)).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(398, Some(1))?;
assert_eq!(row_count, Precision::Inexact(1));
- let row_count = row_number_inexact_statistics_for_global_limit(398, None).await?;
+ let row_count = row_number_inexact_statistics_for_global_limit(398, None)?;
assert_eq!(row_count, Precision::Inexact(2));
let row_count =
- row_number_inexact_statistics_for_global_limit(0, Some(usize::MAX)).await?;
+ row_number_inexact_statistics_for_global_limit(0, Some(usize::MAX))?;
assert_eq!(row_count, Precision::Inexact(400));
let row_count =
- row_number_inexact_statistics_for_global_limit(398, Some(usize::MAX)).await?;
+ row_number_inexact_statistics_for_global_limit(398, Some(usize::MAX))?;
assert_eq!(row_count, Precision::Inexact(2));
Ok(())
}
- #[tokio::test]
- async fn test_row_number_statistics_for_local_limit() -> Result<()> {
- let row_count = row_number_statistics_for_local_limit(4, 10).await?;
+ #[test]
+ fn test_row_number_statistics_for_local_limit() -> Result<()> {
+ let row_count = row_number_statistics_for_local_limit(4, 10)?;
assert_eq!(row_count, Precision::Exact(10));
Ok(())
}
- async fn row_number_statistics_for_global_limit(
+ fn row_number_statistics_for_global_limit(
skip: usize,
fetch: Option<usize>,
) -> Result<Precision<usize>> {
@@ -940,7 +933,7 @@ mod tests {
PhysicalGroupBy::new_single(group_by_expr.clone())
}
- async fn row_number_inexact_statistics_for_global_limit(
+ fn row_number_inexact_statistics_for_global_limit(
skip: usize,
fetch: Option<usize>,
) -> Result<Precision<usize>> {
@@ -971,7 +964,7 @@ mod tests {
.num_rows)
}
- async fn row_number_statistics_for_local_limit(
+ fn row_number_statistics_for_local_limit(
num_partitions: usize,
fetch: usize,
) -> Result<Precision<usize>> {
diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs
index a9b754dee6..4b30aede7d 100644
--- a/datafusion/physical-plan/src/sorts/sort.rs
+++ b/datafusion/physical-plan/src/sorts/sort.rs
@@ -406,7 +406,7 @@ impl ExternalSorter {
/// Appending globally sorted batches to the in-progress spill file, and clears
/// the `globally_sorted_batches` (also its memory reservation) afterwards.
- async fn consume_and_spill_append(
+ fn consume_and_spill_append(
&mut self,
globally_sorted_batches: &mut Vec<RecordBatch>,
) -> Result<()> {
@@ -445,7 +445,7 @@ impl ExternalSorter {
}
/// Finishes the in-progress spill file and moves it to the finished spill files.
- async fn spill_finish(&mut self) -> Result<()> {
+ fn spill_finish(&mut self) -> Result<()> {
let (mut in_progress_file, max_record_batch_memory) =
self.in_progress_spill_file.take().ok_or_else(|| {
internal_datafusion_err!("Should be called after `spill_append`")
@@ -500,8 +500,7 @@ impl ExternalSorter {
// already in memory, so it's okay to combine it with previously
// sorted batches, and spill together.
globally_sorted_batches.push(batch);
- self.consume_and_spill_append(&mut globally_sorted_batches)
- .await?; // reservation is freed in spill()
+ self.consume_and_spill_append(&mut globally_sorted_batches)?; // reservation is freed in spill()
} else {
globally_sorted_batches.push(batch);
}
@@ -511,9 +510,8 @@ impl ExternalSorter {
// upcoming `self.reserve_memory_for_merge()` may fail due to insufficient memory.
drop(sorted_stream);
- self.consume_and_spill_append(&mut globally_sorted_batches)
- .await?;
- self.spill_finish().await?;
+ self.consume_and_spill_append(&mut globally_sorted_batches)?;
+ self.spill_finish()?;
// Sanity check after spilling
let buffers_cleared_property =
diff --git a/datafusion/sqllogictest/bin/sqllogictests.rs b/datafusion/sqllogictest/bin/sqllogictests.rs
index cd51dc47ef..da0beb0c29 100644
--- a/datafusion/sqllogictest/bin/sqllogictests.rs
+++ b/datafusion/sqllogictest/bin/sqllogictests.rs
@@ -473,6 +473,10 @@ async fn run_test_file_substrait_round_trip(
}
#[cfg(not(feature = "substrait"))]
+#[expect(
+ clippy::unused_async,
+ reason = "matches the substrait-enabled implementation"
+)]
async fn run_test_file_substrait_round_trip(
_test_file: TestFile,
_validator: Validator,
@@ -646,6 +650,10 @@ async fn run_test_file_with_postgres(
}
#[cfg(not(feature = "postgres"))]
+#[expect(
+ clippy::unused_async,
+ reason = "matches the postgres-enabled implementation"
+)]
async fn run_test_file_with_postgres(
_test_file: TestFile,
_validator: Validator,
@@ -771,6 +779,10 @@ async fn run_complete_file_with_postgres(
}
#[cfg(not(feature = "postgres"))]
+#[expect(
+ clippy::unused_async,
+ reason = "matches the postgres-enabled implementation"
+)]
async fn run_complete_file_with_postgres(
_test_file: TestFile,
_validator: Validator,
diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs
index fdb04edc05..99c3179ef1 100644
--- a/datafusion/sqllogictest/src/test_context.rs
+++ b/datafusion/sqllogictest/src/test_context.rs
@@ -142,15 +142,15 @@ impl TestContext {
}
"information_schema_table_types.slt" => {
info!("Registering local temporary table");
- register_temp_table(test_ctx.session_ctx()).await;
+ register_temp_table(test_ctx.session_ctx());
}
"information_schema_columns.slt" => {
info!("Registering table with many types");
- register_table_with_many_types(test_ctx.session_ctx()).await;
+ register_table_with_many_types(test_ctx.session_ctx());
}
"map.slt" => {
info!("Registering table with map");
- register_table_with_map(test_ctx.session_ctx()).await;
+ register_table_with_map(test_ctx.session_ctx());
}
"avro.slt" => {
#[cfg(feature = "avro")]
@@ -173,7 +173,7 @@ impl TestContext {
test_ctx.ctx.register_udf(example_udf);
register_partition_table(&mut test_ctx).await;
info!("Registering table with many types");
- register_table_with_many_types(test_ctx.session_ctx()).await;
+ register_table_with_many_types(test_ctx.session_ctx());
}
"range_partitioning.slt" => {
info!("Registering range partitioned table");
@@ -181,7 +181,7 @@ impl TestContext {
}
"metadata.slt" | "arrow_field.slt" => {
info!("Registering metadata table tables");
- register_metadata_tables(test_ctx.session_ctx()).await;
+ register_metadata_tables(test_ctx.session_ctx());
}
"union_function.slt" => {
info!("Registering table with union column");
@@ -370,7 +370,7 @@ pub async fn register_partition_table(test_ctx: &mut TestContext) {
}
// registers a LOCAL TEMPORARY table.
-pub async fn register_temp_table(ctx: &SessionContext) {
+pub fn register_temp_table(ctx: &SessionContext) {
#[derive(Debug)]
struct TestTable(TableType);
@@ -402,7 +402,7 @@ pub async fn register_temp_table(ctx: &SessionContext) {
.unwrap();
}
-pub async fn register_table_with_many_types(ctx: &SessionContext) {
+pub fn register_table_with_many_types(ctx: &SessionContext) {
let catalog = MemoryCatalogProvider::new();
let schema = MemorySchemaProvider::new();
@@ -418,7 +418,7 @@ pub async fn register_table_with_many_types(ctx: &SessionContext) {
.unwrap();
}
-pub async fn register_table_with_map(ctx: &SessionContext) {
+pub fn register_table_with_map(ctx: &SessionContext) {
let key = Field::new("key", DataType::Int64, false);
let value = Field::new("value", DataType::Int64, true);
let map_field =
@@ -468,7 +468,7 @@ fn table_with_many_types() -> Arc<dyn TableProvider> {
}
/// Registers a table_with_metadata that contains both field level and Table level metadata
-pub async fn register_metadata_tables(ctx: &SessionContext) {
+pub fn register_metadata_tables(ctx: &SessionContext) {
let id = Field::new("id", DataType::Int32, true).with_metadata(HashMap::from([(
String::from("metadata_key"),
String::from("the id field"),
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/scalar_function.rs b/datafusion/substrait/src/logical_plan/consumer/expr/scalar_function.rs
index 4cd856fc56..47a944504c 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/scalar_function.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/scalar_function.rs
@@ -88,7 +88,7 @@ pub async fn from_scalar_function(
// In those cases we build a balanced tree of BinaryExprs
arg_list_to_binary_op_tree(op, args)
} else if let Some(builder) = BuiltinExprBuilder::try_from_name(fn_name) {
- builder.build(consumer, f, args).await
+ builder.build(consumer, f, args)
} else {
not_impl_err!("Unsupported function name: {fn_name:?}")
}
@@ -206,34 +206,32 @@ impl BuiltinExprBuilder {
}
}
- pub async fn build(
+ pub fn build(
self,
consumer: &impl SubstraitConsumer,
f: &ScalarFunction,
args: Vec<Expr>,
) -> Result<Expr> {
match self.expr_name.as_str() {
- "like" => Self::build_like_expr(false, false, f, args).await,
- "ilike" => Self::build_like_expr(true, false, f, args).await,
- "like_match" => Self::build_like_expr(false, false, f, args).await,
- "like_imatch" => Self::build_like_expr(true, false, f, args).await,
- "like_not_match" => Self::build_like_expr(false, true, f, args).await,
- "like_not_imatch" => Self::build_like_expr(true, true, f, args).await,
+ "like" => Self::build_like_expr(false, false, f, args),
+ "ilike" => Self::build_like_expr(true, false, f, args),
+ "like_match" => Self::build_like_expr(false, false, f, args),
+ "like_imatch" => Self::build_like_expr(true, false, f, args),
+ "like_not_match" => Self::build_like_expr(false, true, f, args),
+ "like_not_imatch" => Self::build_like_expr(true, true, f, args),
"not" | "negative" | "negate" | "is_null" | "is_not_null" | "is_true"
| "is_false" | "is_not_true" | "is_not_false" | "is_unknown"
- | "is_not_unknown" => Self::build_unary_expr(&self.expr_name, args).await,
- "and_not" | "xor" => Self::build_binary_expr(&self.expr_name, args).await,
- "between" => Self::build_between_expr(&self.expr_name, args).await,
- "logb" => {
- Self::build_custom_handling_expr(consumer, &self.expr_name, args).await
- }
+ | "is_not_unknown" => Self::build_unary_expr(&self.expr_name, args),
+ "and_not" | "xor" => Self::build_binary_expr(&self.expr_name, args),
+ "between" => Self::build_between_expr(&self.expr_name, args),
+ "logb" => Self::build_custom_handling_expr(consumer, &self.expr_name, args),
_ => {
not_impl_err!("Unsupported builtin expression: {}", self.expr_name)
}
}
}
- async fn build_unary_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
+ fn build_unary_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
let [arg] = match args.try_into() {
Ok(args_arr) => args_arr,
Err(_) => return substrait_err!("Expected one argument for {fn_name} expr"),
@@ -257,7 +255,7 @@ impl BuiltinExprBuilder {
Ok(expr)
}
- async fn build_like_expr(
+ fn build_like_expr(
case_insensitive: bool,
negated: bool,
f: &ScalarFunction,
@@ -306,7 +304,7 @@ impl BuiltinExprBuilder {
}))
}
- async fn build_binary_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
+ fn build_binary_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
let [a, b] = match args.try_into() {
Ok(args_arr) => args_arr,
Err(_) => {
@@ -330,7 +328,7 @@ impl BuiltinExprBuilder {
Self::build_and_not_expr(or_expr, and_expr)
}
- async fn build_between_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
+ fn build_between_expr(fn_name: &str, args: Vec<Expr>) -> Result<Expr> {
let [expression, low, high] = match args.try_into() {
Ok(args_arr) => args_arr,
Err(_) => {
@@ -347,18 +345,18 @@ impl BuiltinExprBuilder {
}
//This handles any functions that require custom handling
- async fn build_custom_handling_expr(
+ fn build_custom_handling_expr(
consumer: &impl SubstraitConsumer,
fn_name: &str,
args: Vec<Expr>,
) -> Result<Expr> {
match fn_name {
- "logb" => Self::build_logb_expr(consumer, args).await,
+ "logb" => Self::build_logb_expr(consumer, args),
_ => not_impl_err!("Unsupported custom handled expression: {}", fn_name),
}
}
- async fn build_logb_expr(
+ fn build_logb_expr(
consumer: &impl SubstraitConsumer,
args: Vec<Expr>,
) -> Result<Expr> {
diff --git a/datafusion/substrait/src/serializer.rs b/datafusion/substrait/src/serializer.rs
index ee71bc3121..bcc9f5cf50 100644
--- a/datafusion/substrait/src/serializer.rs
+++ b/datafusion/substrait/src/serializer.rs
@@ -70,12 +70,12 @@ pub async fn deserialize(path: impl AsRef<Path>) -> Result<Box<Plan>> {
let mut file = OpenOptions::new().read(true).open(path).await?;
file.read_to_end(&mut protobuf_in).await?;
- deserialize_bytes(protobuf_in).await
+ deserialize_bytes(&protobuf_in)
}
/// Deserializes a plan from the bytes.
-pub async fn deserialize_bytes(proto_bytes: Vec<u8>) -> Result<Box<Plan>> {
- Ok(Box::new(Message::decode(&*proto_bytes).map_err(|e| {
+pub fn deserialize_bytes(proto_bytes: &[u8]) -> Result<Box<Plan>> {
+ Ok(Box::new(Message::decode(proto_bytes).map_err(|e| {
DataFusionError::Substrait(format!("Failed to decode plan: {e}"))
})?))
}
diff --git a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
index 018e1aef80..f084d3170e 100644
--- a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
@@ -2224,7 +2224,7 @@ fn check_post_join_filters(rel: &Rel) -> Result<()> {
}
}
-async fn verify_post_join_filter_value(proto: Box<Plan>) -> Result<()> {
+fn verify_post_join_filter_value(proto: &Plan) -> Result<()> {
for relation in &proto.relations {
match relation.rel_type.as_ref() {
Some(rt) => match rt {
@@ -2263,10 +2263,7 @@ fn count_read_filters(rel: &Rel, filter_count: &mut u32) -> Result<()> {
}
}
-async fn assert_read_filter_count(
- proto: Box<Plan>,
- expected_filter_count: u32,
-) -> Result<()> {
+fn assert_read_filter_count(proto: &Plan, expected_filter_count: u32) -> Result<()> {
let mut filter_count: u32 = 0;
for relation in &proto.relations {
match relation.rel_type.as_ref() {
@@ -2644,7 +2641,7 @@ async fn roundtrip_verify_post_join_filter(sql: &str) -> Result<()> {
let proto = roundtrip_with_ctx(sql, ctx).await?;
// verify that the join filters are None
- verify_post_join_filter_value(proto).await
+ verify_post_join_filter_value(&proto)
}
async fn roundtrip_verify_read_filter_count(
@@ -2655,7 +2652,7 @@ async fn roundtrip_verify_read_filter_count(
let proto = roundtrip_with_ctx(sql, ctx).await?;
// verify that filter counts in read relations are as expected
- assert_read_filter_count(proto, expected_filter_count).await
+ assert_read_filter_count(&proto, expected_filter_count)
}
async fn roundtrip_all_types(sql: &str) -> Result<()> {
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md
index 5f1609af30..6ca4d71a78 100644
--- a/docs/source/library-user-guide/upgrading/55.0.0.md
+++ b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -847,6 +847,35 @@ fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
See [PR #23703](https://github.com/apache/datafusion/pull/23703) for details.
+### Unused `async` removed from several public functions
+
+Public functions that were declared `async` but never awaited anything are now
+synchronous:
+
+- `CsvFormat::read_to_delimited_chunks_from_stream` (in
+ `datafusion_datasource_csv`, re-exported as
+ `datafusion::datasource::file_format::csv::CsvFormat`)
+- `datafusion_substrait::serializer::deserialize_bytes`, which now also borrows
+ its input as `&[u8]` instead of taking an owned `Vec<u8>`
+- `datafusion::test_util::parquet::TestParquetFile::create_scan`
+
+**Migration guide:**
+
+Remove `.await` from call sites; the compiler flags each one, since `.await`
+on a non-future value does not compile:
+
+```rust,ignore
+// Before
+let stream = csv_format
+ .read_to_delimited_chunks_from_stream(input)
+ .await;
+let plan = deserialize_bytes(proto_bytes).await?;
+
+// After
+let stream = csv_format.read_to_delimited_chunks_from_stream(input);
+let plan = deserialize_bytes(&proto_bytes)?;
+```
+
### `MSRV` updated to 1.94.0
The Minimum Supported Rust Version (MSRV) has been updated to [`1.94.0`].
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org