From 4a227c524f9130eb9b911af17a25c23b25aa41fa Mon Sep 17 00:00:00 2001
From: Piotr Findeisen <piotr.findeisen@gmail.com>
Date: Tue, 3 Sep 2024 09:51:19 +0200
Subject: [PATCH] Remove unsafe Send impl from PriorityMap (#12289)

It's not necessary to use unsafe Send impl. It's enough to require the
referenced trait objects as Send.
---
 .../physical-plan/src/aggregates/topk/hash_table.rs      | 5 ++++-
 datafusion/physical-plan/src/aggregates/topk/heap.rs     | 6 +++++-
 .../physical-plan/src/aggregates/topk/priority_map.rs    | 9 ++-------
 3 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/datafusion/physical-plan/src/aggregates/topk/hash_table.rs b/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
index 2b02fff1f573..232b87de3231 100644
--- a/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
@@ -367,7 +367,10 @@ has_integer!(u8, u16, u32, u64);
 has_integer!(IntervalDayTime, IntervalMonthDayNano);
 hash_float!(f16, f32, f64);
 
-pub fn new_hash_table(limit: usize, kt: DataType) -> Result<Box<dyn ArrowHashTable>> {
+pub fn new_hash_table(
+    limit: usize,
+    kt: DataType,
+) -> Result<Box<dyn ArrowHashTable + Send>> {
     macro_rules! downcast_helper {
         ($kt:ty, $d:ident) => {
             return Ok(Box::new(PrimitiveHashTable::<$kt>::new(limit)))
diff --git a/datafusion/physical-plan/src/aggregates/topk/heap.rs b/datafusion/physical-plan/src/aggregates/topk/heap.rs
index 81eadbc018b3..e694422e443d 100644
--- a/datafusion/physical-plan/src/aggregates/topk/heap.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/heap.rs
@@ -459,7 +459,11 @@ compare_integer!(u8, u16, u32, u64);
 compare_integer!(IntervalDayTime, IntervalMonthDayNano);
 compare_float!(f16, f32, f64);
 
-pub fn new_heap(limit: usize, desc: bool, vt: DataType) -> Result<Box<dyn ArrowHeap>> {
+pub fn new_heap(
+    limit: usize,
+    desc: bool,
+    vt: DataType,
+) -> Result<Box<dyn ArrowHeap + Send>> {
     macro_rules! downcast_helper {
         ($vt:ty, $d:ident) => {
             return Ok(Box::new(PrimitiveHeap::<$vt>::new(limit, desc, vt)))
diff --git a/datafusion/physical-plan/src/aggregates/topk/priority_map.rs b/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
index 668018b9c24c..ed41d22e935b 100644
--- a/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
@@ -25,17 +25,12 @@ use datafusion_common::Result;
 
 /// A `Map<K, V>` / `PriorityQueue` combo that evicts the worst values after reaching `capacity`
 pub struct PriorityMap {
-    map: Box<dyn ArrowHashTable>,
-    heap: Box<dyn ArrowHeap>,
+    map: Box<dyn ArrowHashTable + Send>,
+    heap: Box<dyn ArrowHeap + Send>,
     capacity: usize,
     mapper: Vec<(usize, usize)>,
 }
 
-// JUSTIFICATION
-//  Benefit:  ~15% speedup + required to index into RawTable from binary heap
-//  Soundness: it is only accessed by one thread at a time, and indexes are kept up to date
-unsafe impl Send for PriorityMap {}
-
 impl PriorityMap {
     pub fn new(
         key_type: DataType,