Also try in-place mutation on the right operand

When the left operand's buffer cannot be reused (shared reference),
try the right operand for in-place mutation. This covers cases like
Scalar-Array and Array-Array where the right side has refcount 1.

For non-commutative ops (sub, rem), the argument order is preserved
correctly: result[i] = op(left[i], right[i]).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/physical-expr-common/src/datum.rs

@@ -87,14 +87,21 @@ pub fn apply_arithmetic(
     let f = arithmetic_op_to_fn(op);
     match (lhs, rhs) {
         (ColumnarValue::Array(left), ColumnarValue::Array(right)) => {
-            // Try in-place on left array with right array values
+            // Try in-place on left array, then right array
             match try_apply_inplace_array(left, &right, op) {
                 Ok(result) => Ok(ColumnarValue::Array(result)),
-                Err(left) => Ok(ColumnarValue::Array(f(&left, &right)?)),
+                Err(left) => match try_apply_inplace_array_rhs(&left, right, op) {
+                    Ok(result) => Ok(ColumnarValue::Array(result)),
+                    Err(right) => Ok(ColumnarValue::Array(f(&left, &right)?)),
+                },
             }
         }
         (ColumnarValue::Scalar(left), ColumnarValue::Array(right)) => {
-            Ok(ColumnarValue::Array(f(&left.to_scalar()?, &right)?))
+            // Try in-place on right array with scalar left (flipped)
+            match try_apply_inplace_scalar_rhs(right, &left, op) {
+                Ok(result) => Ok(ColumnarValue::Array(result)),
+                Err(right) => Ok(ColumnarValue::Array(f(&left.to_scalar()?, &right)?)),
+            }
         }
         (ColumnarValue::Array(left), ColumnarValue::Scalar(right)) => {
             // Try in-place on left array with scalar right
@@ -308,6 +315,188 @@ where
     }
 }
 
+/// Try to apply arithmetic in-place on `right` array with a scalar `left`.
+/// The operation is `result[i] = op(scalar_left, right[i])`, stored in `right`'s buffer.
+/// Returns `Ok(result)` on success, or `Err(right)` if in-place not possible.
+fn try_apply_inplace_scalar_rhs(
+    right: ArrayRef,
+    left: &ScalarValue,
+    op: ArithmeticOp,
+) -> Result<ArrayRef, ArrayRef> {
+    if left.is_null() {
+        return Err(right);
+    }
+    macro_rules! dispatch_inplace_scalar_rhs {
+        ($($arrow_type:ident),*) => {
+            match right.data_type() {
+                $(
+                    dt if dt == &<$arrow_type as ArrowPrimitiveType>::DATA_TYPE => {
+                        let scalar_val = left
+                            .to_scalar()
+                            .map_err(|_| Arc::clone(&right))?;
+                        let scalar_arr = scalar_val.get().0;
+                        let lhs_val = scalar_arr
+                            .as_any()
+                            .downcast_ref::<PrimitiveArray<$arrow_type>>()
+                            .ok_or_else(|| Arc::clone(&right))?
+                            .value(0);
+                        try_inplace_unary_rhs::<$arrow_type>(right, lhs_val, op)
+                    }
+                )*
+                _ => Err(right),
+            }
+        };
+    }
+    dispatch_inplace_scalar_rhs!(
+        Int8Type,
+        Int16Type,
+        Int32Type,
+        Int64Type,
+        UInt8Type,
+        UInt16Type,
+        UInt32Type,
+        UInt64Type,
+        Float16Type,
+        Float32Type,
+        Float64Type
+    )
+}
+
+/// Try to apply arithmetic in-place on `right` array using values from `left` array.
+/// The operation is `result[i] = op(left[i], right[i])`, stored in `right`'s buffer.
+/// Returns `Ok(result)` on success, or `Err(right)` if in-place not possible.
+fn try_apply_inplace_array_rhs(
+    left: &ArrayRef,
+    right: ArrayRef,
+    op: ArithmeticOp,
+) -> Result<ArrayRef, ArrayRef> {
+    if left.data_type() != right.data_type() {
+        return Err(right);
+    }
+    macro_rules! dispatch_inplace_array_rhs {
+        ($($arrow_type:ident),*) => {
+            match right.data_type() {
+                $(
+                    dt if dt == &<$arrow_type as ArrowPrimitiveType>::DATA_TYPE => {
+                        try_inplace_binary_rhs::<$arrow_type>(left, right, op)
+                    }
+                )*
+                _ => Err(right),
+            }
+        };
+    }
+    dispatch_inplace_array_rhs!(
+        Int8Type,
+        Int16Type,
+        Int32Type,
+        Int64Type,
+        UInt8Type,
+        UInt16Type,
+        UInt32Type,
+        UInt64Type,
+        Float16Type,
+        Float32Type,
+        Float64Type
+    )
+}
+
+/// Attempt in-place mutation on the right PrimitiveArray: result[i] = op(scalar, right[i]).
+fn try_inplace_unary_rhs<T: ArrowPrimitiveType>(
+    array: ArrayRef,
+    scalar: T::Native,
+    op: ArithmeticOp,
+) -> Result<ArrayRef, ArrayRef>
+where
+    T::Native: ArrowNativeTypeOp,
+{
+    let primitive = array
+        .as_any()
+        .downcast_ref::<PrimitiveArray<T>>()
+        .ok_or_else(|| Arc::clone(&array))?
+        .clone();
+    drop(array);
+
+    // For right-side mutation: result = op(scalar, element)
+    // Commutative ops: same as op(element, scalar)
+    // Non-commutative: need reversed argument order
+    type BinFn<N> = fn(N, N) -> N;
+    let op_fn: Option<BinFn<T::Native>> = match op {
+        ArithmeticOp::AddWrapping => Some(ArrowNativeTypeOp::add_wrapping),
+        ArithmeticOp::MulWrapping => Some(ArrowNativeTypeOp::mul_wrapping),
+        ArithmeticOp::SubWrapping => Some(ArrowNativeTypeOp::sub_wrapping),
+        ArithmeticOp::Rem if !scalar.is_zero() => Some(ArrowNativeTypeOp::mod_wrapping),
+        _ => None,
+    };
+
+    let Some(op_fn) = op_fn else {
+        return Err(Arc::new(primitive));
+    };
+
+    // Note: op(scalar, v) — scalar is the left operand
+    match primitive.unary_mut(|v| op_fn(scalar, v)) {
+        Ok(result) => Ok(Arc::new(result)),
+        Err(arr) => Err(Arc::new(arr)),
+    }
+}
+
+/// Attempt in-place mutation on the right PrimitiveArray: result[i] = op(left[i], right[i]).
+fn try_inplace_binary_rhs<T: ArrowPrimitiveType>(
+    left: &ArrayRef,
+    right: ArrayRef,
+    op: ArithmeticOp,
+) -> Result<ArrayRef, ArrayRef>
+where
+    T::Native: ArrowNativeTypeOp,
+{
+    let left_primitive = left
+        .as_any()
+        .downcast_ref::<PrimitiveArray<T>>()
+        .ok_or_else(|| Arc::clone(&right))?;
+
+    let right_primitive = right
+        .as_any()
+        .downcast_ref::<PrimitiveArray<T>>()
+        .ok_or_else(|| Arc::clone(&right))?
+        .clone();
+    drop(right);
+
+    let mut builder = match right_primitive.into_builder() {
+        Ok(b) => b,
+        Err(arr) => return Err(Arc::new(arr)),
+    };
+
+    type BinFn<N> = fn(N, N) -> N;
+    let op_fn: Option<BinFn<T::Native>> = match op {
+        ArithmeticOp::AddWrapping => Some(ArrowNativeTypeOp::add_wrapping),
+        ArithmeticOp::SubWrapping => Some(ArrowNativeTypeOp::sub_wrapping),
+        ArithmeticOp::MulWrapping => Some(ArrowNativeTypeOp::mul_wrapping),
+        _ => None,
+    };
+
+    let Some(op_fn) = op_fn else {
+        return Err(Arc::new(builder.finish()));
+    };
+
+    let right_slice = builder.values_slice_mut();
+    let left_values = left_primitive.values();
+
+    // Note: op(left[i], right[i]) — left is the first operand
+    right_slice
+        .iter_mut()
+        .zip(left_values.iter())
+        .for_each(|(r, l)| *r = op_fn(*l, *r));
+
+    // Merge null buffers from both sides
+    let result = builder.finish();
+    if left_primitive.nulls().is_some() {
+        let merged = NullBuffer::union(result.nulls(), left_primitive.nulls());
+        let result = PrimitiveArray::<T>::new(result.values().clone(), merged);
+        Ok(Arc::new(result))
+    } else {
+        Ok(Arc::new(result))
+    }
+}
+
 /// Applies a binary [`Datum`] comparison kernel `f` to `lhs` and `rhs`
 pub fn apply_cmp(
     lhs: &ColumnarValue,