diff --git a/datafusion/physical-expr/src/aggregate/median.rs b/datafusion/physical-expr/src/aggregate/median.rs
index 1049187a5..0e9b0b87d 100644
--- a/datafusion/physical-expr/src/aggregate/median.rs
+++ b/datafusion/physical-expr/src/aggregate/median.rs
@@ -23,7 +23,7 @@ use crate::{AggregateExpr, PhysicalExpr};
 use arrow::array::{Array, ArrayRef};
 use arrow::datatypes::{DataType, Field};
 use arrow_array::cast::AsArray;
-use arrow_array::{downcast_integer, ArrowNativeTypeOp, ArrowNumericType};
+use arrow_array::{downcast_integer, ArrowNativeTypeOp, ArrowNumericType, ListArray};
 use arrow_buffer::ArrowNativeType;
 use datafusion_common::{DataFusionError, Result, ScalarValue};
 use datafusion_expr::Accumulator;
@@ -71,7 +71,7 @@ impl AggregateExpr for Median {
         use arrow_array::types::*;
         macro_rules! helper {
             ($t:ty, $dt:expr) => {
-                Ok(Box::new(MedianAccumulator::<$t> {
+                Ok(Box::new(MedianAccumulator::<$t, Vec<<$t as ArrowPrimitiveType>::Native>> {
                     data_type: $dt.clone(),
                     all_values: vec![],
                 }))
@@ -127,6 +127,39 @@ impl PartialEq<dyn Any> for Median {
     }
 }

+/// A trait for a container of numeric types that can be compared
+/// A `Vec` is used for Median and `HashSet` for DistinctMedian
+trait MedianValues: Send + Sync + std::fmt::Debug {
+    type T: ArrowNativeType;
+
+    fn reserve(&mut self, additional: usize);
+    fn extend(&mut self, values: impl Iterator<Item = Self::T>);
+    fn into_iter(self) -> Box<dyn Iterator<Item = Self::T>>;
+    /// Convert the elements of this container into a ListArray
+    fn into_list_array(self) -> ListArray;
+}
+
+impl <T:ArrowNativeType> MedianValues for Vec<T> {
+    type T = T;
+
+    fn reserve(&mut self, additional: usize) {
+        todo!()
+    }
+
+    fn extend(&mut self, values: impl Iterator<Item=Self::T>) {
+        todo!()
+    }
+
+    fn into_iter(self) -> Box<dyn Iterator<Item=Self::T>> {
+        todo!()
+    }
+
+    fn into_list_array(self) -> ListArray {
+        todo!()
+    }
+}
+
+
 /// The median accumulator accumulates the raw input values
 /// as `ScalarValue`s
 ///
@@ -134,18 +167,18 @@ impl PartialEq<dyn Any> for Median {
 /// `merge_batch` and a `Vec` of `ArrayRef` that are converted to scalar values
 /// in the final evaluation step so that we avoid expensive conversions and
 /// allocations during `update_batch`.
-struct MedianAccumulator<T: ArrowNumericType> {
+struct MedianAccumulator<T: ArrowNumericType, V: MedianValues<T = T>>  {
     data_type: DataType,
-    all_values: Vec<T::Native>,
+    all_values: V,
 }

-impl<T: ArrowNumericType> std::fmt::Debug for MedianAccumulator<T> {
+impl<T: ArrowNumericType, V: MedianValues<T = T>> std::fmt::Debug for MedianAccumulator<T, V> {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(f, "MedianAccumulator({})", self.data_type)
     }
 }

-impl<T: ArrowNumericType> Accumulator for MedianAccumulator<T> {
+impl<T: ArrowNumericType, V: MedianValues<T = T>> Accumulator for MedianAccumulator<T, V> {
     fn state(&mut self) -> Result<Vec<ScalarValue>> {
         let all_values = self
             .all_values