diff --git a/datafusion/physical-plan/src/aggregates/topk/hash_table.rs b/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
index 61410893df865..c20284a5baa6f 100644
--- a/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/hash_table.rs
@@ -30,16 +30,12 @@ use datafusion_common::Result;
 use datafusion_common::exec_datafusion_err;
 use datafusion_common::hash_utils::RandomState;
 use half::f16;
-use hashbrown::hash_table::HashTable;
+use hashbrown::hash_table::{Entry, HashTable};
+use std::borrow::BorrowMut;
 use std::fmt::Debug;
 use std::hash::BuildHasher;
 use std::sync::Arc;
 
-/// A "type alias" for Keys which are stored in our map
-pub trait KeyType: Clone + Comparable + Debug {}
-
-impl<T> KeyType for T where T: Clone + Comparable + Debug {}
-
 /// `heap_idx` assigned to groups whose aggregate values are all NULL. Such
 /// groups are tracked in the hash table only (they never enter the heap), so
 /// they can be emitted with a NULL aggregate value at the end.
@@ -49,9 +45,9 @@ const NULL_HEAP_IDX: usize = usize::MAX;
 /// 1. memoizes the hash
 /// 2. contains the key (ID)
 /// 3. contains the value (heap_idx - an index into the corresponding heap)
