diff --git a/Cargo.toml b/Cargo.toml index fef23162ffac4..52ffc917843d0 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 e0e302e466840..a8e202888794f 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 8aaa8025d1c3a..e847f7fdb501b 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::() { diff --git a/datafusion-cli/src/exec.rs b/datafusion-cli/src/exec.rs index f43854821b2d5..1ea9995947ba1 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::() { - 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 4293788e0c03a..e2ba992961c40 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 a67738520b010..a2d7d7699927f 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>, schema: SchemaRef, @@ -207,7 +207,7 @@ impl TableProvider for CustomDataSource { _filters: &[Expr], _limit: Option, ) -> Result> { - 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 16814752b3ec2..a24ca2238181d 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 { // 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> { 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 { 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 2581f4a2ce247..6077a982c320d 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 = 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 31bb234e287f5..8ee59fa14d9cd 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 3f525b5d46afa..d534eda24ba64 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 c087019c687c5..2366b36fa8b40 100644 --- a/datafusion-examples/examples/query_planning/expr_api.rs +++ b/datafusion-examples/examples/query_planning/expr_api.rs @@ -57,7 +57,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 d3f99aedceb3d..2e4310082c9dd 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 7fdc4a7952d68..e88b996b0ae29 100644 --- a/datafusion-examples/examples/query_planning/pruning.rs +++ b/datafusion-examples/examples/query_planning/pruning.rs @@ -43,7 +43,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 ca5060896f787..d9ad7791af67c 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 3b80518d32dcd..6ddf6fa31820a 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 96434fc379ea6..848d5a3c3e5de 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 c78b1ea494407..d8ca0d58b8d21 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 651a15d776e4d..90d7eb3b41388 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 08c7463e211c6..ed5a1f6f6adb9 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 { 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 { + fn create_catalog_schema(&self, cmd: CreateCatalogSchema) -> Result { let CreateCatalogSchema { schema_name, if_not_exists, @@ -1028,7 +1022,7 @@ impl SessionContext { } } - async fn create_catalog(&self, cmd: CreateCatalog) -> Result { + fn create_catalog(&self, cmd: CreateCatalog) -> Result { let CreateCatalog { catalog_name, if_not_exists, @@ -1078,7 +1072,7 @@ impl SessionContext { } } - async fn drop_schema(&self, cmd: DropCatalogSchema) -> Result { + fn drop_schema(&self, cmd: DropCatalogSchema) -> Result { 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 { + fn drop_function(&self, stmt: &DropFunction) -> Result { // 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 c53495421307b..72e983ecd1e0a 100644 --- a/datafusion/core/src/test_util/parquet.rs +++ b/datafusion/core/src/test_util/parquet.rs @@ -149,7 +149,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, diff --git a/datafusion/core/tests/fuzz_cases/pruning.rs b/datafusion/core/tests/fuzz_cases/pruning.rs index 8ce5207f91190..7624c97cf47f7 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, - schema: Arc, - row_groups: Vec>, + schema: &Arc, + row_groups: &[Vec], ) -> 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 ebbe4312b1e1a..d6e38b5d01995 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 5dfcd50c014c9..dabb2f35b24b1 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 1cc4bb32d9eba..7066a4147c017 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 { } /// 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, @@ -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 6f7e2e328d0c3..be45ab38dabad 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, ) -> Result<(Vec, Vec)> { // 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 e9ad978b2e0cb..9338fcc0bce35 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 { 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 2db9f18f31f7f..dfb877e1b7d57 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -1191,8 +1191,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. @@ -1239,13 +1239,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 { @@ -1332,13 +1332,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] @@ -1387,13 +1387,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]; @@ -1487,12 +1487,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 8733b9e87b57a..c6c3f02829c43 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 ede40d5c4ceca..b209e91cc81e7 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 { @@ -348,7 +348,7 @@ pub async fn run_snapshot_test( test_data: &TestData, sql: &str, ) -> Result> { - 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 { @@ -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 { @@ -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 { @@ -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 { 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 dd91267d583fe..5b552e5369ef7 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 { Ok(arg1) } diff --git a/datafusion/datasource-csv/src/file_format.rs b/datafusion/datasource-csv/src/file_format.rs index 89c3d374e68fc..a7f01f6ffec13 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>, ) -> BoxStream<'a, Result> { diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index ad1caa59b8d32..56abf52144028 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) -> Result<()> { + fn cache_metadata(&self, metadata: Arc) -> 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 f15f67aab0a87..df2f17c6be22d 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, @@ -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 a84984d192d1f..e271145e03de6 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 2f63ac05e8c0b..8fe400fe828bc 100644 --- a/datafusion/physical-plan/src/limit.rs +++ b/datafusion/physical-plan/src/limit.rs @@ -749,80 +749,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, ) -> Result> { @@ -850,7 +843,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, ) -> Result> { @@ -881,7 +874,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> { diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index 685a5e4bbf55a..afab2656ee471 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, ) -> 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 cd51dc47ef5fc..da0beb0c29a28 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 e0aaa91ef6369..a3a7803e52190 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"); @@ -366,7 +366,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); @@ -398,7 +398,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(); @@ -414,7 +414,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 = @@ -464,7 +464,7 @@ fn table_with_many_types() -> Arc { } /// 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 4cd856fc562e8..47a944504c510 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, ) -> Result { 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) -> Result { + fn build_unary_expr(fn_name: &str, args: Vec) -> Result { 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) -> Result { + fn build_binary_expr(fn_name: &str, args: Vec) -> Result { 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) -> Result { + fn build_between_expr(fn_name: &str, args: Vec) -> Result { 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, ) -> Result { 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, ) -> Result { diff --git a/datafusion/substrait/src/serializer.rs b/datafusion/substrait/src/serializer.rs index ee71bc3121afe..bcc9f5cf50eac 100644 --- a/datafusion/substrait/src/serializer.rs +++ b/datafusion/substrait/src/serializer.rs @@ -70,12 +70,12 @@ pub async fn deserialize(path: impl AsRef) -> Result> { 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) -> Result> { - Ok(Box::new(Message::decode(&*proto_bytes).map_err(|e| { +pub fn deserialize_bytes(proto_bytes: &[u8]) -> Result> { + 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 018e1aef80ea1..f084d3170edcc 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) -> 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, - 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<()> {