set msrv via Cargo.toml, use 2024 edition

adriangb · adriangb · commit ab2a38440cc5 · 2025-09-23T09:52:06.000-07:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -8,7 +8,8 @@ datafusion-proto = { version = "49.0.0" }
 [package]
 name = "datafusion-distributed"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
+rust-version = "1.86.0"
 
 [dependencies]
 chrono = { version = "0.4.42" }
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
diff --git a/src/common/ttl_map.rs b/src/common/ttl_map.rs
@@ -27,10 +27,10 @@ use dashmap::{DashMap, Entry};
 use datafusion::error::DataFusionError;
 use std::collections::HashSet;
 use std::hash::Hash;
+use std::sync::Arc;
 use std::sync::atomic::AtomicU64;
 #[cfg(test)]
 use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
-use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 
@@ -289,7 +289,7 @@ where
 mod tests {
     use super::*;
     use std::sync::atomic::Ordering;
-    use tokio::time::{sleep, Duration};
+    use tokio::time::{Duration, sleep};
 
     #[tokio::test]
     async fn test_basic_insert_and_get() {
diff --git a/src/config_extension_ext.rs b/src/config_extension_ext.rs
@@ -1,4 +1,4 @@
-use datafusion::common::{internal_datafusion_err, DataFusionError};
+use datafusion::common::{DataFusionError, internal_datafusion_err};
 use datafusion::config::ConfigExtension;
 use datafusion::prelude::SessionConfig;
 use http::{HeaderMap, HeaderName};
@@ -84,8 +84,8 @@ impl ContextGrpcMetadata {
 #[cfg(test)]
 mod tests {
     use crate::config_extension_ext::{
-        set_distributed_option_extension, set_distributed_option_extension_from_headers,
-        ContextGrpcMetadata,
+        ContextGrpcMetadata, set_distributed_option_extension,
+        set_distributed_option_extension_from_headers,
     };
     use datafusion::common::extensions_options;
     use datafusion::config::ConfigExtension;
diff --git a/src/distributed_ext.rs b/src/distributed_ext.rs
@@ -1,9 +1,9 @@
+use crate::ChannelResolver;
 use crate::channel_resolver_ext::set_distributed_channel_resolver;
 use crate::config_extension_ext::{
     set_distributed_option_extension, set_distributed_option_extension_from_headers,
 };
 use crate::protobuf::set_distributed_user_codec;
-use crate::ChannelResolver;
 use datafusion::common::DataFusionError;
 use datafusion::config::ConfigExtension;
 use datafusion::execution::{SessionState, SessionStateBuilder};
diff --git a/src/distributed_physical_optimizer_rule.rs b/src/distributed_physical_optimizer_rule.rs
@@ -12,7 +12,7 @@ use datafusion::{
     config::ConfigOptions,
     error::Result,
     physical_optimizer::PhysicalOptimizerRule,
-    physical_plan::{repartition::RepartitionExec, ExecutionPlan},
+    physical_plan::{ExecutionPlan, repartition::RepartitionExec},
 };
 use uuid::Uuid;
 
diff --git a/src/execution_plans/arrow_flight_read.rs b/src/execution_plans/arrow_flight_read.rs
@@ -1,15 +1,15 @@
+use crate::ChannelResolver;
 use crate::channel_resolver_ext::get_distributed_channel_resolver;
 use crate::config_extension_ext::ContextGrpcMetadata;
 use crate::errors::{map_flight_to_datafusion_error, map_status_to_datafusion_error};
 use crate::execution_plans::StageExec;
 use crate::flight_service::DoGet;
 use crate::metrics::proto::MetricsSetProto;
-use crate::protobuf::{proto_from_stage, DistributedCodec, StageKey};
-use crate::ChannelResolver;
+use crate::protobuf::{DistributedCodec, StageKey, proto_from_stage};
+use arrow_flight::Ticket;
 use arrow_flight::decode::FlightRecordBatchStream;
 use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_client::FlightServiceClient;
-use arrow_flight::Ticket;
 use dashmap::DashMap;
 use datafusion::arrow::datatypes::SchemaRef;
 use datafusion::common::{exec_err, internal_datafusion_err, internal_err, plan_err};
@@ -27,8 +27,8 @@ use prost::Message;
 use std::any::Any;
 use std::fmt::Formatter;
 use std::sync::Arc;
-use tonic::metadata::MetadataMap;
 use tonic::Request;
+use tonic::metadata::MetadataMap;
 
 /// This node has two variants.
 /// 1. Pending: it acts as a placeholder for the distributed optimization step to mark it as ready.
@@ -158,7 +158,9 @@ impl ExecutionPlan for ArrowFlightReadExec {
         context: Arc<TaskContext>,
     ) -> Result<SendableRecordBatchStream, DataFusionError> {
         let ArrowFlightReadExec::Ready(self_ready) = self else {
-            return exec_err!("ArrowFlightReadExec is not ready, was the distributed optimization step performed?");
+            return exec_err!(
+                "ArrowFlightReadExec is not ready, was the distributed optimization step performed?"
+            );
         };
 
         // get the channel manager and current stage from our context
diff --git a/src/execution_plans/metrics.rs b/src/execution_plans/metrics.rs
@@ -4,7 +4,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use crate::execution_plans::{ArrowFlightReadExec, StageExec};
-use crate::metrics::proto::{metrics_set_proto_to_df, MetricsSetProto};
+use crate::metrics::proto::{MetricsSetProto, metrics_set_proto_to_df};
 use crate::protobuf::StageKey;
 use datafusion::common::internal_err;
 use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeRewriter};
@@ -143,7 +143,11 @@ impl TaskMetricsRewriter {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let transformed = plan.rewrite(&mut self)?;
         if self.idx != self.metrics.len() {
-            return internal_err!("too many metrics sets provided to rewrite task: {} metrics sets provided, {} nodes in plan", self.metrics.len(), self.idx);
+            return internal_err!(
+                "too many metrics sets provided to rewrite task: {} metrics sets provided, {} nodes in plan",
+                self.metrics.len(),
+                self.idx
+            );
         }
         Ok(transformed.data)
     }
@@ -269,11 +273,11 @@ mod tests {
     use datafusion::arrow::array::{Int32Array, StringArray};
     use datafusion::arrow::record_batch::RecordBatch;
 
-    use crate::test_utils::in_memory_channel_resolver::InMemoryChannelResolver;
-    use crate::test_utils::session_context::register_temp_parquet_table;
     use crate::DistributedExt;
     use crate::DistributedPhysicalOptimizerRule;
-    use datafusion::execution::{context::SessionContext, SessionStateBuilder};
+    use crate::test_utils::in_memory_channel_resolver::InMemoryChannelResolver;
+    use crate::test_utils::session_context::register_temp_parquet_table;
+    use datafusion::execution::{SessionStateBuilder, context::SessionContext};
     use datafusion::physical_plan::metrics::MetricValue;
     use datafusion::prelude::SessionConfig;
     use datafusion::{
diff --git a/src/execution_plans/mod.rs b/src/execution_plans/mod.rs
@@ -5,4 +5,4 @@ mod stage;
 
 pub use arrow_flight_read::ArrowFlightReadExec;
 pub use partition_isolator::{PartitionGroup, PartitionIsolatorExec};
-pub use stage::{display_plan_graphviz, ExecutionTask, StageExec};
+pub use stage::{ExecutionTask, StageExec, display_plan_graphviz};
diff --git a/src/execution_plans/stage.rs b/src/execution_plans/stage.rs
@@ -7,7 +7,7 @@ use datafusion::common::{
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::{
-    displayable, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties,
+    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, displayable,
 };
 use datafusion::prelude::SessionContext;
 use itertools::Itertools;
diff --git a/src/flight_service/do_get.rs b/src/flight_service/do_get.rs
@@ -3,18 +3,18 @@ use crate::errors::datafusion_error_to_tonic_status;
 use crate::execution_plans::{PartitionGroup, StageExec};
 use crate::flight_service::service::ArrowFlightEndpoint;
 use crate::flight_service::session_builder::DistributedSessionBuilderContext;
-use crate::protobuf::{stage_from_proto, DistributedCodec, StageExecProto, StageKey};
+use crate::protobuf::{DistributedCodec, StageExecProto, StageKey, stage_from_proto};
+use arrow_flight::Ticket;
 use arrow_flight::encode::FlightDataEncoderBuilder;
 use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_server::FlightService;
-use arrow_flight::Ticket;
 use datafusion::execution::{SendableRecordBatchStream, SessionState};
 use futures::TryStreamExt;
 use http::HeaderMap;
 use prost::Message;
 use std::fmt::Display;
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use tokio::sync::OnceCell;
 use tonic::{Request, Response, Status};
 
@@ -172,17 +172,17 @@ fn record_batch_stream_to_response(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::ExecutionTask;
     use crate::flight_service::session_builder::DefaultSessionBuilder;
     use crate::protobuf::proto_from_stage;
-    use crate::ExecutionTask;
     use arrow::datatypes::{Schema, SchemaRef};
     use arrow_flight::Ticket;
     use datafusion::physical_expr::Partitioning;
+    use datafusion::physical_plan::ExecutionPlan;
     use datafusion::physical_plan::empty::EmptyExec;
     use datafusion::physical_plan::repartition::RepartitionExec;
-    use datafusion::physical_plan::ExecutionPlan;
     use datafusion_proto::physical_plan::DefaultPhysicalExtensionCodec;
-    use prost::{bytes::Bytes, Message};
+    use prost::{Message, bytes::Bytes};
     use tonic::Request;
     use uuid::Uuid;
 
diff --git a/src/flight_service/service.rs b/src/flight_service/service.rs
@@ -1,6 +1,6 @@
 use crate::common::ttl_map::{TTLMap, TTLMapConfig};
-use crate::flight_service::do_get::TaskData;
 use crate::flight_service::DistributedSessionBuilder;
+use crate::flight_service::do_get::TaskData;
 use crate::protobuf::StageKey;
 use arrow_flight::flight_service_server::FlightService;
 use arrow_flight::{
diff --git a/src/lib.rs b/src/lib.rs
@@ -18,7 +18,7 @@ pub use channel_resolver_ext::{BoxCloneSyncChannel, ChannelResolver};
 pub use distributed_ext::DistributedExt;
 pub use distributed_physical_optimizer_rule::DistributedPhysicalOptimizerRule;
 pub use execution_plans::{
-    display_plan_graphviz, ArrowFlightReadExec, ExecutionTask, PartitionIsolatorExec, StageExec,
+    ArrowFlightReadExec, ExecutionTask, PartitionIsolatorExec, StageExec, display_plan_graphviz,
 };
 pub use flight_service::{
     ArrowFlightEndpoint, DefaultSessionBuilder, DistributedSessionBuilder,
diff --git a/src/protobuf/distributed_codec.rs b/src/protobuf/distributed_codec.rs
@@ -5,9 +5,9 @@ use datafusion::arrow::datatypes::Schema;
 use datafusion::execution::FunctionRegistry;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionConfig;
+use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use datafusion_proto::physical_plan::from_proto::parse_protobuf_partitioning;
 use datafusion_proto::physical_plan::to_proto::serialize_partitioning;
-use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use datafusion_proto::protobuf;
 use datafusion_proto::protobuf::proto_error;
 use prost::Message;
@@ -171,8 +171,8 @@ mod tests {
     use datafusion::physical_expr::LexOrdering;
     use datafusion::{
         execution::registry::MemoryFunctionRegistry,
-        physical_expr::{expressions::col, expressions::Column, Partitioning, PhysicalSortExpr},
-        physical_plan::{displayable, sorts::sort::SortExec, union::UnionExec, ExecutionPlan},
+        physical_expr::{Partitioning, PhysicalSortExpr, expressions::Column, expressions::col},
+        physical_plan::{ExecutionPlan, displayable, sorts::sort::SortExec, union::UnionExec},
     };
 
     fn schema_i32(name: &str) -> Arc<Schema> {
diff --git a/src/protobuf/mod.rs b/src/protobuf/mod.rs
@@ -3,5 +3,5 @@ mod stage_proto;
 mod user_codec;
 
 pub(crate) use distributed_codec::DistributedCodec;
-pub(crate) use stage_proto::{proto_from_stage, stage_from_proto, StageExecProto, StageKey};
+pub(crate) use stage_proto::{StageExecProto, StageKey, proto_from_stage, stage_from_proto};
 pub(crate) use user_codec::{get_distributed_user_codec, set_distributed_user_codec};
diff --git a/src/protobuf/stage_proto.rs b/src/protobuf/stage_proto.rs
@@ -2,7 +2,7 @@ use crate::execution_plans::{ExecutionTask, StageExec};
 use datafusion::{
     common::internal_datafusion_err,
     error::{DataFusionError, Result},
-    execution::{runtime_env::RuntimeEnv, FunctionRegistry},
+    execution::{FunctionRegistry, runtime_env::RuntimeEnv},
     physical_plan::ExecutionPlan,
 };
 use datafusion_proto::{
@@ -152,9 +152,9 @@ pub fn stage_from_proto(
 mod tests {
     use std::sync::Arc;
 
+    use crate::StageExec;
     use crate::protobuf::stage_proto::StageExecProto;
     use crate::protobuf::{proto_from_stage, stage_from_proto};
-    use crate::StageExec;
     use datafusion::{
         arrow::{
             array::{RecordBatch, StringArray, UInt8Array},
diff --git a/src/test_utils/localhost.rs b/src/test_utils/localhost.rs
@@ -5,10 +5,10 @@ use crate::{
 };
 use arrow_flight::flight_service_server::FlightServiceServer;
 use async_trait::async_trait;
-use datafusion::common::runtime::JoinSet;
 use datafusion::common::DataFusionError;
-use datafusion::execution::runtime_env::RuntimeEnv;
+use datafusion::common::runtime::JoinSet;
 use datafusion::execution::SessionStateBuilder;
+use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::prelude::SessionContext;
 use std::error::Error;
 use std::sync::Arc;
diff --git a/tests/custom_config_extension.rs b/tests/custom_config_extension.rs
@@ -12,7 +12,7 @@ mod tests {
     use datafusion::physical_plan::repartition::RepartitionExec;
     use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
     use datafusion::physical_plan::{
-        execute_stream, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+        DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, execute_stream,
     };
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
     use datafusion_distributed::{ArrowFlightReadExec, DistributedPhysicalOptimizerRule};
diff --git a/tests/custom_extension_codec.rs b/tests/custom_extension_codec.rs
@@ -10,7 +10,7 @@ mod tests {
         FunctionRegistry, SendableRecordBatchStream, SessionState, SessionStateBuilder, TaskContext,
     };
     use datafusion::logical_expr::Operator;
-    use datafusion::physical_expr::expressions::{col, lit, BinaryExpr};
+    use datafusion::physical_expr::expressions::{BinaryExpr, col, lit};
     use datafusion::physical_expr::{
         EquivalenceProperties, LexOrdering, Partitioning, PhysicalSortExpr,
     };
@@ -20,16 +20,16 @@ mod tests {
     use datafusion::physical_plan::sorts::sort::SortExec;
     use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
     use datafusion::physical_plan::{
-        displayable, execute_stream, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+        DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, displayable, execute_stream,
     };
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
+    use datafusion_distributed::{ArrowFlightReadExec, DistributedPhysicalOptimizerRule};
     use datafusion_distributed::{
-        assert_snapshot, DistributedExt, DistributedSessionBuilderContext,
+        DistributedExt, DistributedSessionBuilderContext, assert_snapshot,
     };
-    use datafusion_distributed::{ArrowFlightReadExec, DistributedPhysicalOptimizerRule};
     use datafusion_proto::physical_plan::PhysicalExtensionCodec;
     use datafusion_proto::protobuf::proto_error;
-    use futures::{stream, TryStreamExt};
+    use futures::{TryStreamExt, stream};
     use prost::Message;
     use std::any::Any;
     use std::fmt::Formatter;
diff --git a/tests/distributed_aggregation.rs b/tests/distributed_aggregation.rs
@@ -6,7 +6,7 @@ mod tests {
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
     use datafusion_distributed::test_utils::parquet::register_parquet_tables;
     use datafusion_distributed::{
-        assert_snapshot, DefaultSessionBuilder, DistributedPhysicalOptimizerRule,
+        DefaultSessionBuilder, DistributedPhysicalOptimizerRule, assert_snapshot,
     };
     use futures::TryStreamExt;
     use std::error::Error;
diff --git a/tests/error_propagation.rs b/tests/error_propagation.rs
@@ -10,7 +10,7 @@ mod tests {
     use datafusion::physical_plan::repartition::RepartitionExec;
     use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
     use datafusion::physical_plan::{
-        execute_stream, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+        DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, execute_stream,
     };
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
     use datafusion_distributed::{
@@ -19,7 +19,7 @@ mod tests {
     };
     use datafusion_proto::physical_plan::PhysicalExtensionCodec;
     use datafusion_proto::protobuf::proto_error;
-    use futures::{stream, TryStreamExt};
+    use futures::{TryStreamExt, stream};
     use prost::Message;
     use std::any::Any;
     use std::error::Error;
diff --git a/tests/highly_distributed_query.rs b/tests/highly_distributed_query.rs
@@ -6,8 +6,8 @@ mod tests {
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
     use datafusion_distributed::test_utils::parquet::register_parquet_tables;
     use datafusion_distributed::{
-        assert_snapshot, ArrowFlightReadExec, DefaultSessionBuilder,
-        DistributedPhysicalOptimizerRule,
+        ArrowFlightReadExec, DefaultSessionBuilder, DistributedPhysicalOptimizerRule,
+        assert_snapshot,
     };
     use futures::TryStreamExt;
     use std::error::Error;
diff --git a/tests/tpch_validation_test.rs b/tests/tpch_validation_test.rs
@@ -7,7 +7,7 @@ mod tests {
     use datafusion_distributed::test_utils::localhost::start_localhost_context;
     use datafusion_distributed::test_utils::tpch;
     use datafusion_distributed::{
-        assert_snapshot, DistributedPhysicalOptimizerRule, DistributedSessionBuilderContext,
+        DistributedPhysicalOptimizerRule, DistributedSessionBuilderContext, assert_snapshot,
     };
     use futures::TryStreamExt;
     use std::error::Error;
