diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 283d22767f9b1..20ea0fa35319d 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -28,9 +28,12 @@ use crate::physical_optimizer::test_utils::{
     sort_preserving_merge_exec, union_exec,
 };
 
-use arrow::array::{Int64Array, RecordBatch, UInt8Array, UInt64Array};
+use arrow::array::{
+    Int64Array, RecordBatch, TimestampNanosecondArray, UInt8Array, UInt64Array,
+};
 use arrow::compute::SortOptions;
-use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
+use datafusion::assert_batches_sorted_eq;
 use datafusion::config::ConfigOptions;
 use datafusion::datasource::MemTable;
 use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
@@ -75,7 +78,7 @@ use datafusion_physical_plan::operator_statistics::{
 
 use datafusion_physical_expr::{
     AcrossPartitions, ConstExpr, Distribution, Partitioning, RangePartitioning,
-    SplitPoint,
+    ScalarFunctionExpr, SplitPoint,
 };
 use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion_physical_plan::execution_plan::ExecutionPlan;
@@ -959,6 +962,233 @@ fn range_aggregate_reuses_range_partitioning() -> Result<()> {
     Ok(())
 }
 
+fn timestamp_range_input(split: i64) -> Result<Arc<dyn ExecutionPlan>> {
+    let schema = Arc::new(Schema::new(vec![Field::new(
+        "timestamp",
+        DataType::Timestamp(TimeUnit::Nanosecond, None),
+        false,
+    )]));
+    let hour = 3_600_000_000_000;
+    let (before_split, after_split): (Vec<_>, Vec<_>) =
+        [0, 1, hour, hour + 1, 2 * hour, 2 * hour + 1]
+            .into_iter()
+            .partition(|timestamp| *timestamp < split);
+    let partitions = [before_split, after_split]
+        .into_iter()
+        .map(|values| -> Result<_> {
+            let batch = RecordBatch::try_new(
+                Arc::clone(&schema),
+                vec![Arc::new(TimestampNanosecondArray::from(values))],
+            )?;
+            Ok(vec![batch])
+        })
+        .collect::<Result<Vec<_>>>()?;
+    let range = RangePartitioning::try_new(
+        [PhysicalSortExpr::new_default(col("timestamp", &schema)?)].into(),
+        vec![SplitPoint::new(vec![ScalarValue::TimestampNanosecond(
+            Some(split),
+            None,
+        )])],
+    )?;
+    let source = MemorySourceConfig::try_new_exec(&partitions, schema, None)?;
+    Ok(Arc::new(
+        source
+            .as_ref()
+            .clone()
+            .with_partitioning(Partitioning::Range(range)),
+    ))
+}
+
+fn date_trunc_expr(
+    input: &dyn ExecutionPlan,
+    precision: &str,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    Ok(Arc::new(ScalarFunctionExpr::new(
+        "date_trunc",
+        datafusion_functions::datetime::date_trunc(),
+        vec![lit(precision), col("timestamp", &input.schema())?],
+        Field::new(
+            "bucket",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+            false,
+        )
+        .into(),
+        Arc::new(ConfigOptions::default()),
+    )))
+}
+
+/// Builds separate Partial and Final stages without projected range metadata.
+/// This exercises checking group locality through the partial stage's input.
+fn partial_final_date_trunc_count(
+    input: Arc<dyn ExecutionPlan>,
+    precision: &str,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let input_schema = input.schema();
+    let group_by = PhysicalGroupBy::new_single(vec![(
+        date_trunc_expr(input.as_ref(), precision)?,
+        "bucket".to_string(),
+    )]);
+    let count = Arc::new(
+        AggregateExprBuilder::new(count_udaf(), vec![col("timestamp", &input_schema)?])
+            .schema(Arc::clone(&input_schema))
+            .alias("count")
+            .build()?,
+    );
+    let partial = Arc::new(AggregateExec::try_new(
+        AggregateMode::Partial,
+        group_by.clone(),
+        vec![Arc::clone(&count)],
+        vec![None],
+        input,
+        Arc::clone(&input_schema),
+    )?);
+    assert!(matches!(
+        partial.properties().output_partitioning(),
+        Partitioning::UnknownPartitioning(2)
+    ));
+
+    Ok(Arc::new(AggregateExec::try_new(
+        AggregateMode::FinalPartitioned,
+        group_by.as_final(),
+        vec![count],
+        vec![None],
+        partial,
+        input_schema,
+    )?))
+}
+
+#[tokio::test]
+async fn aligned_hour_groups_avoid_shuffle_through_partial_aggregate() -> Result<()> {
+    let hour = 3_600_000_000_000;
+    let input = timestamp_range_input(hour)?;
+    let aggregate = partial_final_date_trunc_count(input, "hour")?;
+    let config = TestConfig::default()
+        .with_query_execution_partitions(2)
+        .config;
+    let optimized = EnsureRequirements::new().optimize(aggregate, &config)?;
+    SanityCheckPlan::new().optimize(Arc::clone(&optimized), &config)?;
+
+    assert_plan!(
+        optimized,
+        @r"
+    AggregateExec: mode=FinalPartitioned, gby=[bucket@0 as bucket], aggr=[count]
+      AggregateExec: mode=Partial, gby=[date_trunc(hour, timestamp@0) as bucket], aggr=[count]
+        DataSourceExec: partitions=2, partition_sizes=[1, 1]
+    "
+    );
+
+    let batches = collect(optimized, SessionContext::new().task_ctx()).await?;
+    assert_batches_sorted_eq!(
+        [
+            "+---------------------+-------+",
+            "| bucket              | count |",
+            "+---------------------+-------+",
+            "| 1970-01-01T00:00:00 | 2     |",
+            "| 1970-01-01T01:00:00 | 2     |",
+            "| 1970-01-01T02:00:00 | 2     |",
+            "+---------------------+-------+",
+        ],
+        &batches
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn crossing_day_groups_keep_shuffle_through_partial_aggregate() -> Result<()> {
+    let hour = 3_600_000_000_000;
+    let input = timestamp_range_input(hour)?;
+    let aggregate = partial_final_date_trunc_count(input, "day")?;
+    let config = TestConfig::default()
+        .with_query_execution_partitions(2)
+        .config;
+    let optimized = EnsureRequirements::new().optimize(aggregate, &config)?;
+    SanityCheckPlan::new().optimize(Arc::clone(&optimized), &config)?;
+
+    assert_plan!(
+        optimized,
+        @r"
+    AggregateExec: mode=FinalPartitioned, gby=[bucket@0 as bucket], aggr=[count]
+      RepartitionExec: partitioning=Hash([bucket@0], 2), input_partitions=2
+        AggregateExec: mode=Partial, gby=[date_trunc(day, timestamp@0) as bucket], aggr=[count]
+          DataSourceExec: partitions=2, partition_sizes=[1, 1]
+    "
+    );
+
+    let batches = collect(optimized, SessionContext::new().task_ctx()).await?;
+    assert_batches_sorted_eq!(
+        [
+            "+---------------------+-------+",
+            "| bucket              | count |",
+            "+---------------------+-------+",
+            "| 1970-01-01T00:00:00 | 6     |",
+            "+---------------------+-------+",
+        ],
+        &batches
+    );
+    Ok(())
+}
+
+#[test]
+fn range_transform_aggregate_optimization_is_idempotent() -> Result<()> {
+    let hour = 3_600_000_000_000;
+    let config = TestConfig::default()
+        .with_query_execution_partitions(2)
+        .config;
+
+    for precision in ["hour", "day"] {
+        let input = timestamp_range_input(hour)?;
+        let aggregate = partial_final_date_trunc_count(input, precision)?;
+        let optimized = EnsureRequirements::new().optimize(aggregate, &config)?;
+        SanityCheckPlan::new().optimize(Arc::clone(&optimized), &config)?;
+        let reoptimized =
+            EnsureRequirements::new().optimize(Arc::clone(&optimized), &config)?;
+        SanityCheckPlan::new().optimize(Arc::clone(&reoptimized), &config)?;
+        assert_plan!(optimized, reoptimized);
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn range_transform_locality_does_not_satisfy_join_distribution() -> Result<()> {
+    let hour = 3_600_000_000_000;
+    let left = timestamp_range_input(hour)?;
+    let right = timestamp_range_input(2 * hour)?;
+    let left_key = date_trunc_expr(left.as_ref(), "hour")?;
+    let right_key = date_trunc_expr(right.as_ref(), "hour")?;
+    for (input, key) in [(&left, &left_key), (&right, &right_key)] {
+        assert!(input.output_partitioning().keeps_keys_local(
+            &[Arc::clone(key)],
+            input.equivalence_properties(),
+            false,
+        ));
+    }
+    let join =
+        hash_join_exec(left, right, &vec![(left_key, right_key)], &JoinType::Inner);
+    let config = TestConfig::default()
+        .with_query_execution_partitions(2)
+        .config;
+    let optimized = EnsureRequirements::new().optimize(join, &config)?;
+    SanityCheckPlan::new().optimize(Arc::clone(&optimized), &config)?;
+    assert_plan!(
+        optimized,
+        @r"
+    HashJoinExec: mode=Partitioned, join_type=Inner, on=[(date_trunc(hour, timestamp@0), date_trunc(hour, timestamp@0))]
+      RepartitionExec: partitioning=Hash([date_trunc(hour, timestamp@0)], 2), input_partitions=2
+        DataSourceExec: partitions=2, partition_sizes=[1, 1]
+      RepartitionExec: partitioning=Hash([date_trunc(hour, timestamp@0)], 2), input_partitions=2
+        DataSourceExec: partitions=2, partition_sizes=[1, 1]
+    "
+    );
+
+    let batches = collect(optimized, SessionContext::new().task_ctx()).await?;
+    assert_eq!(
+        batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+        3 * 2 * 2,
+        "three hour buckets, each joining two left rows with two right rows"
+    );
+    Ok(())
+}
+
 #[test]
 fn range_grouping_set_aggregate_rehashes_with_grouping_id() -> Result<()> {
     let input = parquet_exec_with_output_partitioning(range_partitioning(
diff --git a/datafusion/expr/src/udf.rs b/datafusion/expr/src/udf.rs
index d937a54029398..f900564d453b6 100644
--- a/datafusion/expr/src/udf.rs
+++ b/datafusion/expr/src/udf.rs
@@ -364,6 +364,15 @@ impl ScalarUDF {
         self.inner.output_ordering(inputs)
     }
 
+    /// See [`ScalarUDFImpl::supports_range_key_locality_analysis`] for more details.
+    pub fn supports_range_key_locality_analysis(
+        &self,
+        argument_types: &[DataType],
+    ) -> bool {
+        self.inner
+            .supports_range_key_locality_analysis(argument_types)
+    }
+
     pub fn preserves_lex_ordering(&self, inputs: &[ExprProperties]) -> Result<bool> {
         self.inner.preserves_lex_ordering(inputs)
     }
@@ -983,6 +992,29 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any {
         Ok(inner(inputs))
     }
 
+    /// Returns whether this function and argument-type combination has been
+    /// audited for range-partitioning locality analysis.
+    ///
+    /// Range-boundary analysis has subtle correctness requirements, so the
+    /// policy is default-deny. Functions may opt in only after individual audit.
+    /// At minimum, an opted-in combination must:
+    ///
+    /// - have same-direction [`SortProperties::Ordered`] metadata over the
+    ///   admitted non-null domain when all other inputs are singleton;
+    /// - preserve nullness exactly, so successful evaluation returns null if and
+    ///   only if the range input is null.
+    ///
+    /// Callers may then prove key locality by evaluating values on both sides of
+    /// each concrete range split. Ordered metadata alone is insufficient because
+    /// a non-strict transform can collapse a split boundary.
+    ///
+    /// This does not claim a concrete partition layout or cross-input
+    /// co-partitioning compatibility. Boundary evaluation errors cause the
+    /// analysis to return `false`. This capability is not transported across FFI.
+    fn supports_range_key_locality_analysis(&self, _argument_types: &[DataType]) -> bool {
+        false
+    }
+
     /// Returns true if the function preserves lexicographical ordering based on
     /// the input ordering.
     ///
@@ -1205,6 +1237,11 @@ impl ScalarUDFImpl for AliasedScalarUDFImpl {
         self.inner.output_ordering(inputs)
     }
 
+    fn supports_range_key_locality_analysis(&self, argument_types: &[DataType]) -> bool {
+        self.inner
+            .supports_range_key_locality_analysis(argument_types)
+    }
+
     fn preserves_lex_ordering(&self, inputs: &[ExprProperties]) -> Result<bool> {
         self.inner.preserves_lex_ordering(inputs)
     }
@@ -1287,6 +1324,12 @@ mod tests {
         assert_eq!(b.partial_cmp(&o), Some(Ordering::Less));
     }
 
+    #[test]
+    fn range_key_locality_analysis_is_disabled_by_default() {
+        let function = test_func("range_function", "a");
+        assert!(!function.supports_range_key_locality_analysis(&[DataType::Int64]));
+    }
+
     fn test_func(name: &'static str, parameter: &'static str) -> ScalarUDF {
         ScalarUDF::from(TestScalarUDFImpl {
             name,
diff --git a/datafusion/functions/src/datetime/date_trunc.rs b/datafusion/functions/src/datetime/date_trunc.rs
index 3d304989afda8..89fd40603f12a 100644
--- a/datafusion/functions/src/datetime/date_trunc.rs
+++ b/datafusion/functions/src/datetime/date_trunc.rs
@@ -571,6 +571,18 @@ impl ScalarUDFImpl for DateTruncFunc {
             Ok(SortProperties::Unordered)
         }
     }
+
+    fn supports_range_key_locality_analysis(&self, argument_types: &[DataType]) -> bool {
+        // Across an America/Goose_Bay transition, ascending epoch seconds
+        // 562129259 < 562129260 truncate to 562129200 > 562125600.
+        // Keep the audited domain to timezone-less timestamp columns.
+        let [precision, timestamp] = argument_types else {
+            return false;
+        };
+
+        precision.is_string() && matches!(timestamp, Timestamp(_, None))
+    }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
@@ -1339,7 +1351,10 @@ mod tests {
     };
 
     use arrow::array::cast::as_primitive_array;
-    use arrow::array::types::{ArrowTimestampType, TimestampNanosecondType};
+    use arrow::array::types::{
+        ArrowTimestampType, TimestampMicrosecondType, TimestampMillisecondType,
+        TimestampNanosecondType, TimestampSecondType,
+    };
     use arrow::array::{
         Array, PrimitiveArray, TimestampMicrosecondArray, TimestampMillisecondArray,
         TimestampNanosecondArray, TimestampSecondArray,
@@ -1394,6 +1409,91 @@ mod tests {
         );
     }
 
+    #[test]
+    fn range_key_locality_analysis_requires_timezone_less_timestamp() {
+        let date_trunc = DateTruncFunc::new();
+
+        assert!(date_trunc.supports_range_key_locality_analysis(&[
+            DataType::Utf8,
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        ]));
+        assert!(!date_trunc.supports_range_key_locality_analysis(&[
+            DataType::Utf8,
+            DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
+        ]));
+        assert!(!date_trunc.supports_range_key_locality_analysis(&[
+            DataType::Int64,
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        ]));
+        assert!(!date_trunc.supports_range_key_locality_analysis(&[DataType::Utf8]));
+    }
+
+    fn invoke_date_trunc(
+        granularity: &str,
+        timestamp: ColumnarValue,
+        timestamp_type: DataType,
+        number_rows: usize,
+    ) -> datafusion_common::Result<ColumnarValue> {
+        DateTruncFunc::new().invoke_with_args(ScalarFunctionArgs {
+            args: vec![ScalarValue::from(granularity).into(), timestamp],
+            arg_fields: vec![
+                Field::new("granularity", DataType::Utf8, false).into(),
+                Field::new("timestamp", timestamp_type.clone(), true).into(),
+            ],
+            number_rows,
+            return_field: Field::new("date_trunc", timestamp_type, true).into(),
+            config_options: Arc::new(ConfigOptions::default()),
+        })
+    }
+
+    fn assert_timezone_less_date_trunc_preserves_nullness<T: ArrowTimestampType>() {
+        let input = PrimitiveArray::<T>::from_iter([Some(0), None]);
+        let timestamp_type = DataType::Timestamp(T::UNIT, None);
+
+        for granularity in [
+            "microsecond",
+            "millisecond",
+            "second",
+            "minute",
+            "hour",
+            "day",
+            "week",
+            "month",
+            "quarter",
+            "year",
+        ] {
+            let output = invoke_date_trunc(
+                granularity,
+                ColumnarValue::Array(Arc::new(input.clone())),
+                timestamp_type.clone(),
+                input.len(),
+            )
+            .unwrap();
+            let ColumnarValue::Array(output) = output else {
+                panic!("expected array output")
+            };
+
+            assert_eq!(
+                input.nulls(),
+                output.nulls(),
+                "{granularity} must preserve nullness for {:?}",
+                T::UNIT
+            );
+        }
+    }
+
+    /// Verifies the exact null contract required by
+    /// [`ScalarUDFImpl::supports_range_key_locality_analysis`] across every
+    /// supported granularity and timezone-less timestamp unit. Ordering metadata
+    /// and concrete range-split separation are tested separately.
+    #[test]
+    fn timezone_less_date_trunc_preserves_nullness() {
+        assert_timezone_less_date_trunc_preserves_nullness::<TimestampSecondType>();
+        assert_timezone_less_date_trunc_preserves_nullness::<TimestampMillisecondType>();
+        assert_timezone_less_date_trunc_preserves_nullness::<TimestampMicrosecondType>();
+        assert_timezone_less_date_trunc_preserves_nullness::<TimestampNanosecondType>();
+    }
+
     #[test]
     fn date_trunc_test() {
         let cases = vec![
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index 1c2d970ccffa4..ed824e5712f96 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -39,6 +39,7 @@ use crate::{
     PhysicalSortRequirement,
 };
 
+use arrow::compute::SortOptions;
 use arrow::datatypes::SchemaRef;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{Constraint, Constraints, HashMap, Result, plan_err};
@@ -1408,6 +1409,29 @@ impl EquivalenceProperties {
             .unwrap_or_else(|_| ExprProperties::new_unknown())
     }
 
+    /// Returns whether `transformed_expr` reports the same ordering direction
+    /// when `source_expr` is ordered and all other inputs are singleton.
+    pub(crate) fn is_same_direction_ordered_transform(
+        &self,
+        transformed_expr: &Arc<dyn PhysicalExpr>,
+        source_expr: &Arc<dyn PhysicalExpr>,
+    ) -> bool {
+        if transformed_expr.eq(source_expr) {
+            return false;
+        }
+
+        let options = SortOptions::default();
+        let dependencies = Dependencies::new(std::iter::once(PhysicalSortExpr::new(
+            Arc::clone(source_expr),
+            options,
+        )));
+        matches!(
+            get_expr_properties(transformed_expr, &dependencies, &self.schema)
+                .map(|properties| properties.sort_properties),
+            Ok(SortProperties::Ordered(actual)) if actual == options
+        )
+    }
+
     /// Transforms this `EquivalenceProperties` by mapping columns in the
     /// original schema to columns in the new schema by index.
     pub fn with_new_schema(mut self, schema: SchemaRef) -> Result<Self> {
diff --git a/datafusion/physical-expr/src/partitioning.rs b/datafusion/physical-expr/src/partitioning.rs
index 336d09b987734..15d7849ebeac1 100644
--- a/datafusion/physical-expr/src/partitioning.rs
+++ b/datafusion/physical-expr/src/partitioning.rs
@@ -17,13 +17,17 @@
 
 //! [`Partitioning`] and [`Distribution`] for `ExecutionPlans`
 
+use crate::expressions::{Literal, UnKnownColumn};
+use crate::simplifier::const_evaluator::create_dummy_batch;
 use crate::{
-    EquivalenceProperties, PhysicalExpr, equivalence::ProjectionMapping,
-    expressions::UnKnownColumn, physical_exprs_contains, physical_exprs_equal,
+    EquivalenceProperties, PhysicalExpr, ScalarFunctionExpr,
+    equivalence::ProjectionMapping, physical_exprs_contains, physical_exprs_equal,
 };
 use arrow::datatypes::Schema;
 pub use datafusion_common::SplitPoint;
-use datafusion_common::{Result, plan_err, validate_range_split_points};
+use datafusion_common::tree_node::{Transformed, TreeNode};
+use datafusion_common::{Result, ScalarValue, plan_err, validate_range_split_points};
+use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr_common::physical_expr::format_physical_expr_list;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
 #[cfg(feature = "proto")]
@@ -532,6 +536,196 @@ fn normalize_exprs(
         .collect()
 }
 
+fn evaluate_with_range_value(
+    expr: &Arc<dyn PhysicalExpr>,
+    range_key: &Arc<dyn PhysicalExpr>,
+    value: &ScalarValue,
+) -> Option<ScalarValue> {
+    let literal: Arc<dyn PhysicalExpr> = Arc::new(Literal::new(value.clone()));
+    let rewritten = Arc::clone(expr)
+        .transform(|node| {
+            if node.eq(range_key) {
+                Ok(Transformed::yes(Arc::clone(&literal)))
+            } else {
+                Ok(Transformed::no(node))
+            }
+        })
+        .ok()?;
+    if !rewritten.transformed {
+        return None;
+    }
+
+    match rewritten.data.evaluate(create_dummy_batch().ok()?).ok()? {
+        ColumnarValue::Scalar(value) => Some(value),
+        ColumnarValue::Array(array) => ScalarValue::try_from_array(&array, 0).ok(),
+    }
+}
+
+fn checked_adjacent_timestamp(
+    value: &ScalarValue,
+    is_successor: bool,
+) -> Option<ScalarValue> {
+    let offset = if is_successor { 1 } else { -1 };
+    match value {
+        ScalarValue::TimestampSecond(Some(value), None) => value
+            .checked_add(offset)
+            .map(|value| ScalarValue::TimestampSecond(Some(value), None)),
+        ScalarValue::TimestampMillisecond(Some(value), None) => value
+            .checked_add(offset)
+            .map(|value| ScalarValue::TimestampMillisecond(Some(value), None)),
+        ScalarValue::TimestampMicrosecond(Some(value), None) => value
+            .checked_add(offset)
+            .map(|value| ScalarValue::TimestampMicrosecond(Some(value), None)),
+        ScalarValue::TimestampNanosecond(Some(value), None) => value
+            .checked_add(offset)
+            .map(|value| ScalarValue::TimestampNanosecond(Some(value), None)),
+        _ => None,
+    }
+}
+
+/// Determines whether a nondecreasing transform keeps equal output keys local.
+///
+/// For one ascending range key, split points `x₁, ..., xₙ` define:
+///
+/// ```text
+/// P₀ = (-∞, x₁)
+/// Pᵢ = [xᵢ, xᵢ₊₁) for 1 ≤ i < n
+/// Pₙ = [xₙ, +∞)
+/// ```
+///
+/// Aggregation can reuse these partitions after applying `f` only if no output
+/// key occurs in two partitions. In other words, the partition images must be
+/// disjoint: `f(Pᵢ) ∩ f(Pⱼ) = ∅` for `i ≠ j`.
+///
+/// Monotonicity alone is insufficient. Ordered metadata gives only
+///
+/// ```text
+/// f(predecessor(xᵢ)) ≤ f(xᵢ)
+/// ```
+///
+/// but locality requires strict separation at every split:
+///
+/// ```text
+/// f(predecessor(xᵢ)) < f(xᵢ)
+/// ```
+///
+/// Since ordering already supplies the non-strict inequality, the analysis
+/// proves strictness by explicitly checking that the two outputs are not equal.
+/// Otherwise, the same transformed key could occur in both adjacent partitions,
+/// so the transform would not preserve key locality. For example, an hourly
+/// truncation maps timestamps immediately before and at a `01:30` split to the
+/// same `01:00` bucket. A split at `01:00` is safe because its predecessor maps
+/// to `00:00`, while the split itself maps to `01:00`.
+///
+/// `predecessor(xᵢ)` is the greatest representable timestamp below the split.
+/// Therefore, for every `x < xᵢ` and `y ≥ xᵢ`, ordering and the strict boundary
+/// check give
+///
+/// ```text
+/// f(x) ≤ f(predecessor(xᵢ)) < f(xᵢ) ≤ f(y)
+/// ```
+///
+/// Thus no transformed key can occur on both sides of the split. The successor of
+/// an ascending split is already in the same partition as `xᵢ`, so checking it
+/// would not say anything about groups crossing that split.
+///
+/// The argument is the same for descending ranges, but `successor(xᵢ)` must be
+/// used instead. Null preservation is also required so the transform cannot
+/// introduce a null group outside the source partition selected by null order.
+fn range_splits_separate_transformed_keys(
+    function: &Arc<dyn PhysicalExpr>,
+    range_key: &Arc<dyn PhysicalExpr>,
+    range: &RangePartitioning,
+    schema: &Schema,
+) -> bool {
+    let Ok(range_type) = range_key.data_type(schema) else {
+        return false;
+    };
+    let sort_options = range.ordering().first().options;
+    if validate_range_split_points(range.split_points(), &[sort_options]).is_err() {
+        return false;
+    }
+    range.split_points().iter().all(|split_point| {
+        let Some(split_value) = split_point.values().first() else {
+            return false;
+        };
+        if split_value.data_type() != range_type {
+            return false;
+        }
+        let Some(adjacent) =
+            checked_adjacent_timestamp(split_value, sort_options.descending)
+        else {
+            return false;
+        };
+        let Some(at_split) = evaluate_with_range_value(function, range_key, split_value)
+        else {
+            return false;
+        };
+        let Some(across_split) =
+            evaluate_with_range_value(function, range_key, &adjacent)
+        else {
+            return false;
+        };
+
+        !at_split.is_null()
+            && !across_split.is_null()
+            && at_split.data_type() == across_split.data_type()
+            && at_split != across_split
+    })
+}
+
+fn range_transform_proves_key_locality(
+    candidate: &Arc<dyn PhysicalExpr>,
+    range_key: &Arc<dyn PhysicalExpr>,
+    range: &RangePartitioning,
+    eq_properties: &EquivalenceProperties,
+) -> bool {
+    let Some(function) = candidate.downcast_ref::<ScalarFunctionExpr>() else {
+        return false;
+    };
+    let schema = eq_properties.schema();
+    if datafusion_physical_expr_common::physical_expr::is_volatile(candidate)
+        || !function
+            .supports_range_key_locality_analysis(schema)
+            .unwrap_or(false)
+        || !eq_properties.is_same_direction_ordered_transform(candidate, range_key)
+    {
+        return false;
+    }
+
+    let mut range_argument_count = 0;
+    for argument in function.args() {
+        if eq_properties.eq_group().exprs_equal(argument, range_key) {
+            range_argument_count += 1;
+        } else if !argument.is::<Literal>() {
+            return false;
+        }
+    }
+
+    range_argument_count == 1
+        && range_splits_separate_transformed_keys(candidate, range_key, range, schema)
+}
+
+fn range_partitioning_keeps_keys_local(
+    range: &RangePartitioning,
+    required_exprs: &[Arc<dyn PhysicalExpr>],
+    eq_properties: &EquivalenceProperties,
+    allow_subset: bool,
+) -> bool {
+    // Currently, transform locality analysis only supports single-key ranges.
+    // Extending the proof to compound range keys is tracked in #24644.
+    if range.ordering().len() != 1 || required_exprs.is_empty() {
+        return false;
+    }
+
+    let range_key = &range.ordering()[0].expr;
+    let has_key_locality_proof = required_exprs.iter().any(|required| {
+        range_transform_proves_key_locality(required, range_key, range, eq_properties)
+    });
+
+    has_key_locality_proof && (required_exprs.len() == 1 || allow_subset)
+}
+
 /// Represents how a [`Partitioning`] satisfies a [`Distribution`] requirement.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PartitioningSatisfaction {
@@ -646,6 +840,41 @@ impl Partitioning {
         }
     }
 
+    /// Returns true when it can prove that equal tuples of `required_exprs`
+    /// cannot occur in different partitions.
+    ///
+    /// Unlike [`Self::satisfaction`], this does not claim a concrete partition
+    /// layout. This weaker guarantee is sufficient for independent keyed
+    /// operations such as aggregation, but not for co-partitioning multiple
+    /// inputs. Transform analysis is currently limited to single-key range
+    /// partitioning.
+    pub fn keeps_keys_local(
+        &self,
+        required_exprs: &[Arc<dyn PhysicalExpr>],
+        eq_properties: &EquivalenceProperties,
+        allow_subset: bool,
+    ) -> bool {
+        let required = Distribution::KeyPartitioned(required_exprs.to_vec());
+        if self
+            .satisfaction(&required, eq_properties, allow_subset)
+            .is_satisfied()
+        {
+            return true;
+        }
+
+        match self {
+            Self::Range(range) => range_partitioning_keeps_keys_local(
+                range,
+                required_exprs,
+                eq_properties,
+                allow_subset,
+            ),
+            Self::RoundRobinBatch(_)
+            | Self::Hash(_, _)
+            | Self::UnknownPartitioning(_) => false,
+        }
+    }
+
     fn key_satisfaction(
         partition_exprs: &[Arc<dyn PhysicalExpr>],
         required_exprs: &[Arc<dyn PhysicalExpr>],
@@ -991,7 +1220,8 @@ mod tests {
     use crate::projection::ProjectionTargets;
 
     use arrow::compute::SortOptions;
-    use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+    use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
+    use datafusion_common::config::ConfigOptions;
     use datafusion_common::{Result, ScalarValue};
 
     struct PartitioningTestFixture {
@@ -1090,6 +1320,17 @@ mod tests {
         ) -> Partitioning {
             Partitioning::Range(self.range(indices, split_points))
         }
+
+        fn range_partitioning_with_ordering(
+            &self,
+            ordering: LexOrdering,
+            split_points: Vec<SplitPoint>,
+        ) -> Partitioning {
+            Partitioning::Range(
+                RangePartitioning::try_new(ordering, split_points)
+                    .expect("test range partitioning should be valid"),
+            )
+        }
     }
 
     fn assert_satisfaction(
@@ -1112,6 +1353,625 @@ mod tests {
         );
     }
 
+    fn assert_key_locality(
+        desc: &str,
+        partitioning: &Partitioning,
+        required_exprs: &[Arc<dyn PhysicalExpr>],
+        eq_properties: &EquivalenceProperties,
+        expected_with_subset: bool,
+        expected_without_subset: bool,
+    ) {
+        assert_eq!(
+            partitioning.keeps_keys_local(required_exprs, eq_properties, true),
+            expected_with_subset,
+            "Failed for {desc} with subset enabled"
+        );
+        assert_eq!(
+            partitioning.keeps_keys_local(required_exprs, eq_properties, false),
+            expected_without_subset,
+            "Failed for {desc} with subset disabled"
+        );
+    }
+
+    fn date_trunc_of(
+        timestamp: Arc<dyn PhysicalExpr>,
+        precision: &str,
+        timezone: Option<&str>,
+    ) -> Arc<dyn PhysicalExpr> {
+        let timezone = timezone.map(Into::into);
+        Arc::new(ScalarFunctionExpr::new(
+            "date_trunc",
+            datafusion_functions::datetime::date_trunc(),
+            vec![
+                Arc::new(Literal::new(ScalarValue::Utf8(Some(precision.into())))),
+                timestamp,
+            ],
+            Field::new(
+                "date_trunc",
+                DataType::Timestamp(TimeUnit::Nanosecond, timezone),
+                true,
+            )
+            .into(),
+            Arc::new(ConfigOptions::default()),
+        ))
+    }
+
+    fn timestamp_split(value: Option<i64>, timezone: Option<&str>) -> SplitPoint {
+        SplitPoint::new(vec![ScalarValue::TimestampNanosecond(
+            value,
+            timezone.map(Into::into),
+        )])
+    }
+
+    /// Assigns sample rows to source ranges and evaluates whole arrays,
+    /// independently of the optimizer's boundary-adjacency proof.
+    fn date_trunc_partition_images_are_disjoint<
+        T: arrow::array::types::ArrowTimestampType,
+    >(
+        function: &Arc<dyn PhysicalExpr>,
+        splits: &[i64],
+        options: SortOptions,
+        schema: &SchemaRef,
+    ) -> Result<bool> {
+        use arrow::array::{PrimitiveArray, RecordBatch};
+        use datafusion_common::HashSet;
+
+        let mut partitions = vec![vec![]; splits.len() + 1];
+        let rows = splits
+            .iter()
+            .flat_map(|split| [Some(split - 1), Some(*split), Some(split + 1)])
+            .chain([None, None]);
+        for value in rows {
+            let index = match value {
+                None if options.nulls_first => 0,
+                None => splits.len(),
+                Some(value) => splits
+                    .iter()
+                    .filter(|split| {
+                        if options.descending {
+                            value <= **split
+                        } else {
+                            value >= **split
+                        }
+                    })
+                    .count(),
+            };
+            partitions[index].push(value);
+        }
+        let images = partitions
+            .into_iter()
+            .map(|values| {
+                let input = Arc::new(PrimitiveArray::<T>::from_iter(values));
+                let batch = RecordBatch::try_new(Arc::clone(schema), vec![input])?;
+                let output = function.evaluate(&batch)?.into_array(batch.num_rows())?;
+                (0..output.len())
+                    .map(|i| ScalarValue::try_from_array(&output, i))
+                    .collect::<Result<HashSet<_>>>()
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(images.iter().enumerate().all(|(i, left)| {
+            images
+                .iter()
+                .skip(i + 1)
+                .all(|right| left.is_disjoint(right))
+        }))
+    }
+
+    fn assert_date_trunc_partition_images<T: arrow::array::types::ArrowTimestampType>()
+    -> Result<()> {
+        let ticks_per_second = match T::UNIT {
+            TimeUnit::Second => 1,
+            TimeUnit::Millisecond => 1_000,
+            TimeUnit::Microsecond => 1_000_000,
+            TimeUnit::Nanosecond => 1_000_000_000,
+        };
+        let hour = 3600 * ticks_per_second;
+        let data_type = DataType::Timestamp(T::UNIT, None);
+        let schema =
+            Arc::new(Schema::new(vec![Field::new("ts", data_type.clone(), true)]));
+        let timestamp: Arc<dyn PhysicalExpr> = Arc::new(Column::new("ts", 0));
+        let properties = EquivalenceProperties::new(Arc::clone(&schema));
+
+        let mut cases = vec![
+            (
+                "aligned ascending hours",
+                "hour",
+                false,
+                vec![-hour, 0, hour],
+                true,
+            ),
+            (
+                "unaligned ascending hours",
+                "hour",
+                false,
+                vec![-hour + 1, 1, hour + 1],
+                false,
+            ),
+            (
+                "days cross hourly splits",
+                "day",
+                false,
+                vec![-hour, 0, hour],
+                false,
+            ),
+            (
+                "descending splits inside hour buckets",
+                "hour",
+                true,
+                vec![-hour, 0, hour],
+                false,
+            ),
+            (
+                "descending splits between hour buckets",
+                "hour",
+                true,
+                vec![-hour - 1, -1, hour - 1],
+                true,
+            ),
+        ];
+        for (precision, divisor) in [("millisecond", 1_000), ("microsecond", 1_000_000)] {
+            let width = (ticks_per_second / divisor).max(1);
+            cases.extend([
+                (
+                    "aligned ascending subsecond splits",
+                    precision,
+                    false,
+                    vec![-width, 0, width],
+                    true,
+                ),
+                // In particular, microsecond input with millisecond precision
+                // must reject split -999: both adjacent array values map to -1000.
+                (
+                    "offset ascending subsecond split",
+                    precision,
+                    false,
+                    vec![-width + 1],
+                    width == 1,
+                ),
+                (
+                    "descending splits at subsecond bucket starts",
+                    precision,
+                    true,
+                    vec![-width, 0, width],
+                    width == 1,
+                ),
+                (
+                    "descending splits at subsecond bucket ends",
+                    precision,
+                    true,
+                    vec![-width - 1, -1, width - 1],
+                    true,
+                ),
+            ]);
+        }
+        for (description, precision, descending, source_splits, expect_disjoint) in cases
+        {
+            for nulls_first in [false, true] {
+                let options = SortOptions::new(descending, nulls_first);
+                let mut splits = source_splits.clone();
+                if descending {
+                    splits.reverse();
+                }
+                let range = Partitioning::Range(RangePartitioning::try_new(
+                    [PhysicalSortExpr::new(Arc::clone(&timestamp), options)].into(),
+                    splits
+                        .iter()
+                        .map(|value| {
+                            SplitPoint::new(vec![ScalarValue::new_timestamp::<T>(
+                                Some(*value),
+                                None,
+                            )])
+                        })
+                        .collect(),
+                )?);
+                let function: Arc<dyn PhysicalExpr> = Arc::new(ScalarFunctionExpr::new(
+                    "date_trunc",
+                    datafusion_functions::datetime::date_trunc(),
+                    vec![
+                        Arc::new(Literal::new(ScalarValue::from(precision))),
+                        Arc::clone(&timestamp),
+                    ],
+                    Field::new("bucket", data_type.clone(), true).into(),
+                    Arc::new(ConfigOptions::default()),
+                ));
+                assert_eq!(
+                    range.keeps_keys_local(&[Arc::clone(&function)], &properties, false),
+                    expect_disjoint,
+                    "optimizer: {description}, {precision}, {:?}, {options:?}, splits={splits:?}",
+                    T::UNIT
+                );
+                assert_eq!(
+                    date_trunc_partition_images_are_disjoint::<T>(
+                        &function, &splits, options, &schema,
+                    )?,
+                    expect_disjoint,
+                    "array evaluation: {description}, {precision}, {:?}, {options:?}, splits={splits:?}",
+                    T::UNIT
+                );
+            }
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn date_trunc_partition_images_are_disjoint_only_for_safe_splits() -> Result<()> {
+        use arrow::array::types::{
+            TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType,
+            TimestampSecondType,
+        };
+        assert_date_trunc_partition_images::<TimestampSecondType>()?;
+        assert_date_trunc_partition_images::<TimestampMillisecondType>()?;
+        assert_date_trunc_partition_images::<TimestampMicrosecondType>()?;
+        assert_date_trunc_partition_images::<TimestampNanosecondType>()
+    }
+
+    #[test]
+    fn range_transform_requires_literal_parameters() -> Result<()> {
+        use crate::expressions::CastExpr;
+        let fixture = PartitioningTestFixture::new(vec![(
+            "ts",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let function = date_trunc_of(fixture.col(0), "hour", None);
+        let scalar_but_not_literal: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new(
+            Arc::new(Literal::new(ScalarValue::from("hour"))),
+            DataType::Utf8View,
+            None,
+        ));
+        let function =
+            function.with_new_children(vec![scalar_but_not_literal, fixture.col(0)])?;
+        let range = fixture.range_partitioning([0], vec![timestamp_split(Some(0), None)]);
+        assert!(!range.keeps_keys_local(&[function], &fixture.eq_properties, false));
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_establishes_key_locality_only() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![
+            ("timestamp", DataType::Timestamp(TimeUnit::Nanosecond, None)),
+            ("tag", DataType::Utf8),
+        ])?;
+        let hour = 1_704_070_800_000_000_000;
+        let aligned =
+            fixture.range_partitioning([0], vec![timestamp_split(Some(hour), None)]);
+        let trunc_hour = date_trunc_of(fixture.col(0), "hour", None);
+
+        let exact_key_exprs = vec![Arc::clone(&trunc_hour)];
+        assert_eq!(
+            aligned.satisfaction(
+                &Distribution::KeyPartitioned(exact_key_exprs.clone()),
+                &fixture.eq_properties,
+                false,
+            ),
+            PartitioningSatisfaction::NotSatisfied
+        );
+        assert_key_locality(
+            "aligned transformed key",
+            &aligned,
+            &exact_key_exprs,
+            &fixture.eq_properties,
+            true,
+            true,
+        );
+
+        assert_key_locality(
+            "additional grouping keys require subset permission",
+            &aligned,
+            &[trunc_hour, fixture.col(1)],
+            &fixture.eq_properties,
+            true,
+            false,
+        );
+
+        let trunc_day = date_trunc_of(fixture.col(0), "day", None);
+        assert_key_locality(
+            "an hour split straddles a day bucket",
+            &aligned,
+            &[trunc_day],
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+
+        let two_aligned = fixture.range_partitioning(
+            [0],
+            vec![
+                timestamp_split(Some(hour), None),
+                timestamp_split(Some(hour + 3_600_000_000_000), None),
+            ],
+        );
+        assert_key_locality(
+            "every split must be aligned",
+            &two_aligned,
+            &exact_key_exprs,
+            &fixture.eq_properties,
+            true,
+            true,
+        );
+
+        let unaligned =
+            fixture.range_partitioning([0], vec![timestamp_split(Some(hour + 1), None)]);
+        assert_key_locality(
+            "an unaligned split straddles an hour bucket",
+            &unaligned,
+            &exact_key_exprs,
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_rejects_compound_range_keys() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![
+            ("key", DataType::Utf8),
+            ("timestamp", DataType::Timestamp(TimeUnit::Nanosecond, None)),
+        ])?;
+        let split = SplitPoint::new(vec![
+            ScalarValue::Utf8(Some("m".into())),
+            ScalarValue::TimestampNanosecond(Some(1_704_070_800_000_000_000), None),
+        ]);
+        let range = fixture.range_partitioning([0, 1], vec![split]);
+
+        // Start with single-key ranges; compound transforms can follow once
+        // tuple-boundary safety is established.
+        assert!(!range.keeps_keys_local(
+            &[fixture.col(0), date_trunc_of(fixture.col(1), "hour", None),],
+            &fixture.eq_properties,
+            false,
+        ));
+        Ok(())
+    }
+
+    #[test]
+    #[expect(
+        deprecated,
+        reason = "test intentionally constructs malformed range splits"
+    )]
+    fn range_transform_rejects_malformed_split_points() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let required = vec![date_trunc_of(fixture.col(0), "hour", None)];
+        let hour = 3_600_000_000_000;
+        let timestamp = |value| ScalarValue::TimestampNanosecond(Some(value), None);
+
+        for (description, descending, splits) in [
+            (
+                "reversed ascending splits",
+                false,
+                vec![vec![timestamp(2 * hour)], vec![timestamp(hour)]],
+            ),
+            (
+                "reversed descending splits",
+                true,
+                vec![vec![timestamp(hour - 1)], vec![timestamp(2 * hour - 1)]],
+            ),
+            (
+                "duplicate ascending splits",
+                false,
+                vec![vec![timestamp(hour)], vec![timestamp(hour)]],
+            ),
+            (
+                "duplicate descending splits",
+                true,
+                vec![vec![timestamp(hour - 1)], vec![timestamp(hour - 1)]],
+            ),
+            (
+                "overwide split tuple",
+                false,
+                vec![vec![timestamp(hour), ScalarValue::Int64(Some(0))]],
+            ),
+            ("empty split tuple", false, vec![vec![]]),
+        ] {
+            let range = Partitioning::Range(RangePartitioning::new(
+                [PhysicalSortExpr::new(
+                    fixture.col(0),
+                    SortOptions::new(descending, false),
+                )]
+                .into(),
+                splits.into_iter().map(SplitPoint::new).collect(),
+            ));
+            assert_key_locality(
+                description,
+                &range,
+                &required,
+                &fixture.eq_properties,
+                false,
+                false,
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_rejects_null_or_minimum_timestamp_splits() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let required = vec![date_trunc_of(fixture.col(0), "hour", None)];
+
+        for (description, split) in [
+            ("null", timestamp_split(None, None)),
+            (
+                "minimum has no predecessor",
+                timestamp_split(Some(i64::MIN), None),
+            ),
+        ] {
+            assert_key_locality(
+                description,
+                &fixture.range_partitioning([0], vec![split]),
+                &required,
+                &fixture.eq_properties,
+                false,
+                false,
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    #[expect(
+        deprecated,
+        reason = "test intentionally constructs a malformed range split"
+    )]
+    fn range_transform_rejects_split_type_mismatch() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let required = vec![date_trunc_of(fixture.col(0), "hour", None)];
+        let mismatched_split = Partitioning::Range(RangePartitioning::new(
+            fixture.range_ordering([0]),
+            vec![SplitPoint::new(vec![ScalarValue::Int64(Some(10))])],
+        ));
+        assert_key_locality(
+            "split scalar type must match the range key",
+            &mismatched_split,
+            &required,
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_rejects_boundary_evaluation_errors() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let invalid_precision =
+            vec![date_trunc_of(fixture.col(0), "not-a-precision", None)];
+        let aligned = fixture.range_partitioning(
+            [0],
+            vec![timestamp_split(Some(1_704_070_800_000_000_000), None)],
+        );
+        assert_key_locality(
+            "split evaluation errors fail closed",
+            &aligned,
+            &invalid_precision,
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_rejects_adjacent_timestamp_evaluation_errors() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let nanoseconds_per_hour = 3_600_000_000_000;
+        let first_representable_hour =
+            i64::MIN / nanoseconds_per_hour * nanoseconds_per_hour;
+        let split =
+            ScalarValue::TimestampNanosecond(Some(first_representable_hour), None);
+        let predecessor =
+            ScalarValue::TimestampNanosecond(Some(first_representable_hour - 1), None);
+        let timestamp = fixture.col(0);
+        let trunc_hour = date_trunc_of(Arc::clone(&timestamp), "hour", None);
+
+        assert_eq!(
+            evaluate_with_range_value(&trunc_hour, &timestamp, &split),
+            Some(split.clone()),
+        );
+        assert_eq!(
+            evaluate_with_range_value(&trunc_hour, &timestamp, &predecessor),
+            None,
+        );
+
+        let range = fixture.range_partitioning([0], vec![SplitPoint::new(vec![split])]);
+        assert_key_locality(
+            "predecessor truncation underflows even though split truncation succeeds",
+            &range,
+            &[trunc_hour],
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn range_transform_rejects_timezone_annotated_timestamps() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
+        )])?;
+        let trunc_hour = date_trunc_of(fixture.col(0), "hour", Some("UTC"));
+        let range = fixture.range_partitioning(
+            [0],
+            vec![timestamp_split(
+                Some(1_704_070_800_000_000_000),
+                Some("UTC"),
+            )],
+        );
+        assert_key_locality(
+            "timezone-aware date_trunc is not audited",
+            &range,
+            &[trunc_hour],
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn descending_range_uses_the_open_side_successor() -> Result<()> {
+        let fixture = PartitioningTestFixture::new(vec![(
+            "timestamp",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+        )])?;
+        let hour = 1_704_070_800_000_000_000;
+        let ordering: LexOrdering =
+            [fixture.range_sort_expr(0, SortOptions::new(true, false))].into();
+        let required = vec![date_trunc_of(fixture.col(0), "hour", None)];
+
+        assert_key_locality(
+            "the successor of an hour boundary is in the same bucket",
+            &fixture.range_partitioning_with_ordering(
+                ordering.clone(),
+                vec![timestamp_split(Some(hour), None)],
+            ),
+            &required,
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        assert_key_locality(
+            "the successor of the final nanosecond starts a new bucket",
+            &fixture.range_partitioning_with_ordering(
+                ordering.clone(),
+                vec![timestamp_split(Some(hour + 3_600_000_000_000 - 1), None)],
+            ),
+            &required,
+            &fixture.eq_properties,
+            true,
+            true,
+        );
+        assert_key_locality(
+            "the maximum timestamp has no successor",
+            &fixture.range_partitioning_with_ordering(
+                ordering,
+                vec![timestamp_split(Some(i64::MAX), None)],
+            ),
+            &required,
+            &fixture.eq_properties,
+            false,
+            false,
+        );
+        Ok(())
+    }
+
     #[test]
     #[expect(
         deprecated,
diff --git a/datafusion/physical-expr/src/scalar_function.rs b/datafusion/physical-expr/src/scalar_function.rs
index f7bd5469cd71e..1d9bc2f3fd55a 100644
--- a/datafusion/physical-expr/src/scalar_function.rs
+++ b/datafusion/physical-expr/src/scalar_function.rs
@@ -158,6 +158,25 @@ impl ScalarFunctionExpr {
         &self.config_options
     }
 
+    /// Returns whether this function expression's argument types have been
+    /// explicitly audited as candidates for range-key locality analysis.
+    ///
+    /// See [`ScalarUDFImpl::supports_range_key_locality_analysis`] for the safety
+    /// requirements.
+    pub(crate) fn supports_range_key_locality_analysis(
+        &self,
+        input_schema: &Schema,
+    ) -> Result<bool> {
+        let argument_types = self
+            .args
+            .iter()
+            .map(|argument| argument.data_type(input_schema))
+            .collect::<Result<Vec<_>>>()?;
+        Ok(self
+            .fun
+            .supports_range_key_locality_analysis(&argument_types))
+    }
+
     /// Given an arbitrary PhysicalExpr attempt to downcast it to a ScalarFunctionExpr
     /// and verify that its inner function is of type T.
     /// If the downcast fails, or the function is not of type T, returns `None`.
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
index 49776d8843d40..168eef838d43a 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
@@ -915,6 +915,20 @@ fn partial_aggregate_output_satisfies_final_partitioning(
         .is_satisfied()
 }
 
+fn partial_aggregate_input_keeps_groups_local(
+    plan: &Arc<dyn ExecutionPlan>,
+    allow_subset_satisfy_partitioning: bool,
+) -> bool {
+    plan.downcast_ref::<AggregateExec>()
+        .is_some_and(|aggregate| {
+            aggregate.mode() == &AggregateMode::Partial
+                && aggregate.input_keeps_groups_local(
+                    aggregate.input().as_ref(),
+                    allow_subset_satisfy_partitioning,
+                )
+        })
+}
+
 /// Adds a [`SortPreservingMergeExec`] or a [`CoalescePartitionsExec`] operator
 /// on top of the given plan node to satisfy a single partition requirement
 /// while preserving ordering constraints.
@@ -1686,10 +1700,13 @@ pub fn ensure_distribution_with_stats(
 
             let preserve_partial_aggregate_partitioning =
                 preserve_file_partition_threshold_met
-                    && partial_aggregate_output_satisfies_final_partitioning(
+                    && (partial_aggregate_output_satisfies_final_partitioning(
+                        &plan,
+                        allow_subset_satisfy_partitioning,
+                    ) || partial_aggregate_input_keeps_groups_local(
                         &plan,
                         allow_subset_satisfy_partitioning,
-                    );
+                    ));
 
             let add_roundrobin = enable_round_robin
                 // Operator benefits from partitioning (e.g. filter):
@@ -1721,6 +1738,9 @@ pub fn ensure_distribution_with_stats(
                 Distribution::HashPartitioned(_) | Distribution::KeyPartitioned(_) => {
                     let child_partitions =
                         child.plan.output_partitioning().partition_count();
+                    // An aggregate can avoid repartitioning if all rows for each group
+                    // are already in one input partition, even when the ordinary
+                    // distribution check fails (#25344). This does not prove join co-partitioning.
                     let partitioning_satisfied = input_distributions
                         .child_satisfaction(
                             child_idx,
@@ -1728,7 +1748,15 @@ pub fn ensure_distribution_with_stats(
                             ChildSatisfactionOptions::new()
                                 .with_allow_subset(allow_subset_satisfy_partitioning),
                         )?
-                        .is_satisfied();
+                        .is_satisfied()
+                        || plan.downcast_ref::<AggregateExec>().is_some_and(
+                            |aggregate| {
+                                aggregate.input_keeps_groups_local(
+                                    child.plan.as_ref(),
+                                    allow_subset_satisfy_partitioning,
+                                )
+                            },
+                        );
                     let preserve_satisfying_file_partitioning =
                         preserve_file_partition_threshold_met
                             && !requires_grouping_id
diff --git a/datafusion/physical-optimizer/src/sanity_checker.rs b/datafusion/physical-optimizer/src/sanity_checker.rs
index 713213b70612d..746746761b864 100644
--- a/datafusion/physical-optimizer/src/sanity_checker.rs
+++ b/datafusion/physical-optimizer/src/sanity_checker.rs
@@ -30,6 +30,7 @@ use datafusion_common::config::{ConfigOptions, OptimizerOptions};
 use datafusion_common::plan_err;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_physical_expr::intervals::utils::{check_support, is_datatype_supported};
+use datafusion_physical_plan::aggregates::AggregateExec;
 use datafusion_physical_plan::execution_plan::{
     Boundedness, EmissionType, InvariantLevel,
 };
@@ -167,14 +168,17 @@ pub fn check_plan_sanity(
             }
         }
 
-        if !input_distributions
-            .child_satisfaction(
-                idx,
-                child.as_ref(),
-                ChildSatisfactionOptions::new().with_allow_subset(true),
-            )?
-            .is_satisfied()
-        {
+        let distribution_satisfaction = input_distributions.child_satisfaction(
+            idx,
+            child.as_ref(),
+            ChildSatisfactionOptions::new().with_allow_subset(true),
+        )?;
+        let accepts_partition_local_groups = plan
+            .downcast_ref::<AggregateExec>()
+            .is_some_and(|aggregate| {
+                aggregate.input_keeps_groups_local(child.as_ref(), true)
+            });
+        if !distribution_satisfaction.is_satisfied() && !accepts_partition_local_groups {
             let plan_str = get_plan_string(plan);
             return plan_err!(
                 "Plan: {:?} does not satisfy distribution requirements: {}. Child-{} output partitioning: {}",
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index 7957c804d5d3b..226ce3d271973 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -1654,6 +1654,52 @@ impl AggregateExec {
         &self.input
     }
 
+    /// Whether a candidate input keeps each group in one partition.
+    ///
+    /// Unlike co-partitioning, aggregation only needs equality groups to be
+    /// local. A final stage can inherit that guarantee from a partial stage
+    /// even when projecting the grouping expressions loses range metadata.
+    /// This check does not require the two stages to be combined.
+    pub fn input_keeps_groups_local(
+        &self,
+        input: &dyn ExecutionPlan,
+        allow_subset: bool,
+    ) -> bool {
+        if !matches!(
+            self.mode,
+            AggregateMode::Partial
+                | AggregateMode::SinglePartitioned
+                | AggregateMode::FinalPartitioned
+        ) || self.group_by().is_empty()
+            || self.group_by().has_grouping_set()
+        {
+            return false;
+        }
+
+        if input.output_partitioning().keeps_keys_local(
+            &self.group_by().input_exprs(),
+            input.equivalence_properties(),
+            allow_subset,
+        ) {
+            return true;
+        }
+
+        self.mode == AggregateMode::FinalPartitioned
+            && input
+                .downcast_ref::<AggregateExec>()
+                .is_some_and(|partial| {
+                    partial.mode == AggregateMode::Partial
+                        && datafusion_physical_expr::physical_exprs_equal(
+                            &partial.output_group_expr(),
+                            &self.group_by().input_exprs(),
+                        )
+                        && partial.input_keeps_groups_local(
+                            partial.input.as_ref(),
+                            allow_subset,
+                        )
+                })
+    }
+
     /// Get the input schema before any aggregates are applied
     pub fn input_schema(&self) -> SchemaRef {
         Arc::clone(&self.input_schema)
diff --git a/datafusion/sqllogictest/src/test_context/range_partitioning.rs b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
index 2cb4f57e29001..4989294146814 100644
--- a/datafusion/sqllogictest/src/test_context/range_partitioning.rs
+++ b/datafusion/sqllogictest/src/test_context/range_partitioning.rs
@@ -423,9 +423,9 @@ type TimeBinRow = (
 /// - partition 1: `[2024-01-01 01:00, 02:00)`
 ///
 /// Files are range-partitioned on `timestamp` and sorted on `(key, timestamp)`.
-/// Because `date_bin(60 seconds, timestamp)` does not straddle the hour split,
-/// grouping by `(key, time_bin)` is partition-disjoint. Today's planner still
-/// inserts a hash shuffle; the test pins that plan so a follow-up can remove it.
+/// The hour split permits partition-local `date_trunc('hour')` aggregation,
+/// but `date_trunc('day')` straddles it. Aligned `date_bin` remains a negative
+/// control because the function is not admitted to range analysis.
 pub(super) fn register_range_sorted_time_bin_table(ctx: &SessionContext) {
     let schema = Arc::new(Schema::new(vec![
         Field::new("key", DataType::Utf8, false),
diff --git a/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt b/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
index 400bbab1a3592..e8c656450ba23 100644
--- a/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
+++ b/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
@@ -30,11 +30,16 @@
 # 2. Two file_groups, so the two 60-minute streams run in parallel
 #
 # Improvement opportunity:
-# date_bin(60s) is monotonic in timestamp and the hour split is aligned to bin
-# boundaries, so (key, time_bin) is partition-disjoint. Aggregation could be a
-# single streaming SinglePartitioned step with no hash shuffle.
+# In this fixture, date_bin(60s) is monotonic in timestamp and the hour split
+# is aligned to bin boundaries, so (key, time_bin) is partition-disjoint.
+# Aggregation could be a single streaming SinglePartitioned step with no
+# hash shuffle.
 #
-# Today's plan still hash-repartitions:
+# This optimization exists for timezone-less date_trunc when every split
+# keeps groups partition-local. Extending it to date_bin is future work
+# requiring a safety audit, including its null behavior (#25344).
+#
+# Today's date_bin plan still hash-repartitions:
 #   Partial AggregateExec (group_clustering_mode=Full)
 #     -> RepartitionExec Hash([key, date_bin(...)])
 #     -> FinalPartitioned AggregateExec (group_clustering_mode=Full)
@@ -82,11 +87,11 @@ physical_plan DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion
 ##########
 # TEST 2: Filtered time-bin aggregation.
 # GROUP BY keys are (key, date_bin(timestamp)). Input is sorted on those keys
-# (date_bin is monotonic in timestamp) and range-partitioned so bins do not
-# overlap across the two 60-minute streams.
-#
-# Today this is still Partial + hash RepartitionExec + Final, even though
-# group_clustering_mode=Full is already recognized.
+# and the split aligns with the bins in this data, so the groups are
+# partition-local. Timezone-less date_trunc can already avoid the shuffle;
+# extending this optimization to date_bin awaits its safety audit (#25344).
+# Until then, Partial + hash RepartitionExec + Final is retained even though
+# group_clustering_mode=Full is recognized.
 ##########
 
 query TT
@@ -117,8 +122,9 @@ k2 2024-01-01T01:30:00 30
 k2 2024-01-01T01:45:00 5
 
 ##########
-# TEST 3: Same aggregation without the col4 filter. The scan still has two
-# 60-minute file groups, and today's plan still hash-repartitions.
+# TEST 3: Same aggregation without the col4 filter. The groups remain
+# partition-local, but date_bin retains the hash shuffle pending its
+# safety audit (#25344).
 ##########
 
 query TT
@@ -146,6 +152,97 @@ k2 2024-01-01T00:30:00 7
 k2 2024-01-01T01:30:00 30
 k2 2024-01-01T01:45:00 5
 
+##########
+# TEST 4: date_trunc('hour') is audited and the split is aligned, so each
+# group is partition-local and aggregation needs no hash shuffle.
+##########
+
+query TT
+EXPLAIN SELECT key, date_trunc('hour', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin;
+----
+physical_plan
+01)ProjectionExec: expr=[key@0 as key, date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)@1 as time_bin, sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
+02)--AggregateExec: mode=SinglePartitioned, gby=[key@0 as key, date_trunc(hour, timestamp@1) as date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
+03)----FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]}, projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC, timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC], [(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a, pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <= col4_max@1, required_guarantees=[col4 in (a)]
+
+query TPI
+SELECT key, date_trunc('hour', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin
+ORDER BY key, time_bin;
+----
+k1 2024-01-01T00:00:00 3
+k1 2024-01-01T01:00:00 30
+k2 2024-01-01T00:00:00 7
+k2 2024-01-01T01:00:00 35
+
+##########
+# TEST 5: date_trunc('day') straddles the hour split, so the hash shuffle
+# remains.
+##########
+
+query TT
+EXPLAIN SELECT key, date_trunc('day', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin;
+----
+physical_plan
+01)ProjectionExec: expr=[key@0 as key, date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)@1 as time_bin, sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)@1 as date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
+03)----RepartitionExec: partitioning=Hash([key@0, date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)@1], 2), input_partitions=2, preserve_order=true, sort_exprs=key@0 ASC, date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)@1 ASC
+04)------AggregateExec: mode=Partial, gby=[key@0 as key, date_trunc(day, timestamp@1) as date_trunc(Utf8("day"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
+05)--------FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
+06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]}, projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC, timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC], [(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a, pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <= col4_max@1, required_guarantees=[col4 in (a)]
+
+query TPI
+SELECT key, date_trunc('day', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin
+ORDER BY key, time_bin;
+----
+k1 2024-01-01T00:00:00 33
+k2 2024-01-01T00:00:00 42
+
+##########
+# TEST 6: Grouping sets can omit the range transform. The per-key subtotal
+# crosses the hour split, so it still needs a hash shuffle.
+##########
+
+query TT
+EXPLAIN SELECT key, date_trunc('hour', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY GROUPING SETS ((key, time_bin), (key));
+----
+physical_plan
+01)ProjectionExec: expr=[key@0 as key, date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)@1 as time_bin, sum(range_sorted_time_bin.value)@3 as sum(range_sorted_time_bin.value)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)@1 as date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp), __grouping_id@2 as __grouping_id], aggr=[sum(range_sorted_time_bin.value)]
+03)----RepartitionExec: partitioning=Hash([key@0, date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)@1, __grouping_id@2], 2), input_partitions=2
+04)------AggregateExec: mode=Partial, gby=[(key@0 as key, date_trunc(hour, timestamp@1) as date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp)), (key@0 as key, NULL as date_trunc(Utf8("hour"),range_sorted_time_bin.timestamp))], aggr=[sum(range_sorted_time_bin.value)]
+05)--------FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
+06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]}, projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC, timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC], [(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a, pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <= col4_max@1, required_guarantees=[col4 in (a)]
+
+query TPI
+SELECT key, date_trunc('hour', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY GROUPING SETS ((key, time_bin), (key))
+ORDER BY key, time_bin NULLS LAST;
+----
+k1 2024-01-01T00:00:00 3
+k1 2024-01-01T01:00:00 30
+k1 NULL 33
+k2 2024-01-01T00:00:00 7
+k2 2024-01-01T01:00:00 35
+k2 NULL 42
+
 ##########
 # CLEANUP
 ##########