@@ -21,8 +21,8 @@
 
 use std::fmt::{Debug, Display, Formatter};
 use std::pin::Pin;
-use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
+use std::sync::{Arc, Weak};
 use std::task::{Context, Poll};
 use std::vec;
 
@@ -33,7 +33,9 @@ use super::{
 };
 use crate::coalesce::LimitedBatchCoalescer;
 use crate::execution_plan::{CardinalityEffect, EvaluationType, SchedulingType};
-use crate::hash_utils::create_hashes;
+#[cfg(not(feature = "force_hash_collisions"))]
+use crate::hash_utils::scatter_dictionary_hashes;
+use crate::hash_utils::{create_hashes, hash_single_array};
 use crate::metrics::{BaselineMetrics, SpillMetrics};
 use crate::projection::{ProjectionExec, all_columns, make_with_child, update_expr};
 use crate::sorts::streaming_merge::StreamingMergeBuilder;
@@ -46,9 +48,12 @@ use crate::{
     PlanProperties, ReplaceChildrenOptions, Statistics, validate_child_count,
 };
 
-use arrow::array::{PrimitiveArray, RecordBatch, RecordBatchOptions, UInt64Array};
+use arrow::array::{
+    Array, ArrayRef, PrimitiveArray, RecordBatch, RecordBatchOptions, UInt64Array,
+};
 use arrow::compute::take_arrays;
 use arrow::datatypes::{DataType, Schema, SchemaRef, UInt32Type};
+use arrow::downcast_dictionary_array;
 use arrow_schema::SortOptions;
 use datafusion_common::ScalarValue;
 use datafusion_common::config::ConfigOptions;
@@ -657,6 +662,7 @@ enum BatchPartitionerState {
         partition_reducer: StrengthReducedU64,
         hash_buffer: Vec<u64>,
         indices: Vec<Vec<u32>>,
+        dictionary_hash_cache: Vec<DictionaryHashCache>,
     },
     RoundRobin {
         num_partitions: usize,
@@ -672,6 +678,40 @@ enum BatchPartitionerState {
     },
 }
 
+#[derive(Default)]
+struct DictionaryHashCache {
+    /// Dictionary values these hashes belong to.
+    values: Option<Weak<dyn Array>>,
+    /// Hash for each dictionary value.
+    hashes: Vec<u64>,
+    /// Number of times values were hashed.
+    #[cfg(test)]
+    rebuilds: usize,
+}
+
+impl DictionaryHashCache {
+    fn update(&mut self, values: &ArrayRef) -> Result<&[u64]> {
+        let unchanged = self.values.as_ref().is_some_and(|cached| {
+            std::ptr::addr_eq(cached.as_ptr(), Arc::as_ptr(values))
+        });
+        if !unchanged {
+            self.hashes.clear();
+            self.hashes.resize(values.len(), 0);
+            create_hashes(
+                [values],
+                REPARTITION_RANDOM_STATE.random_state(),
+                &mut self.hashes,
+            )?;
+            self.values = Some(Arc::downgrade(values));
+            #[cfg(test)]
+            {
+                self.rebuilds += 1;
+            }
+        }
+        Ok(&self.hashes)
+    }
+}
+
 /// Fixed RandomState used for hash repartitioning to ensure consistent behavior across
 /// executions and runs.
 pub const REPARTITION_RANDOM_STATE: SeededRandomState = SeededRandomState::with_seed(0);
@@ -1036,6 +1076,7 @@ impl BatchPartitioner {
                 partition_reducer: StrengthReducedU64::new(num_partitions as u64),
                 hash_buffer: vec![],
                 indices: vec![vec![]; num_partitions],
+                dictionary_hash_cache: vec![],
             },
             timer,
         })
