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:08 UTC
(datafusion) branch gh-readonly-queue/main/pr-23679-efa84e8309ced6614e62d1607899c47ed0b6b08a created (now 7269d13e63)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23679-efa84e8309ced6614e62d1607899c47ed0b6b08a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 7269d13e63 chore: Enable `unused_async` lint, make some functions sync (#23679)
This branch includes the following new commits:
new 7269d13e63 chore: Enable `unused_async` lint, make some functions sync (#23679)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: chore: Enable `unused_async` lint, make some functions sync (#23679)
Posted by gi...@apache.org.
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