Create Clippy rule dis-allowing std::collection::hashmap (datafusion-contrib#447)

see datafusion-contrib#445

Author: Rich-T-kid

Cargo.toml

@@ -59,12 +59,7 @@ hyper-util = { version = "0.1.16", optional = true }
 
 [features]
 avro = ["datafusion/avro"]
-integration = [
-  "insta",
-  "parquet",
-  "arrow",
-  "hyper-util",
-]
+integration = ["insta", "parquet", "arrow", "hyper-util"]
 
 system-metrics = ["sysinfo"]
 
@@ -83,3 +78,7 @@ tokio-stream = { version = "0.1.17", features = ["sync"] }
 hyper-util = "0.1.16"
 pretty_assertions = "1.4"
 test-case = "3.3.1"
+
+[workspace.lints.clippy]
+disallowed_types = "deny"
+disallowed-methods = "deny"

benchmarks/cdk/ballista/src/ballista_http.rs

@@ -7,11 +7,11 @@ use ballista::datafusion::physical_plan::execute_stream;
 use ballista::datafusion::prelude::SessionConfig;
 use ballista::datafusion::prelude::SessionContext;
 use ballista::prelude::*;
+use datafusion::common::HashMap;
 use futures::{StreamExt, TryFutureExt};
 use log::{error, info};
 use object_store::aws::AmazonS3Builder;
 use serde::Serialize;
-use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::Display;
 use std::sync::Arc;

benchmarks/cdk/bin/worker.rs

@@ -260,6 +260,7 @@ struct Ec2WorkerResolver {
 }
 
 async fn background_ec2_worker_resolver(urls: Arc<RwLock<Vec<Url>>>) {
+    #[allow(clippy::disallowed_methods)]
     tokio::spawn(async move {
         let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
         let ec2_client = Ec2Client::new(&config);
@@ -317,6 +318,7 @@ async fn background_ec2_worker_resolver(urls: Arc<RwLock<Vec<Url>>>) {
 impl Ec2WorkerResolver {
     fn new() -> Self {
         let urls = Arc::new(RwLock::new(Vec::new()));
+        #[allow(clippy::disallowed_methods)]
         tokio::spawn(background_ec2_worker_resolver(urls.clone()));
         Self { urls }
     }

clippy.toml

@@ -0,0 +1,8 @@
+disallowed-types = [
+  { path = "std::collections::HashMap", reason = "Use hashbrown::HashMap (or datafusion_common::HashMap) for better performance" },
+  { path = "std::collections::HashSet", reason = "Use hashbrown::HashSet (or datafusion_common::HashSet) for better performance" },
+]
+
+disallowed-methods = [
+    { path = "tokio::spawn", reason = "use datafusion::common::runtime::SpawnTask instead to correctly propagate task context" },
+]

console/examples/cluster.rs

@@ -51,6 +51,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     for listener in listeners {
         let resolver = localhost_resolver.clone();
+        #[allow(clippy::disallowed_methods)]
         tokio::spawn(async move {
             let worker = Worker::default();
 

examples/in_memory_cluster.rs

@@ -92,6 +92,7 @@ impl InMemoryChannelResolver {
             async move { Ok(ctx.builder.with_distributed_channel_resolver(this).build()) }
         });
 
+        #[allow(clippy::disallowed_methods)]
         tokio::spawn(async move {
             Server::builder()
                 .add_service(endpoint.into_worker_server())

src/execution_plans/broadcast.rs

@@ -446,6 +446,7 @@ mod tests {
         let mut stream1 = broadcast.execute(0, task_ctx.clone())?;
         assert_eq!(execute_counts[0].load(Ordering::SeqCst), 1);
 
+        #[allow(clippy::disallowed_methods)]
         let handle = tokio::spawn(async move { stream1.next().await });
 
         // Cancel this consumer (simulates a cancellation like a TopK)

src/execution_plans/distributed.rs

@@ -528,6 +528,7 @@ impl<'a> CoordinatorToWorkerTaskSpawner<'a> {
             task_number: task_i as u64,
         };
         let task_metrics_collection = Arc::clone(self.task_metrics);
+        #[allow(clippy::disallowed_methods)]
         tokio::spawn(async move {
             while let Some(Ok(msg)) = worker_to_coordinator_rx.recv().await {
                 let Some(worker_to_coordinator_msg::Inner::TaskMetrics(pre_order_metrics)) =

src/observability/service.rs

@@ -40,6 +40,7 @@ impl ObservabilityServiceImpl {
             let mut sys = sysinfo::System::new_all();
 
             // Spawn background task to periodically collect and send system metrics.
+            #[allow(clippy::disallowed_methods)]
             tokio::task::spawn(async move {
                 loop {
                     sys.refresh_process_specifics(

src/test_utils/in_memory_channel_resolver.rs

@@ -51,6 +51,7 @@ impl InMemoryChannelResolver {
             Ok(builder.with_distributed_channel_resolver(this).build())
         }));
 
+        #[allow(clippy::disallowed_methods)]
         tokio::spawn(async move {
             Server::builder()
                 .add_service(endpoint.into_worker_server())