@@ -1250,6 +1291,7 @@ impl BatchPartitioner {
                     partition_reducer,
                     hash_buffer,
                     indices,
+                    dictionary_hash_cache,
                 } => {
                     // Tracking time required for distributing indexes across output partitions
                     let timer = self.timer.timer();
@@ -1260,11 +1302,55 @@ impl BatchPartitioner {
                     hash_buffer.clear();
                     hash_buffer.resize(batch.num_rows(), 0);
 
-                    create_hashes(
-                        &arrays,
-                        REPARTITION_RANDOM_STATE.random_state(),
-                        hash_buffer,
-                    )?;
+                    dictionary_hash_cache
+                        .resize_with(arrays.len(), DictionaryHashCache::default);
+                    for (index, (array, cache)) in arrays
+                        .iter()
+                        .zip(dictionary_hash_cache.iter_mut())
+                        .enumerate()
+                    {
+                        let array = array.as_ref();
+                        if matches!(array.data_type(), DataType::Dictionary(_, _)) {
+                            downcast_dictionary_array! {
+                                array => {
+                                    let hashes = cache.update(array.values())?;
+                                    #[cfg(not(feature = "force_hash_collisions"))]
+                                    scatter_dictionary_hashes(
+                                        array,
+                                        hashes,
+                                        hash_buffer,
+                                        index != 0,
+                                    );
+                                    #[cfg(feature = "force_hash_collisions")]
+                                    let _ = hashes;
+                                    #[cfg(feature = "force_hash_collisions")]
+                                    hash_single_array(
+                                        array,
+                                        REPARTITION_RANDOM_STATE.random_state(),
+                                        hash_buffer,
+                                        index != 0,
+                                    )?;
+                                },
+                                _ => unreachable!(),
+                            }
+                        } else {
+                            *cache = DictionaryHashCache::default();
+                            hash_single_array(
+                                array,
+                                REPARTITION_RANDOM_STATE.random_state(),
+                                hash_buffer,
+                                index != 0,
+                            )?;
+                        }
+                    }
+
+                    if arrays.is_empty() {
+                        create_hashes(
+                            arrays,
+                            REPARTITION_RANDOM_STATE.random_state(),
+                            hash_buffer,
+                        )?;
+                    }
 
                     for v in indices.iter_mut() {
                         v.clear();
@@ -2660,10 +2746,10 @@ mod tests {
     };
 
     use arrow::array::{
-        Array, ArrayRef, Decimal128Array, Int64Array, StringArray,
-        TimestampNanosecondArray, UInt32Array,
+        Array, ArrayRef, Decimal128Array, DictionaryArray, Int32Array, Int64Array,
+        StringArray, TimestampNanosecondArray, UInt32Array,
     };
-    use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
+    use arrow::datatypes::{DataType, Field, Int32Type, Schema, TimeUnit};
     use datafusion_common::ScalarValue;
     use datafusion_common::cast::{as_string_array, as_uint32_array};
     use datafusion_common::config::SpillCompression;
@@ -2918,6 +3004,83 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn dictionary_hash_cache_rebuilds_when_values_change() -> Result<()> {
+        let first: ArrayRef =
+            Arc::new(StringArray::from(vec![Some("a"), None, Some("ccc")]));
+        let second: ArrayRef =
+            Arc::new(StringArray::from(vec![Some("x"), None, Some("zzz")]));
+        let mut cache = DictionaryHashCache::default();
+
+        let first_hashes = cache.update(&first)?.to_vec();
+        cache.update(&first)?;
+        assert_eq!(cache.rebuilds, 1);
+
+        let second_hashes = cache.update(&second)?.to_vec();
+        assert_eq!(cache.rebuilds, 2);
+        assert_ne!(first_hashes, second_hashes);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn hash_partition_reuses_dictionary_values_across_batches() -> Result<()> {
+        let dictionary_type =
+            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("key", dictionary_type, true),
+            Field::new("id", DataType::UInt32, false),
+        ]));
+        let values: ArrayRef = Arc::new(StringArray::from(vec!["a", "bb", "ccc"]));
+        let make_batch = |keys: Vec<Option<i32>>, ids: Vec<u32>| -> Result<RecordBatch> {
+            let dictionary: ArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
+                Int32Array::from(keys),
+                Arc::clone(&values),
+            )?);
+            Ok(RecordBatch::try_new(
+                Arc::clone(&schema),
+                vec![dictionary, Arc::new(UInt32Array::from(ids))],
+            )?)
+        };
+        let batches = vec![
+            make_batch(vec![Some(0), Some(1), Some(2), None], vec![0, 1, 2, 3])?,
+            make_batch(vec![Some(2), Some(0), None, Some(1)], vec![4, 5, 6, 7])?,
+        ];
+
+        let mut expected = vec![Vec::new(); 4];
+        for batch in &batches {
+            let mut hashes = vec![0; batch.num_rows()];
+            create_hashes(
+                [batch.column(0)],
+                REPARTITION_RANDOM_STATE.random_state(),
+                &mut hashes,
+            )?;
+            let ids = as_uint32_array(batch.column(1))?;
+            for (hash, id) in hashes.into_iter().zip(ids.values().iter().copied()) {
+                expected[(hash % 4) as usize].push(id);
+            }
+        }
+
+        let output = repartition(
+            &schema,
+            vec![batches],
+            Partitioning::Hash(vec![col("key", &schema)?], 4),
+        )
+        .await?;
+        let mut actual = vec![Vec::new(); 4];
+        for (partition, batches) in output.iter().enumerate() {
+            for batch in batches {
+                actual[partition]
+                    .extend(as_uint32_array(batch.column(1))?.values().iter().copied());
+            }
+        }
+        for values in expected.iter_mut().chain(actual.iter_mut()) {
+            values.sort_unstable();
+        }
+
+        assert_eq!(actual, expected);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn many_to_many_range_partition() -> Result<()> {
         let schema = test_schema(false);