-pub struct HashTableItem<ID: KeyType> {
+pub struct HashTableItem<ID> {
     hash: u64,
-    pub id: ID,
+    pub id: Option<ID>,
     pub heap_idx: usize,
 }
 
@@ -59,12 +55,14 @@ pub struct HashTableItem<ID: KeyType> {
 /// 1. limits the number of entries to the top K
 /// 2. Allocates a capacity greater than top K to maintain a low-fill factor and prevent resizing
 /// 3. Tracks indexes to allow corresponding heap to refer to entries by index vs hash
-struct TopKHashTable<ID: KeyType> {
+struct TopKHashTable<ID> {
     map: HashTable<usize>,
     // Store the actual items separately to allow for index-based access
-    store: Vec<Option<HashTableItem<ID>>>,
+    store: Vec<HashTableItem<ID>>,
     // Free indexes in the store for reuse
     free_indices: Vec<usize>,
+    // Pool of reusable value locations, usually Strings
+    free_slots: Vec<ID>,
     // The maximum number of entries allowed
     limit: usize,
     // Number of entries registered as all-NULL (heap_idx == NULL_HEAP_IDX)
@@ -126,7 +124,7 @@ where
     for<'a> &'a S: StringArrayType<'a>,
 {
     owned: S,
-    map: TopKHashTable<Option<String>>,
+    map: TopKHashTable<String>,
     rnd: RandomState,
 }
 
@@ -136,8 +134,9 @@ where
     Option<<VAL as ArrowPrimitiveType>::Native>: Comparable,
 {
     owned: PrimitiveArray<VAL>,
-    map: TopKHashTable<Option<VAL::Native>>,
+    map: TopKHashTable<VAL::Native>,
     rnd: RandomState,
+    value_type: DataType,
 }
 
 impl<S> StringHashTable<S>
@@ -146,9 +145,8 @@ where
     for<'a> &'a S: StringArrayType<'a>,
 {
     pub fn new(limit: usize) -> Self {
-        let owned = S::from(Vec::new());
         Self {
-            owned,
+            owned: S::from(Vec::new()),
             map: TopKHashTable::new(limit, limit * 10),
             rnd: RandomState::default(),
         }
@@ -200,19 +198,14 @@ where
         let hash = self.rnd.hash_one(id);
 
         // Use entry API to avoid double lookup
-        self.map.find_or_insert(
-            hash,
-            id.map(ToOwned::to_owned),
-            replace_idx,
-            Self::eq_fn(id),
-        )
+        self.map
+            .find_or_insert(hash, id, replace_idx, Self::eq_fn(id))
     }
 
     fn insert_null(&mut self, row_idx: usize) -> bool {
         let id = some_value(&self.owned, row_idx);
         let hash = self.rnd.hash_one(id);
-        self.map
-            .insert_null(hash, id.map(ToOwned::to_owned), Self::eq_fn(id))
+        self.map.insert_null(hash, id, Self::eq_fn(id))
     }
 
     fn remove_if_null(&mut self, row_idx: usize) -> bool {
@@ -230,14 +223,12 @@ impl<VAL: ArrowPrimitiveType> PrimitiveHashTable<VAL>
 where
     Option<<VAL as ArrowPrimitiveType>::Native>: Comparable + HashValue,
 {
-    pub fn new(limit: usize, kt: DataType) -> Self {
-        let owned = PrimitiveArray::<VAL>::builder(0)
-            .with_data_type(kt)
-            .finish();
+    pub fn new(limit: usize, value_type: DataType) -> Self {
         Self {
-            owned,
+            owned: PrimitiveArray::<VAL>::new_null(0).with_data_type(value_type.clone()),
             map: TopKHashTable::new(limit, limit * 10),
             rnd: RandomState::default(),
+            value_type,
         }
     }
 
@@ -277,13 +268,10 @@ where
 
     fn take_all(&mut self, indexes: Vec<usize>) -> ArrayRef {
         let ids = self.map.take_all(indexes);
-        let mut builder: PrimitiveBuilder<VAL> = PrimitiveArray::builder(ids.len())
-            .with_data_type(self.owned.data_type().clone());
+        let mut builder: PrimitiveBuilder<VAL> =
+            PrimitiveArray::builder(ids.len()).with_data_type(self.value_type.clone());
         for id in ids.into_iter() {
-            match id {
-                None => builder.append_null(),
-                Some(id) => builder.append_value(id),
-            }
+            builder.append_option(id);
         }
         let ids = builder.finish();
         Arc::new(ids)
@@ -297,12 +285,12 @@ where
         let (id, hash) = self.id_and_hash(row_idx);
         // Use entry API to avoid double lookup
         self.map
-            .find_or_insert(hash, id, replace_idx, Self::eq_fn(id))
+            .find_or_insert(hash, id.as_ref(), replace_idx, Self::eq_fn(id))
     }
 
     fn insert_null(&mut self, row_idx: usize) -> bool {
         let (id, hash) = self.id_and_hash(row_idx);
-        self.map.insert_null(hash, id, Self::eq_fn(id))
+        self.map.insert_null(hash, id.as_ref(), Self::eq_fn(id))
     }
 
     fn remove_if_null(&mut self, row_idx: usize) -> bool {
@@ -315,34 +303,39 @@ where
     }
 }
 
-use hashbrown::hash_table::Entry;
-impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
+impl<ID: PartialEq> TopKHashTable<ID> {
     pub fn new(limit: usize, capacity: usize) -> Self {
         Self {
             map: HashTable::with_capacity(capacity),
             store: Vec::with_capacity(capacity),
             free_indices: Vec::new(),
+            free_slots: Vec::new(),
             limit,
             null_count: 0,
         }
     }
 
     pub fn heap_idx_at(&self, map_idx: usize) -> usize {
-        self.store[map_idx].as_ref().unwrap().heap_idx
+        self.store[map_idx].heap_idx
     }
 
     /// Remove the entry stored at `map_idx`, freeing its store slot for reuse
     fn remove_at(&mut self, map_idx: usize) {
-        let item_to_remove = self.store[map_idx].as_ref().unwrap();
+        let item_to_remove = &self.store[map_idx];
         let hash = item_to_remove.hash;
         let id_to_remove = &item_to_remove.id;
 
-        let eq = |&idx: &usize| self.store[idx].as_ref().unwrap().id == *id_to_remove;
-        let hasher = |idx: &usize| self.store[*idx].as_ref().unwrap().hash;
+        let eq = |&idx: &usize| self.store[idx].id == *id_to_remove;
+        let hasher = |idx: &usize| self.store[*idx].hash;
         match self.map.entry(hash, eq, hasher) {
             Entry::Occupied(entry) => {
                 let (removed_idx, _) = entry.remove();
-                self.store[removed_idx] = None;
+                match self.store[removed_idx].id.take() {
+                    Some(slot) if Self::use_free_slots() => {
+                        self.free_slots.push(slot);
+                    }
+                    _ => (),
+                }
                 self.free_indices.push(removed_idx);
             }
             Entry::Vacant(_) => unreachable!(),
@@ -363,63 +356,91 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
 
     fn update_heap_idx(&mut self, mapper: &[(usize, usize)]) {
         for (m, h) in mapper {
-            self.store[*m].as_mut().unwrap().heap_idx = *h;
+            self.store[*m].heap_idx = *h;
         }
     }
 
+    /// Used to avoid pushing pointless copies of primitives to the `free_slots` pool.
+    const fn use_free_slots() -> bool {
+        std::mem::needs_drop::<ID>()
+    }
+
     /// Find an existing entry or insert a new one, avoiding double hash table lookup.
     /// Returns (map_idx, kind) where kind describes whether the group already
     /// existed, was newly inserted, or was converted from an all-NULL group.
     /// If inserting a new entry and the table is full, replaces the entry at replace_idx.
-    pub fn find_or_insert(
+    pub fn find_or_insert<Q>(
         &mut self,
         hash: u64,
-        id: ID,
+        id: Option<&Q>,
         replace_idx: usize,
-        mut eq: impl FnMut(&ID) -> bool,
-    ) -> (usize, InsertKind) {
+        mut eq: impl FnMut(&Option<ID>) -> bool,
+    ) -> (usize, InsertKind)
+    where
+        Q: ToOwned<Owned = ID> + ?Sized,
+    {
         // Check if entry exists - this is the only hash table lookup
         let mut replaced_null = false;
-        {
-            let eq_fn = |idx: &usize| eq(&self.store[*idx].as_ref().unwrap().id);
-            if let Some(&map_idx) = self.map.find(hash, eq_fn) {
-                if self.store[map_idx].as_ref().unwrap().heap_idx == NULL_HEAP_IDX {
-                    // This group was registered as all-NULL but now produced a
-                    // value: unregister it so it is inserted as a valued group
-                    self.remove_at(map_idx);
-                    self.null_count -= 1;
-                    replaced_null = true;
-                } else {
-                    return (map_idx, InsertKind::Existing);
-                }
+
+        let eq_fn = |idx: &usize| eq(&self.store[*idx].id);
+        if let Some(&map_idx) = self.map.find(hash, eq_fn) {
+            if self.store[map_idx].is_null() {
+                // This group was registered as all-NULL but now produced a
+                // value: unregister it so it is inserted as a valued group
+                self.remove_at(map_idx);
+                self.null_count -= 1;
+                replaced_null = true;
+            } else {
+                return (map_idx, InsertKind::Existing);
             }
         }
 
         // Entry doesn't exist - compute heap_idx and prepare item
         let heap_idx = self.remove_if_full(replace_idx);
+        let store_idx = self.push_store_item(hash, id, heap_idx);
+        let kind = if replaced_null {
+            InsertKind::ReplacedNull
+        } else {
+            InsertKind::New
+        };
+        (store_idx, kind)
+    }
+
+    fn push_store_item<Q>(&mut self, hash: u64, id: Option<&Q>, heap_idx: usize) -> usize
+    where
+        Q: ToOwned<Owned = ID> + ?Sized,
+    {
+        let id = if Self::use_free_slots() {
+            id.map(|id| match self.free_slots.pop() {
+                Some(mut slot) => {
+                    id.clone_into(&mut slot);
+                    slot
+                }
+                _ => id.to_owned(),
+            })
+        } else {
+            debug_assert!(self.free_slots.is_empty(), "primitives should not pool");
+            id.map(ToOwned::to_owned)
+        };
         let mi = HashTableItem::new(hash, id, heap_idx);
         let store_idx = if let Some(idx) = self.free_indices.pop() {
-            self.store[idx] = Some(mi);
+            debug_assert!(self.store[idx].id.is_none(), "slot should be empty");
+            self.store[idx] = mi;
             idx
         } else {
-            self.store.push(Some(mi));
+            self.store.push(mi);
             self.store.len() - 1
         };
 
         // Reserve space if needed
-        let hasher = |idx: &usize| self.store[*idx].as_ref().unwrap().hash;
+        let hasher = |idx: &usize| self.store[*idx].hash;
         if self.map.len() == self.map.capacity() {
             self.map.reserve(self.limit, hasher);
         }
 
         // Insert without checking again since we already confirmed it doesn't exist
         self.map.insert_unique(hash, store_idx, hasher);
-        let kind = if replaced_null {
-            InsertKind::ReplacedNull
-        } else {
-            InsertKind::New
-        };
-        (store_idx, kind)
+        store_idx
     }
 
     /// Register a group whose aggregate values are all NULL, unless it is
@@ -427,13 +448,17 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
     /// never enter the heap. At most `limit` NULL groups are tracked: they all
     /// tie on the sort key, so any `limit` of them is a valid top-k superset.
     /// Returns true if the group was newly registered.
-    pub fn insert_null(
+    pub fn insert_null<Q>(
         &mut self,
         hash: u64,
-        id: ID,
-        mut eq: impl FnMut(&ID) -> bool,
-    ) -> bool {
-        let eq_fn = |idx: &usize| eq(&self.store[*idx].as_ref().unwrap().id);
+        id: Option<&Q>,
+        mut eq: impl FnMut(&Option<ID>) -> bool,
+    ) -> bool
+    where
+        Q: ToOwned<Owned = ID> + ?Sized,
+        ID: BorrowMut<Q::Owned>,
+    {
+        let eq_fn = |idx: &usize| eq(&self.store[*idx].id);
         if self.map.find(hash, eq_fn).is_some() {
             return false;
         }
@@ -442,20 +467,7 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
             return false;
         }
 
-        let mi = HashTableItem::new(hash, id, NULL_HEAP_IDX);
-        let store_idx = if let Some(idx) = self.free_indices.pop() {
-            self.store[idx] = Some(mi);
-            idx
-        } else {
-            self.store.push(Some(mi));
-            self.store.len() - 1
-        };
-
-        let hasher = |idx: &usize| self.store[*idx].as_ref().unwrap().hash;
-        if self.map.len() == self.map.capacity() {
-            self.map.reserve(self.limit, hasher);
-        }
-        self.map.insert_unique(hash, store_idx, hasher);
+        _ = self.push_store_item(hash, id, NULL_HEAP_IDX);
         self.null_count += 1;
         true
     }
@@ -464,10 +476,14 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
     /// all-NULL group produces a value that loses to the current top-k: the
     /// group can no longer reach the top-k, but it must not be emitted with a
     /// NULL value either. Returns true if a NULL registration was removed.
-    pub fn remove_if_null(&mut self, hash: u64, mut eq: impl FnMut(&ID) -> bool) -> bool {
-        let eq_fn = |idx: &usize| eq(&self.store[*idx].as_ref().unwrap().id);
+    pub fn remove_if_null(
+        &mut self,
+        hash: u64,
+        mut eq: impl FnMut(&Option<ID>) -> bool,
+    ) -> bool {
+        let eq_fn = |idx: &usize| eq(&self.store[*idx].id);
         if let Some(&map_idx) = self.map.find(hash, eq_fn)
-            && self.store[map_idx].as_ref().unwrap().heap_idx == NULL_HEAP_IDX
+            && self.store[map_idx].is_null()
         {
             self.remove_at(map_idx);
             self.null_count -= 1;
@@ -482,9 +498,7 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
             .iter()
             .enumerate()
             .filter_map(|(idx, item)| {
-                item.as_ref()
-                    .filter(|item| item.heap_idx == NULL_HEAP_IDX)
-                    .map(|_| idx)
+                (item.id.is_some() && item.is_null()).then_some(idx)
             })
             .collect()
     }
@@ -493,10 +507,10 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
         self.map.len()
     }
 
-    pub fn take_all(&mut self, idxs: Vec<usize>) -> Vec<ID> {
+    pub fn take_all(&mut self, idxs: Vec<usize>) -> Vec<Option<ID>> {
         let ids = idxs
             .into_iter()
-            .map(|idx| self.store[idx].take().unwrap().id)
+            .map(|idx| self.store[idx].id.take())
             .collect();
         self.map.clear();
         self.store.clear();
@@ -506,10 +520,15 @@ impl<ID: KeyType + PartialEq> TopKHashTable<ID> {
     }
 }
 
-impl<ID: KeyType> HashTableItem<ID> {
-    pub fn new(hash: u64, id: ID, heap_idx: usize) -> Self {
+impl<ID> HashTableItem<ID> {
+    pub fn new(hash: u64, id: Option<ID>, heap_idx: usize) -> Self {
         Self { hash, id, heap_idx }
     }
+
+    #[inline]
+    pub fn is_null(&self) -> bool {
+        self.heap_idx == NULL_HEAP_IDX
+    }
 }
 
 impl HashValue for Option<String> {
@@ -603,14 +622,14 @@ mod tests {
     fn should_resize_properly() -> Result<()> {
         let mut heap_to_map = BTreeMap::<usize, usize>::new();
         // Create TopKHashTable with limit=5 and capacity=3 to force resizing
-        let mut map = TopKHashTable::<Option<String>>::new(5, 3);
+        let mut map = TopKHashTable::<String>::new(5, 3);
 
         // Insert 5 entries, tracking the heap-to-map index mapping
         for (heap_idx, id) in ["1", "2", "3", "4", "5"].iter().enumerate() {
             let value = Some(id.to_string());
             let hash = heap_idx as u64;
             let (map_idx, kind) =
-                map.find_or_insert(hash, value.clone(), heap_idx, |v| *v == value);
+                map.find_or_insert(hash, value.as_ref(), heap_idx, |v| *v == value);
             assert_eq!(kind, InsertKind::New, "Entry should be new");
             heap_to_map.insert(heap_idx, map_idx);
         }
@@ -636,23 +655,23 @@ mod tests {
 
     #[test]
     fn should_track_null_groups() -> Result<()> {
-        let mut map = TopKHashTable::<Option<String>>::new(2, 10);
+        let mut map = TopKHashTable::<String>::new(2, 10);
 
         let a = Some("a".to_string());
         let b = Some("b".to_string());
         let c = Some("c".to_string());
 
         // register two all-NULL groups; the third exceeds the NULL group limit
-        assert!(map.insert_null(100, a.clone(), |v| *v == a));
-        assert!(map.insert_null(200, b.clone(), |v| *v == b));
-        assert!(!map.insert_null(300, c.clone(), |v| *v == c));
+        assert!(map.insert_null(100, a.as_ref(), |v| *v == a));
+        assert!(map.insert_null(200, b.as_ref(), |v| *v == b));
+        assert!(!map.insert_null(300, c.as_ref(), |v| *v == c));
         // re-registering an existing NULL group is a no-op
-        assert!(!map.insert_null(100, a.clone(), |v| *v == a));
+        assert!(!map.insert_null(100, a.as_ref(), |v| *v == a));
         assert_eq!(map.null_count, 2);
         assert_eq!(map.null_map_idxs(), vec![0, 1]);
 
         // a valued insert for a NULL group converts it to a valued group
-        let (map_idx, kind) = map.find_or_insert(200, b.clone(), 0, |v| *v == b);
+        let (map_idx, kind) = map.find_or_insert(200, b.as_ref(), 0, |v| *v == b);
         assert_eq!(kind, InsertKind::ReplacedNull, "NULL group should convert");
         assert_eq!(map.heap_idx_at(map_idx), 0, "Heap should append at 0");
         assert_eq!(map.null_count, 1);
@@ -672,24 +691,24 @@ mod tests {
 
     #[test]
     fn should_reuse_all_freed_store_slots() -> Result<()> {
-        let mut map = TopKHashTable::<Option<String>>::new(1, 10);
+        let mut map = TopKHashTable::<String>::new(1, 10);
 
         let a = Some("a".to_string());
         let b = Some("b".to_string());
         let c = Some("c".to_string());
 
-        let (b_idx, kind) = map.find_or_insert(100, b.clone(), 0, |v| *v == b);
+        let (b_idx, kind) = map.find_or_insert(100, b.as_ref(), 0, |v| *v == b);
         assert_eq!(kind, InsertKind::New);
-        assert!(map.insert_null(200, a.clone(), |v| *v == a));
+        assert!(map.insert_null(200, a.as_ref(), |v| *v == a));
 
         // Converting a NULL group while the valued heap is full frees two
         // slots: the NULL registration and the evicted valued group.
-        let (_, kind) = map.find_or_insert(200, a.clone(), b_idx, |v| *v == a);
+        let (_, kind) = map.find_or_insert(200, a.as_ref(), b_idx, |v| *v == a);
         assert_eq!(kind, InsertKind::ReplacedNull);
 
         // Both freed slots must remain reusable. Otherwise repeated
         // conversions make the backing store grow without bound.
-        assert!(map.insert_null(300, c.clone(), |v| *v == c));
+        assert!(map.insert_null(300, c.as_ref(), |v| *v == c));
         assert_eq!(map.store.len(), 2);
 
         Ok(())
diff --git a/datafusion/physical-plan/src/aggregates/topk/heap.rs b/datafusion/physical-plan/src/aggregates/topk/heap.rs
index aef6bb5596c2e..fa6815371bfcc 100644
--- a/datafusion/physical-plan/src/aggregates/topk/heap.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/heap.rs
@@ -23,12 +23,10 @@
 //! Supported value types include Arrow primitives (integers, floats, decimals, intervals)
 //! and UTF-8 strings (`Utf8`, `LargeUtf8`, `Utf8View`) using lexicographic ordering.
 
-use arrow::array::{ArrayRef, ArrowPrimitiveType, PrimitiveArray, downcast_primitive};
-use arrow::array::{LargeStringBuilder, StringBuilder, StringViewBuilder};
+use arrow::array::types::{IntervalDayTime, IntervalMonthDayNano};
 use arrow::array::{
-    StringArray,
-    cast::AsArray,
-    types::{IntervalDayTime, IntervalMonthDayNano},
+    Array, ArrayAccessor, ArrayRef, ArrowPrimitiveType, AsArray, LargeStringArray,
+    PrimitiveArray, StringArray, StringArrayType, StringViewArray, downcast_primitive,
 };
 use arrow::buffer::ScalarBuffer;
 use arrow::datatypes::{DataType, i256};
@@ -41,8 +39,20 @@ use std::fmt::{Debug, Display, Formatter};
 use std::sync::Arc;
 
 /// A custom version of `Ord` that only exists to we can implement it for the Values in our heap
-pub trait Comparable {
-    fn comp(&self, other: &Self) -> Ordering;
+pub trait Comparable<Rhs: ?Sized = Self> {
+    fn comp(&self, other: &Rhs) -> Ordering;
+}
+
+impl Comparable<String> for str {
+    fn comp(&self, other: &String) -> Ordering {
+        self.cmp(other.as_str())
+    }
+}
+
+impl Comparable for String {
+    fn comp(&self, other: &Self) -> Ordering {
+        self.cmp(other)
+    }
 }
 
 impl Comparable for Option<String> {
@@ -52,15 +62,52 @@ impl Comparable for Option<String> {
 }
 
 /// A "type alias" for Values which are stored in our heap
-pub trait ValueType: Comparable + Clone + Debug {}
+pub trait ValueType: Comparable + Clone + Debug + Default {}
 
-impl<T> ValueType for T where T: Comparable + Clone + Debug {}
+impl<T> ValueType for T where T: Comparable + Clone + Debug + Default {}
+
+const VACANT_MAP_IDX: usize = usize::MAX;
 
 /// An entry in our heap, which contains both the value and a index into an external HashTable
 struct HeapItem<VAL: ValueType> {
     val: VAL,
     map_idx: usize,
 }
+impl<VAL: ValueType> HeapItem<VAL> {
+    fn is_vacant(&self) -> bool {
+        self.map_idx == VACANT_MAP_IDX
+    }
+
+    fn is_occupied(&self) -> bool {
+        self.map_idx != VACANT_MAP_IDX
+    }
+
+    fn map_idx(&self) -> Option<usize> {
+        self.is_occupied().then_some(self.map_idx)
+    }
+
+    fn take(&mut self) -> Option<Self> {
+        self.is_occupied().then(|| std::mem::take(self))
+    }
+
+    fn as_ref(&self) -> &Self {
+        debug_assert!(self.is_occupied());
+        self
+    }
+
+    fn as_mut(&mut self) -> &mut Self {
+        debug_assert!(self.is_occupied());
+        self
+    }
+}
+impl<VAL: ValueType> Default for HeapItem<VAL> {
+    fn default() -> Self {
+        Self {
+            val: VAL::default(),
+            map_idx: VACANT_MAP_IDX,
+        }
+    }
+}
 
 /// A custom heap implementation that allows several things that couldn't be achieved with
 /// `collections::BinaryHeap`:
@@ -72,11 +119,12 @@ struct TopKHeap<VAL: ValueType> {
     desc: bool,
     len: usize,
     capacity: usize,
-    heap: Vec<Option<HeapItem<VAL>>>,
+    heap: Vec<HeapItem<VAL>>,
 }
 
 /// An interface to hide the generic type signature of TopKHeap behind arrow arrays
 pub trait ArrowHeap {
+    fn value_type(&self) -> &DataType;
     fn set_batch(&mut self, vals: ArrayRef);
     fn is_worse(&self, idx: usize) -> bool;
     fn worst_map_idx(&self) -> usize;
@@ -98,20 +146,19 @@ where
     batch: PrimitiveArray<VAL>,
     heap: TopKHeap<VAL::Native>,
     desc: bool,
-    data_type: DataType,
+    value_type: DataType,
 }
 
 impl<VAL: ArrowPrimitiveType> PrimitiveHeap<VAL>
 where
     <VAL as ArrowPrimitiveType>::Native: Comparable,
 {
-    pub fn new(limit: usize, desc: bool, data_type: DataType) -> Self {
-        let batch = PrimitiveArray::<VAL>::builder(0).finish();
+    pub fn new(limit: usize, desc: bool, value_type: DataType) -> Self {
         Self {
-            batch,
+            batch: PrimitiveArray::<VAL>::new_null(0).with_data_type(value_type.clone()),
             heap: TopKHeap::new(limit, desc),
             desc,
-            data_type,
+            value_type,
         }
     }
 }
@@ -120,8 +167,12 @@ impl<VAL: ArrowPrimitiveType> ArrowHeap for PrimitiveHeap<VAL>
 where
     <VAL as ArrowPrimitiveType>::Native: Comparable,
 {
+    fn value_type(&self) -> &DataType {
+        &self.value_type
+    }
+
     fn set_batch(&mut self, vals: ArrayRef) {
-        self.batch = PrimitiveArray::from(vals.to_data());
+        self.batch = vals.as_primitive().clone();
     }
 
     fn is_worse(&self, row_idx: usize) -> bool {
@@ -139,7 +190,7 @@ where
 
     fn insert(&mut self, row_idx: usize, map_idx: usize, map: &mut Vec<(usize, usize)>) {
         let new_val = self.batch.value(row_idx);
-        self.heap.append_or_replace(new_val, map_idx, map);
+        self.heap.append_or_replace(&new_val, map_idx, map);
     }
 
     fn replace_if_better(
@@ -149,14 +200,14 @@ where
         map: &mut Vec<(usize, usize)>,
     ) {
         let new_val = self.batch.value(row_idx);
-        self.heap.replace_if_better(heap_idx, new_val, map);
+        self.heap.replace_if_better(&new_val, heap_idx, map);
     }
 
     fn drain(&mut self) -> (ArrayRef, Vec<usize>) {
         let nulls = None;
         let (vals, map_idxs) = self.heap.drain();
         let arr = PrimitiveArray::<VAL>::new(ScalarBuffer::from(vals), nulls)
-            .with_data_type(self.data_type.clone());
+            .with_data_type(self.value_type.clone());
         (Arc::new(arr), map_idxs)
     }
 }
@@ -168,58 +219,46 @@ where
 /// borrowed strings are compared before allocation, and only allocated when the
 /// heap confirms they improve the top-K set.
 ///
-pub struct StringHeap {
-    batch: ArrayRef,
-    heap: TopKHeap<Option<String>>,
+pub struct StringHeap<S: Array>
+where
+    for<'a> &'a S: StringArrayType<'a>,
+{
+    batch: S,
+    heap: TopKHeap<String>,
     desc: bool,
-    data_type: DataType,
 }
 
-impl StringHeap {
-    pub fn new(limit: usize, desc: bool, data_type: DataType) -> Self {
-        let batch: ArrayRef = Arc::new(StringArray::from(Vec::<&str>::new()));
+impl<S> StringHeap<S>
+where
+    S: Array + From<Vec<Option<String>>>,
+    for<'a> &'a S: StringArrayType<'a>,
+{
+    pub fn new(limit: usize, desc: bool) -> Self {
+        let batch = S::from(Vec::new());
         Self {
             batch,
             heap: TopKHeap::new(limit, desc),
             desc,
-            data_type,
         }
     }
-
-    /// Extracts a string value from the current batch at the given row index.
-    ///
-    /// Panics if the row index is out of bounds or if the data type is not one of
-    /// the supported UTF-8 string types.
-    ///
-    /// Note: Null values should not appear in the input; the aggregation layer
-    /// ensures nulls are filtered before reaching this code.
-    fn value(&self, row_idx: usize) -> &str {
-        extract_string_value(&self.batch, &self.data_type, row_idx)
-    }
 }
 
-/// Helper to extract a string value from an ArrayRef at a given index.
-///
-/// Supports `Utf8`, `LargeUtf8`, and `Utf8View` data types.
-///
-/// # Panics
-/// Panics if the index is out of bounds or if the data type is unsupported.
-fn extract_string_value<'a>(
-    batch: &'a ArrayRef,
-    data_type: &DataType,
-    idx: usize,
-) -> &'a str {
-    match data_type {
-        DataType::Utf8 => batch.as_string::<i32>().value(idx),
-        DataType::LargeUtf8 => batch.as_string::<i64>().value(idx),
-        DataType::Utf8View => batch.as_string_view().value(idx),
-        _ => unreachable!("Unsupported string type: {data_type}"),
+impl<S> ArrowHeap for StringHeap<S>
+where
+    S: Array + Clone + From<Vec<String>> + 'static,
+    for<'a> &'a S: StringArrayType<'a>,
+{
+    fn value_type(&self) -> &DataType {
+        // Strings don't store any metadata.
+        self.batch.data_type()
     }
-}
 
-impl ArrowHeap for StringHeap {
     fn set_batch(&mut self, vals: ArrayRef) {
-        self.batch = vals;
+        self.batch = vals
+            .as_any()
+            .downcast_ref::<S>()
+            .expect("Unsupported data type")
+            .clone();
     }
 
     fn is_worse(&self, row_idx: usize) -> bool {
@@ -229,15 +268,9 @@ impl ArrowHeap for StringHeap {
         // Compare borrowed `&str` against the worst heap value first to avoid
         // allocating a `String` unless this row would actually replace an
         // existing heap entry.
-        let new_val = self.value(row_idx);
-        let worst_val = self.heap.worst_val().expect("Missing root");
-        match worst_val {
-            None => false,
-            Some(worst_str) => {
-                (!self.desc && new_val > worst_str.as_str())
-                    || (self.desc && new_val < worst_str.as_str())
-            }
-        }
+        let new_val = (&self.batch).value(row_idx);
+        let worst_val = self.heap.worst_val().expect("Missing root").as_str();
+        (!self.desc && new_val > worst_val) || (self.desc && new_val < worst_val)
     }
 
     fn worst_map_idx(&self) -> usize {
@@ -245,12 +278,7 @@ impl ArrowHeap for StringHeap {
     }
 
     fn insert(&mut self, row_idx: usize, map_idx: usize, map: &mut Vec<(usize, usize)>) {
-        // When appending (heap not full) we must allocate to own the string
-        // because it will be stored in the heap. For replacements we avoid
-        // allocation until `replace_if_better` confirms a replacement is
-        // necessary.
-        let new_str = self.value(row_idx).to_string();
-        let new_val = Some(new_str);
+        let new_val = (&self.batch).value(row_idx);
         self.heap.append_or_replace(new_val, map_idx, map);
     }
 
@@ -260,62 +288,14 @@ impl ArrowHeap for StringHeap {
         row_idx: usize,
         map: &mut Vec<(usize, usize)>,
     ) {
-        let new_str = self.value(row_idx);
-        let existing = self.heap.heap[heap_idx]
-            .as_ref()
-            .expect("Missing heap item");
-
-        // Compare borrowed reference first—no allocation yet.
-        // We compare the borrowed `&str` with the stored `Option<String>` and
-        // only allocate (`to_string()`) when a replacement is required.
-        match &existing.val {
-            None => {
-                // Existing is null; new value always wins
-                let new_val = Some(new_str.to_string());
-                self.heap.replace_if_better(heap_idx, new_val, map);
-            }
-            Some(existing_str) => {
-                // Compare borrowed strings first
-                if (!self.desc && new_str < existing_str.as_str())
-                    || (self.desc && new_str > existing_str.as_str())
-                {
-                    let new_val = Some(new_str.to_string());
-                    self.heap.replace_if_better(heap_idx, new_val, map);
-                }
-                // Else: no improvement, no allocation
-            }
-        }
+        let new_val = (&self.batch).value(row_idx);
+        self.heap.replace_if_better(new_val, heap_idx, map);
     }
 
     fn drain(&mut self) -> (ArrayRef, Vec<usize>) {
         let (vals, map_idxs) = self.heap.drain();
-        // Use Arrow builders to safely construct arrays from the owned
-        // `Option<String>` values. Builders avoid needing to maintain
-        // references to temporary storage.
-
-        // Macro to eliminate duplication across string builder types.
-        // All three builders share the same interface for append_value,
-        // append_null, and finish, differing only in their concrete types.
-        macro_rules! build_string_array {
-            ($builder_type:ty) => {{
-                let mut builder = <$builder_type>::new();
-                for val in vals {
-                    match val {
-                        Some(s) => builder.append_value(&s),
-                        None => builder.append_null(),
-                    }
-                }
-                Arc::new(builder.finish())
-            }};
-        }
-
-        let arr: ArrayRef = match self.data_type {
-            DataType::Utf8 => build_string_array!(StringBuilder),
-            DataType::LargeUtf8 => build_string_array!(LargeStringBuilder),
-            DataType::Utf8View => build_string_array!(StringViewBuilder),
-            _ => unreachable!("Unsupported string type: {}", self.data_type),
-        };
-        (arr, map_idxs)
+        let vals = Arc::new(S::from(vals));
+        (vals, map_idxs)
     }
 }
 
@@ -325,18 +305,17 @@ impl<VAL: ValueType> TopKHeap<VAL> {
             desc,
             capacity: limit,
             len: 0,
-            heap: (0..=limit).map(|_| None).collect::<Vec<_>>(),
+            heap: (0..=limit).map(|_| HeapItem::default()).collect::<Vec<_>>(),
         }
     }
 
     pub fn worst_val(&self) -> Option<&VAL> {
         let root = self.heap.first()?;
-        let hi = root.as_ref()?;
-        Some(&hi.val)
+        root.is_occupied().then_some(&root.val)
     }
 
     pub fn worst_map_idx(&self) -> usize {
-        self.heap[0].as_ref().map(|hi| hi.map_idx).unwrap_or(0)
+        self.heap[0].map_idx().unwrap_or(0)
     }
 
     pub fn is_full(&self) -> bool {
@@ -347,12 +326,14 @@ impl<VAL: ValueType> TopKHeap<VAL> {
         self.len
     }
 
-    pub fn append_or_replace(
+    pub fn append_or_replace<Q>(
         &mut self,
-        new_val: VAL,
+        new_val: &Q,
         map_idx: usize,
         map: &mut Vec<(usize, usize)>,
-    ) {
+    ) where
+        Q: ToOwned<Owned = VAL> + ?Sized,
+    {
         if self.is_full() {
             self.replace_root(new_val, map_idx, map);
         } else {
@@ -360,9 +341,14 @@ impl<VAL: ValueType> TopKHeap<VAL> {
         }
     }
 
-    fn append(&mut self, new_val: VAL, map_idx: usize, mapper: &mut Vec<(usize, usize)>) {
-        let hi = HeapItem::new(new_val, map_idx);
-        self.heap[self.len] = Some(hi);
+    fn append<Q>(&mut self, new_val: &Q, map_idx: usize, mapper: &mut Vec<(usize, usize)>)
+    where
+        Q: ToOwned<Owned = VAL> + ?Sized,
+    {
+        let hi = &mut self.heap[self.len];
+        debug_assert!(hi.is_vacant());
+        hi.map_idx = map_idx;
+        new_val.clone_into(&mut hi.val);
         self.heapify_up(self.len, mapper);
         self.len += 1;
     }
@@ -383,43 +369,48 @@ impl<VAL: ValueType> TopKHeap<VAL> {
     }
 
     pub fn drain(&mut self) -> (Vec<VAL>, Vec<usize>) {
-        let mut map = Vec::with_capacity(self.len);
+        let mut map = Vec::new();
         let mut vals = Vec::with_capacity(self.len);
         let mut map_idxs = Vec::with_capacity(self.len);
         while let Some(worst_hi) = self.pop(&mut map) {
             vals.push(worst_hi.val);
             map_idxs.push(worst_hi.map_idx);
+            map.clear();
         }
         vals.reverse();
         map_idxs.reverse();
         (vals, map_idxs)
     }
 
-    fn replace_root(
+    fn replace_root<Q>(
         &mut self,
-        new_val: VAL,
+        new_val: &Q,
         map_idx: usize,
         mapper: &mut Vec<(usize, usize)>,
-    ) {
-        let hi = self.heap[0].as_mut().expect("No root");
-        hi.val = new_val;
+    ) where
+        Q: ToOwned<Owned = VAL> + ?Sized,
+    {
+        let hi = self.heap[0].as_mut();
+        new_val.clone_into(&mut hi.val);
         hi.map_idx = map_idx;
         self.heapify_down(0, mapper);
     }
 
-    pub fn replace_if_better(
+    pub fn replace_if_better<Q>(
         &mut self,
+        new_val: &Q,
         heap_idx: usize,
-        new_val: VAL,
         mapper: &mut Vec<(usize, usize)>,
-    ) {
-        let existing = self.heap[heap_idx].as_mut().expect("Missing heap item");
+    ) where
+        Q: ToOwned<Owned = VAL> + Comparable<VAL> + ?Sized,
+    {
+        let existing = self.heap[heap_idx].as_mut();
         if (!self.desc && new_val.comp(&existing.val) != Ordering::Less)
             || (self.desc && new_val.comp(&existing.val) != Ordering::Greater)
         {
             return;
         }
-        existing.val = new_val;
+        new_val.clone_into(&mut existing.val);
         self.heapify_down(heap_idx, mapper);
     }
 
@@ -427,8 +418,8 @@ impl<VAL: ValueType> TopKHeap<VAL> {
         let desc = self.desc;
         while idx != 0 {
             let parent_idx = (idx - 1) / 2;
-            let node = self.heap[idx].as_ref().expect("No heap item");
-            let parent = self.heap[parent_idx].as_ref().expect("No heap item");
+            let node = self.heap[idx].as_ref();
+            let parent = self.heap[parent_idx].as_ref();
             if (!desc && node.val.comp(&parent.val) != Ordering::Greater)
                 || (desc && node.val.comp(&parent.val) != Ordering::Less)
             {
@@ -439,40 +430,45 @@ impl<VAL: ValueType> TopKHeap<VAL> {
         }
     }
 
+    #[inline]
     fn swap(&mut self, a_idx: usize, b_idx: usize, mapper: &mut Vec<(usize, usize)>) {
         self.heap.swap(a_idx, b_idx);
 
-        let b_hi = self.heap[b_idx].as_ref().expect("Missing heap entry");
-        let a_hi = self.heap[a_idx].as_ref().expect("Missing heap entry");
+        let b_hi = self.heap[b_idx].as_ref();
+        let a_hi = self.heap[a_idx].as_ref();
 
-        mapper.push((b_hi.map_idx, b_idx));
-        mapper.push((a_hi.map_idx, a_idx));
+        mapper.extend([(b_hi.map_idx, b_idx), (a_hi.map_idx, a_idx)]);
     }
 
-    fn heapify_down(&mut self, node_idx: usize, mapper: &mut Vec<(usize, usize)>) {
-        let left_child = node_idx * 2 + 1;
+    fn heapify_down(&mut self, mut node_idx: usize, mapper: &mut Vec<(usize, usize)>) {
         let desc = self.desc;
-        let entry = self.heap.get(node_idx).expect("Missing node!");
-        let entry = entry.as_ref().expect("Missing node!");
-        let mut best_idx = node_idx;
-        let mut best_val = &entry.val;
-        for child_idx in left_child..=left_child + 1 {
-            if let Some(Some(child)) = self.heap.get(child_idx)
-                && ((!desc && child.val.comp(best_val) == Ordering::Greater)
-                    || (desc && child.val.comp(best_val) == Ordering::Less))
-            {
-                best_val = &child.val;
-                best_idx = child_idx;
+        loop {
+            let left_child = node_idx * 2 + 1;
+            let entry = self.heap[node_idx].as_ref();
+            let mut best_idx = node_idx;
+            let mut best_val = &entry.val;
+            for child_idx in left_child..=left_child + 1 {
+                if let Some(child) = self.heap.get(child_idx)
+                    && child.is_occupied()
+                    && ((!desc && child.val.comp(best_val) == Ordering::Greater)
+                        || (desc && child.val.comp(best_val) == Ordering::Less))
+                {
+                    best_val = &child.val;
+                    best_idx = child_idx;
+                }
+            }
+            if best_val.comp(&entry.val) == Ordering::Equal {
+                break;
             }
-        }
-        if best_val.comp(&entry.val) != Ordering::Equal {
             self.swap(best_idx, node_idx, mapper);
-            self.heapify_down(best_idx, mapper);
+            node_idx = best_idx;
         }
     }
 
     fn _tree_print(&self, idx: usize, prefix: &str, is_tail: bool, output: &mut String) {
-        if let Some(Some(hi)) = self.heap.get(idx) {
+        if let Some(hi) = self.heap.get(idx)
+            && hi.is_occupied()
+        {
             let connector = if idx != 0 {
                 if is_tail { "└── " } else { "├── " }
             } else {
@@ -511,15 +507,6 @@ impl<VAL: ValueType> Display for TopKHeap<VAL> {
     }
 }
 
-impl<VAL: ValueType> HeapItem<VAL> {
-    pub fn new(val: VAL, buk_idx: usize) -> Self {
-        Self {
-            val,
-            map_idx: buk_idx,
-        }
-    }
-}
-
 impl<VAL: ValueType> Debug for HeapItem<VAL> {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         f.write_str("bucket=")?;
@@ -615,13 +602,6 @@ pub fn new_heap(
     desc: bool,
     vt: DataType,
 ) -> Result<Box<dyn ArrowHeap + Send>> {
-    if matches!(
-        vt,
-        DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
-    ) {
-        return Ok(Box::new(StringHeap::new(limit, desc, vt)));
-    }
-
     macro_rules! downcast_helper {
         ($vt:ty, $d:ident) => {
             return Ok(Box::new(PrimitiveHeap::<$vt>::new(limit, desc, vt)))
@@ -630,6 +610,9 @@ pub fn new_heap(
 
     downcast_primitive! {
         vt => (downcast_helper, vt),
+        DataType::Utf8 => return Ok(Box::new(StringHeap::<StringArray>::new(limit, desc))),
+        DataType::LargeUtf8 => return Ok(Box::new(StringHeap::<LargeStringArray>::new(limit, desc))),
+        DataType::Utf8View => return Ok(Box::new(StringHeap::<StringViewArray>::new(limit, desc))),
         _ => {}
     }
 
@@ -648,7 +631,7 @@ mod tests {
     fn should_append() -> Result<()> {
         let mut map = vec![];
         let mut heap = TopKHeap::new(10, false);
-        heap.append_or_replace(1, 1, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
 
         let actual = heap.to_string();
         assert_snapshot!(actual, @"val=1 idx=0, bucket=1");
@@ -661,10 +644,10 @@ mod tests {
         let mut map = vec![];
         let mut heap = TopKHeap::new(10, false);
 
-        heap.append_or_replace(1, 1, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
         assert_eq!(map, vec![]);
 
-        heap.append_or_replace(2, 2, &mut map);
+        heap.append_or_replace(&2, 2, &mut map);
         assert_eq!(map, vec![(2, 0), (1, 1)]);
 
         let actual = heap.to_string();
@@ -681,9 +664,9 @@ mod tests {
         let mut map = vec![];
         let mut heap = TopKHeap::new(3, false);
 
-        heap.append_or_replace(1, 1, &mut map);
-        heap.append_or_replace(2, 2, &mut map);
-        heap.append_or_replace(3, 3, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
+        heap.append_or_replace(&2, 2, &mut map);
+        heap.append_or_replace(&3, 3, &mut map);
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
         val=3 idx=0, bucket=3
@@ -692,7 +675,7 @@ mod tests {
         ");
 
         let mut map = vec![];
-        heap.append_or_replace(0, 0, &mut map);
+        heap.append_or_replace(&0, 0, &mut map);
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
         val=2 idx=0, bucket=2
@@ -709,10 +692,10 @@ mod tests {
         let mut map = vec![];
         let mut heap = TopKHeap::new(4, false);
 
-        heap.append_or_replace(1, 1, &mut map);
-        heap.append_or_replace(2, 2, &mut map);
-        heap.append_or_replace(3, 3, &mut map);
-        heap.append_or_replace(4, 4, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
+        heap.append_or_replace(&2, 2, &mut map);
+        heap.append_or_replace(&3, 3, &mut map);
+        heap.append_or_replace(&4, 4, &mut map);
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
         val=4 idx=0, bucket=4
@@ -722,7 +705,7 @@ mod tests {
         ");
 
         let mut map = vec![];
-        heap.replace_if_better(1, 0, &mut map);
+        heap.replace_if_better(&0, 1, &mut map);
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
         val=4 idx=0, bucket=4
@@ -740,8 +723,8 @@ mod tests {
         let mut map = vec![];
         let mut heap = TopKHeap::new(10, false);
 
-        heap.append_or_replace(1, 1, &mut map);
-        heap.append_or_replace(2, 2, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
+        heap.append_or_replace(&2, 2, &mut map);
 
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
@@ -760,8 +743,8 @@ mod tests {
         let mut map = vec![];
         let mut heap = TopKHeap::new(10, false);
 
-        heap.append_or_replace(1, 1, &mut map);
-        heap.append_or_replace(2, 2, &mut map);
+        heap.append_or_replace(&1, 1, &mut map);
+        heap.append_or_replace(&2, 2, &mut map);
 
         let actual = heap.to_string();
         assert_snapshot!(actual, @r"
@@ -776,4 +759,6 @@ mod tests {
 
         Ok(())
     }
+
+    // TODO: test TopKHeap of String?
 }
diff --git a/datafusion/physical-plan/src/aggregates/topk/priority_map.rs b/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
index f46cb22a7a63c..33b27767b29cc 100644
--- a/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/priority_map.rs
@@ -30,7 +30,6 @@ pub struct PriorityMap {
     heap: Box<dyn ArrowHeap + Send>,
     capacity: usize,
     mapper: Vec<(usize, usize)>,
-    val_type: DataType,
     /// Mirror of the map's all-NULL group count, kept as a plain field so the
     /// per-row `insert` path can check it without a `dyn` call (measured to
     /// regress the topk_aggregate benchmarks when read through the trait)
@@ -46,10 +45,9 @@ impl PriorityMap {
     ) -> Result<Self> {
         Ok(Self {
             map: new_hash_table(capacity, key_type)?,
-            heap: new_heap(capacity, descending, val_type.clone())?,
+            heap: new_heap(capacity, descending, val_type)?,
             capacity,
             mapper: Vec::with_capacity(capacity),
-            val_type,
             null_count: 0,
         })
     }
@@ -60,13 +58,14 @@ impl PriorityMap {
     }
 
     pub fn insert(&mut self, row_idx: usize) -> Result<()> {
-        assert!(self.map.len() <= self.capacity, "Overflow");
         debug_assert_eq!(self.null_count, 0);
 
         // if we're full, and the new val is worse than all our values, just bail
         if self.heap.is_worse(row_idx) {
             return Ok(());
         }
+        assert!(self.map.len() <= self.capacity, "Overflow");
+
         self.insert_eligible(row_idx)
     }
 
@@ -74,10 +73,6 @@ impl PriorityMap {
     /// separate from [`Self::insert`] so the common no-NULL path does not pay
     /// for NULL bookkeeping on every row.
     pub fn insert_with_null_groups(&mut self, row_idx: usize) -> Result<()> {
-        // valued groups are capped at `capacity`; up to `capacity` additional
-        // all-NULL groups may be tracked alongside them
-        assert!(self.map.len() <= 2 * self.capacity, "Overflow");
-
         if self.heap.is_worse(row_idx) {
             // A group that was registered as all-NULL now has a value that
             // loses to the current top-k: it can no longer reach the top-k,
@@ -87,9 +82,19 @@ impl PriorityMap {
             }
             return Ok(());
         }
+        self.null_capacity_assert();
+
         self.insert_eligible(row_idx)
     }
 
+    #[inline]
+    fn null_capacity_assert(&self) {
+        // TODO: do debug_assert instead?
+        // valued groups are capped at `capacity`; up to `capacity` additional
+        // all-NULL groups may be tracked alongside them
+        assert!(self.map.len() <= 2 * self.capacity, "Overflow");
+    }
+
     fn insert_eligible(&mut self, row_idx: usize) -> Result<()> {
         let map = &mut self.mapper;
 
@@ -125,8 +130,8 @@ impl PriorityMap {
     /// must still appear in the aggregation output; such groups all tie on the
     /// sort key, so tracking up to `capacity` of them preserves top-k semantics.
     pub fn insert_null(&mut self, row_idx: usize) {
-        assert!(self.map.len() <= 2 * self.capacity, "Overflow");
         if self.map.insert_null(row_idx) {
+            self.null_capacity_assert();
             self.null_count += 1;
         }
     }
@@ -140,7 +145,7 @@ impl PriorityMap {
             vals
         } else {
             map_idxs.extend(null_idxs.iter().copied());
-            let nulls = new_null_array(&self.val_type, null_idxs.len());
+            let nulls = new_null_array(self.heap.value_type(), null_idxs.len());
             concat(&[vals.as_ref(), nulls.as_ref()])?
         };
         let ids = self.map.take_all(map_idxs);