diff --git a/crates/sail-plan/src/function/scalar/math.rs b/crates/sail-plan/src/function/scalar/math.rs
index fb0bdf9ac6..75bfff11d4 100644
--- a/crates/sail-plan/src/function/scalar/math.rs
+++ b/crates/sail-plan/src/function/scalar/math.rs
@@ -75,9 +75,14 @@ fn spark_plus(input: ScalarFunctionInput) -> PlanResult<Expr> {
             (Ok(DataType::Date32), Ok(right_type)) if right_type.is_numeric() => {
                 cast(cast(left, DataType::Int32) + right, DataType::Date32)
             }
+            (Ok(left_type), Ok(right_type)) => {
+                let (left, right) =
+                    coerce_spark_arithmetic_operands(left, right, &left_type, &right_type);
+                left + right
+            }
             // TODO: In case getting the type fails, we don't want to fail the query.
             //  Future work is needed here, ideally we create something like `Operator::SparkPlus`.
-            (Ok(_), Ok(_)) | (Err(_), _) | (_, Err(_)) => left + right,
+            (Err(_), _) | (_, Err(_)) => left + right,
         })
     }
 }
@@ -126,9 +131,14 @@ fn spark_minus(input: ScalarFunctionInput) -> PlanResult<Expr> {
             (Ok(DataType::Date32), Ok(right_type)) if right_type.is_numeric() => {
                 cast(cast(left, DataType::Int32) - right, DataType::Date32)
             }
+            (Ok(left_type), Ok(right_type)) => {
+                let (left, right) =
+                    coerce_spark_arithmetic_operands(left, right, &left_type, &right_type);
+                left - right
+            }
             // TODO: In case getting the type fails, we don't want to fail the query.
             //  Future work is needed here, ideally we create something like `Operator::SparkMinus`.
-            (Ok(_), Ok(_)) | (Err(_), _) | (_, Err(_)) => left - right,
+            (Err(_), _) | (_, Err(_)) => left - right,
         })
     }
 }
@@ -176,12 +186,111 @@ fn spark_multiply(input: ScalarFunctionInput) -> PlanResult<Expr> {
                 DataType::Duration(TimeUnit::Microsecond),
             )
         }
+        (Ok(left_type), Ok(right_type)) => {
+            let (left, right) =
+                coerce_spark_arithmetic_operands(left, right, &left_type, &right_type);
+            left * right
+        }
         // TODO: In case getting the type fails, we don't want to fail the query.
         //  Future work is needed here, ideally we create something like `Operator::SparkMultiply`.
-        (Ok(_), Ok(_)) | (Err(_), _) | (_, Err(_)) => left * right,
+        (Err(_), _) | (_, Err(_)) => left * right,
     })
 }
 
+/// Spark-specific operand coercion for `+ - *` applied at plan-construction time,
+/// so the logical plan is valid by construction (rather than relying on a later
+/// analyzer rule, which would run after `ExprSchemable::get_type` has already typed
+/// the binary op via DataFusion's `BinaryTypeCoercer`).
+///
+/// Covers the cases where DataFusion's default coercion diverges from Spark:
+///   - FLOAT/DOUBLE combined with DECIMAL: Spark promotes both to DOUBLE.
+///   - integer LITERAL combined with DECIMAL: Spark narrows the literal to its
+///     minimal-precision decimal (so `dec(10,2) * 3` => `decimal(12,2)`).
+fn coerce_spark_arithmetic_operands(
+    left: Expr,
+    right: Expr,
+    left_type: &DataType,
+    right_type: &DataType,
+) -> (Expr, Expr) {
+    // FLOAT/DOUBLE x DECIMAL -> DOUBLE.
+    // https://github.com/apache/spark/blob/v4.1.1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
+    if is_float_decimal_pair(left_type, right_type) {
+        return (
+            cast(left, DataType::Float64),
+            cast(right, DataType::Float64),
+        );
+    }
+    // integer literal x DECIMAL -> narrow the literal to its minimal decimal.
+    // https://github.com/apache/spark/blob/v4.1.1/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DecimalType.scala
+    let left = match spark_decimal_literal_datatype(&left, right_type) {
+        Some(target) => cast(left, target),
+        None => left,
+    };
+    let right = match spark_decimal_literal_datatype(&right, left_type) {
+        Some(target) => cast(right, target),
+        None => right,
+    };
+    (left, right)
+}
+
+/// True when one operand is a floating-point type and the other a decimal (either
+/// order) — the pair Spark promotes to `DoubleType` in arithmetic.
+fn is_float_decimal_pair(a: &DataType, b: &DataType) -> bool {
+    fn is_decimal(dt: &DataType) -> bool {
+        matches!(dt, DataType::Decimal128(_, _) | DataType::Decimal256(_, _))
+    }
+    (a.is_floating() && is_decimal(b)) || (is_decimal(a) && b.is_floating())
+}
+
+/// When `expr` is an integer literal and `other_type` is a decimal, returns the
+/// literal's minimal-precision decimal (`Decimal(digit_count, 0)`), matching Spark's
+/// `DecimalType.fromLiteral`. DataFusion would instead widen it to `Decimal(10, 0)`.
+fn spark_decimal_literal_datatype(expr: &Expr, other_type: &DataType) -> Option<DataType> {
+    let is_256 = match other_type {
+        DataType::Decimal128(_, _) => false,
+        DataType::Decimal256(_, _) => true,
+        _ => return None,
+    };
+    let value = match expr {
+        Expr::Literal(scalar, _) => scalar_integer_value(scalar)?,
+        _ => return None,
+    };
+    let precision = integer_digit_count(value);
+    Some(if is_256 {
+        DataType::Decimal256(precision, 0)
+    } else {
+        DataType::Decimal128(precision, 0)
+    })
+}
+
+fn scalar_integer_value(scalar: &ScalarValue) -> Option<i128> {
+    Some(match scalar {
+        ScalarValue::Int8(Some(v)) => i128::from(*v),
+        ScalarValue::Int16(Some(v)) => i128::from(*v),
+        ScalarValue::Int32(Some(v)) => i128::from(*v),
+        ScalarValue::Int64(Some(v)) => i128::from(*v),
+        ScalarValue::UInt8(Some(v)) => i128::from(*v),
+        ScalarValue::UInt16(Some(v)) => i128::from(*v),
+        ScalarValue::UInt32(Some(v)) => i128::from(*v),
+        ScalarValue::UInt64(Some(v)) => i128::from(*v),
+        _ => return None,
+    })
+}
+
+/// Number of base-10 digits in `value` (sign ignored), minimum 1.
+fn integer_digit_count(value: i128) -> u8 {
+    let mut n = value.unsigned_abs();
+    let mut digits = 0u8;
+    loop {
+        digits += 1;
+        n /= 10;
+        if n == 0 {
+            break;
+        }
+    }
+    digits
+}
+
 /// Check if an expression represents a zero literal value.
 /// Handles both direct literals and CAST expressions wrapping literals.
 fn is_zero_literal(expr: &Expr) -> bool {
@@ -734,3 +843,73 @@ pub(super) fn list_built_in_math_functions() -> Vec<(&'static str, ScalarFunctio
         ("width_bucket", F::quaternary(math_fn::width_bucket)),
     ]
 }
+
+#[cfg(test)]
+mod tests {
+    use datafusion_expr::col;
+
+    use super::*;
+
+    #[test]
+    fn integer_digit_count_counts_digits_sign_ignored() {
+        assert_eq!(integer_digit_count(0), 1);
+        assert_eq!(integer_digit_count(3), 1);
+        assert_eq!(integer_digit_count(-3), 1);
+        assert_eq!(integer_digit_count(100), 3);
+        assert_eq!(integer_digit_count(-12345), 5);
+    }
+
+    #[test]
+    fn spark_decimal_literal_datatype_narrows_only_integer_literal_vs_decimal() {
+        let dec = DataType::Decimal128(10, 2);
+        // integer literal against a decimal -> minimal decimal
+        assert_eq!(
+            spark_decimal_literal_datatype(&lit(3_i32), &dec),
+            Some(DataType::Decimal128(1, 0))
+        );
+        assert_eq!(
+            spark_decimal_literal_datatype(&lit(100_i64), &dec),
+            Some(DataType::Decimal128(3, 0))
+        );
+        // other side not a decimal -> no narrowing
+        assert_eq!(
+            spark_decimal_literal_datatype(&lit(3_i32), &DataType::Int32),
+            None
+        );
+        // not an integer literal -> no narrowing
+        assert_eq!(spark_decimal_literal_datatype(&lit(2.0_f64), &dec), None);
+        assert_eq!(spark_decimal_literal_datatype(&col("x"), &dec), None);
+    }
+
+    #[test]
+    fn is_float_decimal_pair_detects_either_order() {
+        let dec = DataType::Decimal128(10, 2);
+        assert!(is_float_decimal_pair(&DataType::Float32, &dec));
+        assert!(is_float_decimal_pair(&dec, &DataType::Float64));
+        assert!(!is_float_decimal_pair(&DataType::Int32, &dec));
+        assert!(!is_float_decimal_pair(
+            &DataType::Float32,
+            &DataType::Float64
+        ));
+    }
+
+    #[test]
+    fn coerce_narrows_integer_literal_against_decimal() {
+        // `dec(10,2) * 3` -> literal narrowed to Decimal(1,0) (=> result decimal(12,2)).
+        let dec = DataType::Decimal128(10, 2);
+        let (left, right) =
+            coerce_spark_arithmetic_operands(col("a"), lit(3_i32), &dec, &DataType::Int32);
+        assert_eq!(left.to_string(), "a");
+        assert_eq!(right.to_string(), "CAST(Int32(3) AS Decimal128(1, 0))");
+    }
+
+    #[test]
+    fn coerce_promotes_float_decimal_to_double() {
+        // `dec(10,2) * float` -> both promoted to Double.
+        let dec = DataType::Decimal128(10, 2);
+        let (left, right) =
+            coerce_spark_arithmetic_operands(col("a"), lit(2.0_f32), &dec, &DataType::Float32);
+        assert_eq!(left.to_string(), "CAST(a AS Float64)");
+        assert_eq!(right.to_string(), "CAST(Float32(2) AS Float64)");
+    }
+}
diff --git a/python/pysail/tests/spark/__snapshots__/test_tpcds.plan.yaml b/python/pysail/tests/spark/__snapshots__/test_tpcds.plan.yaml
index 5a414a05c6..e10e523d7e 100644
--- a/python/pysail/tests/spark/__snapshots__/test_tpcds.plan.yaml
+++ b/python/pysail/tests/spark/__snapshots__/test_tpcds.plan.yaml
@@ -733,7 +733,7 @@ data:
     ProjectionExec: expr=[#85@0 as i_item_id, #86@1 as i_item_desc, #87@2 as i_category, #88@3 as i_class, #89@4 as i_current_price, #90@5 as itemrevenue, #91@6 as revenueratio]
       SortPreservingMergeExec: [#87@2 ASC, #88@3 ASC, #85@0 ASC, #86@1 ASC, #91@6 ASC], fetch=100
         SortExec: TopK(fetch=100), expr=[#87@2 ASC, #88@3 ASC, #85@0 ASC, #86@1 ASC, #91@6 ASC], preserve_partitioning=[true]
-          ProjectionExec: expr=[#35@0 as #85, #38@1 as #86, #46@2 as #87, #44@3 as #88, #39@4 as #89, sum(web_sales.#23)@5 as #90, sum(web_sales.#23)@5 * Some(100),10,0 / CASE WHEN sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #91]
+          ProjectionExec: expr=[#35@0 as #85, #38@1 as #86, #46@2 as #87, #44@3 as #88, #39@4 as #89, sum(web_sales.#23)@5 as #90, sum(web_sales.#23)@5 * Some(100),3,0 / CASE WHEN sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #91]
             WindowAggExec: wdw=[sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(sum(web_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Decimal128(27, 2), nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
               SortExec: expr=[#44@3 ASC NULLS LAST], preserve_partitioning=[true]
                 RepartitionExec: partitioning=Hash([#44@3], 4), input_partitions=4
@@ -1540,7 +1540,7 @@ data:
     ProjectionExec: expr=[#85@0 as i_item_id, #86@1 as i_item_desc, #87@2 as i_category, #88@3 as i_class, #89@4 as i_current_price, #90@5 as itemrevenue, #91@6 as revenueratio]
       SortPreservingMergeExec: [#87@2 ASC, #88@3 ASC, #85@0 ASC, #86@1 ASC, #91@6 ASC], fetch=100
         SortExec: TopK(fetch=100), expr=[#87@2 ASC, #88@3 ASC, #85@0 ASC, #86@1 ASC, #91@6 ASC], preserve_partitioning=[true]
-          ProjectionExec: expr=[#35@0 as #85, #38@1 as #86, #46@2 as #87, #44@3 as #88, #39@4 as #89, sum(catalog_sales.#23)@5 as #90, sum(catalog_sales.#23)@5 * Some(100),10,0 / CASE WHEN sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #91]
+          ProjectionExec: expr=[#35@0 as #85, #38@1 as #86, #46@2 as #87, #44@3 as #88, #39@4 as #89, sum(catalog_sales.#23)@5 as #90, sum(catalog_sales.#23)@5 * Some(100),3,0 / CASE WHEN sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #91]
             WindowAggExec: wdw=[sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(sum(catalog_sales.#23)) PARTITION BY [item.#44] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Decimal128(27, 2), nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
               SortExec: expr=[#44@3 ASC NULLS LAST], preserve_partitioning=[true]
                 RepartitionExec: partitioning=Hash([#44@3], 4), input_partitions=4
@@ -3802,7 +3802,7 @@ data:
       ScalarSubqueryExec: subqueries=3
         SortPreservingMergeExec: [#421@0 ASC, #422@1 ASC], fetch=100
           SortExec: TopK(fetch=100), expr=[#421@0 ASC, #422@1 ASC], preserve_partitioning=[true]
-            ProjectionExec: expr=[#131@2 as #421, #132@3 as #422, #132@3 / __common_expr_4@0 * Some(100),10,0 as #423, #276@4 as #424, #276@4 / __common_expr_4@0 * Some(100),10,0 as #425, #420@5 as #426, #420@5 / __common_expr_4@0 * Some(100),10,0 as #427, __common_expr_5@1 as #428]
+            ProjectionExec: expr=[#131@2 as #421, #132@3 as #422, #132@3 / __common_expr_4@0 * Some(100),3,0 as #423, #276@4 as #424, #276@4 / __common_expr_4@0 * Some(100),3,0 as #425, #420@5 as #426, #420@5 / __common_expr_4@0 * Some(100),3,0 as #427, __common_expr_5@1 as #428]
               ProjectionExec: expr=[CASE WHEN __common_expr_6@0 = Some(0),23,6 THEN CAST(raise_error(Division by zero) AS Decimal128(23, 6)) ELSE __common_expr_6@0 END as __common_expr_4, __common_expr_6@0 as __common_expr_5, #131@1 as #131, #132@2 as #132, #276@3 as #276, #420@4 as #420]
                 ProjectionExec: expr=[(#132@0 + #276@1 + #420@2) / Some(3),10,0 as __common_expr_6, #131@3 as #131, #132@0 as #132, #276@1 as #276, #420@2 as #420]
                   HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#131@0, #419@0)], filter=CAST(#132@0 AS Decimal128(19, 3)) >= Some(9),1,1 * #420@2 AND CAST(#132@0 AS Decimal128(20, 3)) <= Some(11),2,1 * #420@2 AND CAST(#276@1 AS Decimal128(19, 3)) >= Some(9),1,1 * #420@2 AND CAST(#276@1 AS Decimal128(20, 3)) <= Some(11),2,1 * #420@2 AND CAST(#420@2 AS Decimal128(19, 3)) >= Some(9),1,1 * #132@0 AND CAST(#420@2 AS Decimal128(20, 3)) <= Some(11),2,1 * #132@0 AND CAST(#420@2 AS Decimal128(19, 3)) >= Some(9),1,1 * #276@1 AND CAST(#420@2 AS Decimal128(20, 3)) <= Some(11),2,1 * #276@1, projection=[#132@1, #276@2, #420@4, #131@0]
@@ -4039,7 +4039,7 @@ data:
     == Physical Plan ==
     ProjectionExec: expr=[#287@0 as promotions, #288@1 as total, #289@2 as ((promotions / total) * 100)]
       SortExec: TopK(fetch=100), expr=[#288@1 ASC], preserve_partitioning=[false]
-        ProjectionExec: expr=[#152@0 as #287, #286@1 as #288, CAST(#152@0 AS Decimal128(15, 4)) / CASE WHEN #286@1 = Some(0),17,2 THEN CAST(raise_error(Division by zero) AS Decimal128(15, 4)) ELSE CAST(#286@1 AS Decimal128(15, 4)) END * Some(100),10,0 as #289]
+        ProjectionExec: expr=[#152@0 as #287, #286@1 as #288, CAST(#152@0 AS Decimal128(15, 4)) / CASE WHEN #286@1 = Some(0),17,2 THEN CAST(raise_error(Division by zero) AS Decimal128(15, 4)) ELSE CAST(#286@1 AS Decimal128(15, 4)) END * Some(100),3,0 as #289]
           CrossJoinExec
             ProjectionExec: expr=[sum(store_sales.#15)@0 as #152]
               AggregateExec: mode=Final, gby=[], aggr=[sum(store_sales.#15)]
@@ -4228,7 +4228,7 @@ data:
                                                                         HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#61@0, #66@1)], projection=[#64@1, #66@2, #67@3, #68@4, #69@5, #70@6, #71@7, #72@8, #75@9, #76@10, #83@11]
                                                                           CoalescePartitionsExec
                                                                             ProjectionExec: expr=[#15@0 as #61]
-                                                                              FilterExec: CAST(sum(catalog_sales.#25)@1 AS Decimal128(30, 2)) > Some(2),10,0 * sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)@2, projection=[#15@0]
+                                                                              FilterExec: CAST(sum(catalog_sales.#25)@1 AS Decimal128(21, 2)) > Some(2),1,0 * sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)@2, projection=[#15@0]
                                                                                 AggregateExec: mode=FinalPartitioned, gby=[#15@0 as #15], aggr=[sum(catalog_sales.#25), sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)]
                                                                                   RepartitionExec: partitioning=Hash([#15@0], 4), input_partitions=4
                                                                                     AggregateExec: mode=Partial, gby=[#15@0 as #15], aggr=[sum(catalog_sales.#25), sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)]
@@ -4316,7 +4316,7 @@ data:
                                                                       HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#61@0, #66@1)], projection=[#64@1, #66@2, #67@3, #68@4, #69@5, #70@6, #71@7, #72@8, #75@9, #76@10, #83@11]
                                                                         CoalescePartitionsExec
                                                                           ProjectionExec: expr=[#15@0 as #61]
-                                                                            FilterExec: CAST(sum(catalog_sales.#25)@1 AS Decimal128(30, 2)) > Some(2),10,0 * sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)@2, projection=[#15@0]
+                                                                            FilterExec: CAST(sum(catalog_sales.#25)@1 AS Decimal128(21, 2)) > Some(2),1,0 * sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)@2, projection=[#15@0]
                                                                               AggregateExec: mode=FinalPartitioned, gby=[#15@0 as #15], aggr=[sum(catalog_sales.#25), sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)]
                                                                                 RepartitionExec: partitioning=Hash([#15@0], 4), input_partitions=4
                                                                                   AggregateExec: mode=Partial, gby=[#15@0 as #15], aggr=[sum(catalog_sales.#25), sum(catalog_returns.#57 + catalog_returns.#58 + catalog_returns.#59)]
@@ -6378,7 +6378,7 @@ data:
     ProjectionExec: expr=[#74@0 as i_item_id, #75@1 as i_item_desc, #76@2 as i_category, #77@3 as i_class, #78@4 as i_current_price, #79@5 as itemrevenue, #80@6 as revenueratio]
       SortPreservingMergeExec: [#76@2 ASC, #77@3 ASC, #74@0 ASC, #75@1 ASC, #80@6 ASC]
         SortExec: expr=[#76@2 ASC, #77@3 ASC, #74@0 ASC, #75@1 ASC, #80@6 ASC], preserve_partitioning=[true]
-          ProjectionExec: expr=[#24@0 as #74, #27@1 as #75, #35@2 as #76, #33@3 as #77, #28@4 as #78, sum(store_sales.#15)@5 as #79, sum(store_sales.#15)@5 * Some(100),10,0 / CASE WHEN sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #80]
+          ProjectionExec: expr=[#24@0 as #74, #27@1 as #75, #35@2 as #76, #33@3 as #77, #28@4 as #78, sum(store_sales.#15)@5 as #79, sum(store_sales.#15)@5 * Some(100),3,0 / CASE WHEN sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 = Some(0),27,2 THEN CAST(raise_error(Division by zero) AS Decimal128(27, 2)) ELSE sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 END as #80]
             WindowAggExec: wdw=[sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(sum(store_sales.#15)) PARTITION BY [item.#33] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Decimal128(27, 2), nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
               SortExec: expr=[#33@3 ASC NULLS LAST], preserve_partitioning=[true]
                 RepartitionExec: partitioning=Hash([#33@3], 4), input_partitions=4
diff --git a/python/pysail/tests/spark/__snapshots__/test_tpch.plan.yaml b/python/pysail/tests/spark/__snapshots__/test_tpch.plan.yaml
index af8f87b25b..95a3d15dfc 100644
--- a/python/pysail/tests/spark/__snapshots__/test_tpch.plan.yaml
+++ b/python/pysail/tests/spark/__snapshots__/test_tpch.plan.yaml
@@ -8,10 +8,10 @@ data:
       SortPreservingMergeExec: [#16@0 ASC, #17@1 ASC]
         SortExec: expr=[#16@0 ASC, #17@1 ASC], preserve_partitioning=[true]
           ProjectionExec: expr=[#8@0 as #16, #9@1 as #17, sum(lineitem.#4)@2 as #18, sum(lineitem.#5)@3 as #19, sum(lineitem.#5 * Int32(1) - lineitem.#6)@4 as #20, sum(lineitem.#5 * Int32(1) - lineitem.#6 * Int32(1) + lineitem.#7)@5 as #21, avg(lineitem.#4)@6 as #22, avg(lineitem.#5)@7 as #23, avg(lineitem.#6)@8 as #24, count(Int64(1))@9 as #25]
-            AggregateExec: mode=FinalPartitioned, gby=[#8@0 as #8, #9@1 as #9], aggr=[sum(lineitem.#4), sum(lineitem.#5), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6), sum(__common_expr_1 * Some(1),10,0 + lineitem.#7) as sum(lineitem.#5 * Int32(1) - lineitem.#6 * Int32(1) + lineitem.#7), avg(lineitem.#4), avg(lineitem.#5), avg(lineitem.#6), count(Int64(1))]
+            AggregateExec: mode=FinalPartitioned, gby=[#8@0 as #8, #9@1 as #9], aggr=[sum(lineitem.#4), sum(lineitem.#5), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6), sum(__common_expr_1 * Some(1),1,0 + lineitem.#7) as sum(lineitem.#5 * Int32(1) - lineitem.#6 * Int32(1) + lineitem.#7), avg(lineitem.#4), avg(lineitem.#5), avg(lineitem.#6), count(Int64(1))]
               RepartitionExec: partitioning=Hash([#8@0, #9@1], 4), input_partitions=4
-                AggregateExec: mode=Partial, gby=[#8@5 as #8, #9@6 as #9], aggr=[sum(lineitem.#4), sum(lineitem.#5), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6), sum(__common_expr_1 * Some(1),10,0 + lineitem.#7) as sum(lineitem.#5 * Int32(1) - lineitem.#6 * Int32(1) + lineitem.#7), avg(lineitem.#4), avg(lineitem.#5), avg(lineitem.#6), count(Int64(1))]
-                  ProjectionExec: expr=[#21@1 * (Some(1),10,0 - #22@2) as __common_expr_1, #20@0 as #4, #21@1 as #5, #22@2 as #6, #23@3 as #7, #24@4 as #8, #25@5 as #9]
+                AggregateExec: mode=Partial, gby=[#8@5 as #8, #9@6 as #9], aggr=[sum(lineitem.#4), sum(lineitem.#5), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6), sum(__common_expr_1 * Some(1),1,0 + lineitem.#7) as sum(lineitem.#5 * Int32(1) - lineitem.#6 * Int32(1) + lineitem.#7), avg(lineitem.#4), avg(lineitem.#5), avg(lineitem.#6), count(Int64(1))]
+                  ProjectionExec: expr=[#21@1 * (Some(1),1,0 - #22@2) as __common_expr_1, #20@0 as #4, #21@1 as #5, #22@2 as #6, #23@3 as #7, #24@4 as #8, #25@5 as #9]
                     RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
                       ProjectionExec: expr=[_4@0 as #20, _5@1 as #21, _6@2 as #22, _7@3 as #23, _8@4 as #24, _9@5 as #25]
                         FilterExec: _10@6 <= 1998-09-24, projection=[_4@0, _5@1, _6@2, _7@3, _8@4, _9@5]
@@ -72,9 +72,9 @@ data:
       SortPreservingMergeExec: [#34@1 DESC NULLS LAST, #35@2 ASC], fetch=10
         SortExec: TopK(fetch=10), expr=[#34@1 DESC NULLS LAST, #35@2 ASC], preserve_partitioning=[true]
           ProjectionExec: expr=[#17@0 as #33, sum(lineitem.#22 * Int32(1) - lineitem.#23)@3 as #34, #12@1 as #35, #15@2 as #36]
-            AggregateExec: mode=FinalPartitioned, gby=[#17@0 as #17, #12@1 as #12, #15@2 as #15], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+            AggregateExec: mode=FinalPartitioned, gby=[#17@0 as #17, #12@1 as #12, #15@2 as #15], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
               RepartitionExec: partitioning=Hash([#17@0, #12@1, #15@2], 4), input_partitions=4
-                AggregateExec: mode=Partial, gby=[#17@2 as #17, #12@0 as #12, #15@1 as #15], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+                AggregateExec: mode=Partial, gby=[#17@2 as #17, #12@0 as #12, #15@1 as #15], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
                   HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#8@0, #17@0)], projection=[#12@1, #15@2, #17@3, #22@4, #23@5]
                     CoalescePartitionsExec
                       HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#0@0, #9@1)], projection=[#8@1, #12@3, #15@4]
@@ -118,9 +118,9 @@ data:
       SortPreservingMergeExec: [#48@1 DESC NULLS LAST]
         SortExec: expr=[#48@1 DESC NULLS LAST], preserve_partitioning=[true]
           ProjectionExec: expr=[#41@0 as #47, sum(lineitem.#22 * Int32(1) - lineitem.#23)@1 as #48]
-            AggregateExec: mode=FinalPartitioned, gby=[#41@0 as #41], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+            AggregateExec: mode=FinalPartitioned, gby=[#41@0 as #41], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
               RepartitionExec: partitioning=Hash([#41@0], 4), input_partitions=4
-                AggregateExec: mode=Partial, gby=[#41@2 as #41], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+                AggregateExec: mode=Partial, gby=[#41@2 as #41], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
                   HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#44@0, #42@3)], projection=[#22@1, #23@2, #41@3]
                     ProjectionExec: expr=[_0@0 as #44]
                       FilterExec: _1@1 = AFRICA, projection=[_0@0]
@@ -171,7 +171,7 @@ data:
             AggregateExec: mode=FinalPartitioned, gby=[#48@0 as #48, #49@1 as #49, #50@2 as #50], aggr=[sum(shipping.#51)]
               RepartitionExec: partitioning=Hash([#48@0, #49@1, #50@2], 4), input_partitions=4
                 AggregateExec: mode=Partial, gby=[#48@0 as #48, #49@1 as #49, #50@2 as #50], aggr=[sum(shipping.#51)]
-                  ProjectionExec: expr=[#41@0 as #48, #45@1 as #49, date_part(year, #17@2) as #50, #12@3 * (Some(1),10,0 - #13@4) as #51]
+                  ProjectionExec: expr=[#41@0 as #48, #45@1 as #49, date_part(year, #17@2) as #50, #12@3 * (Some(1),1,0 - #13@4) as #51]
                     HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#35@3, #44@0)], filter=#41@0 = GERMANY AND #45@1 = IRAQ OR #41@0 = IRAQ AND #45@1 = GERMANY, projection=[#41@4, #45@6, #17@2, #12@0, #13@1]
                       CoalescePartitionsExec
                         HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#3@0, #40@0)], projection=[#12@1, #13@2, #17@3, #35@4, #41@6]
@@ -205,11 +205,11 @@ data:
     ProjectionExec: expr=[#63@0 as o_year, #64@1 as mkt_share]
       SortPreservingMergeExec: [#63@0 ASC]
         SortExec: expr=[#63@0 ASC], preserve_partitioning=[true]
-          ProjectionExec: expr=[#60@0 as #63, sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END)@1 / CASE WHEN sum(all_nations.#61)@2 = Some(0),31,4 THEN CAST(raise_error(Division by zero) AS Decimal128(31, 4)) ELSE sum(all_nations.#61)@2 END as #64]
-            AggregateExec: mode=FinalPartitioned, gby=[#60@0 as #60], aggr=[sum(CASE WHEN all_nations.#62 = IRAQ THEN all_nations.#61 ELSE Some(0),21,4 END) as sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END), sum(all_nations.#61)]
+          ProjectionExec: expr=[#60@0 as #63, sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END)@1 / CASE WHEN sum(all_nations.#61)@2 = Some(0),22,4 THEN CAST(raise_error(Division by zero) AS Decimal128(22, 4)) ELSE sum(all_nations.#61)@2 END as #64]
+            AggregateExec: mode=FinalPartitioned, gby=[#60@0 as #60], aggr=[sum(CASE WHEN all_nations.#62 = IRAQ THEN all_nations.#61 ELSE Some(0),14,4 END) as sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END), sum(all_nations.#61)]
               RepartitionExec: partitioning=Hash([#60@0], 4), input_partitions=4
-                AggregateExec: mode=Partial, gby=[#60@0 as #60], aggr=[sum(CASE WHEN all_nations.#62 = IRAQ THEN all_nations.#61 ELSE Some(0),21,4 END) as sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END), sum(all_nations.#61)]
-                  ProjectionExec: expr=[date_part(year, #36@0) as #60, #21@1 * (Some(1),10,0 - #22@2) as #61, #54@3 as #62]
+                AggregateExec: mode=Partial, gby=[#60@0 as #60], aggr=[sum(CASE WHEN all_nations.#62 = IRAQ THEN all_nations.#61 ELSE Some(0),14,4 END) as sum(CASE WHEN all_nations.#62 = Utf8("IRAQ") THEN all_nations.#61 WHEN Boolean(true) THEN Int32(0) END), sum(all_nations.#61)]
+                  ProjectionExec: expr=[date_part(year, #36@0) as #60, #21@1 * (Some(1),1,0 - #22@2) as #61, #54@3 as #62]
                     HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#57@0, #51@3)], projection=[#36@3, #21@1, #22@2, #54@5]
                       ProjectionExec: expr=[_0@0 as #57]
                         FilterExec: _1@1 = MIDDLE EAST, projection=[_0@0]
@@ -255,7 +255,7 @@ data:
             AggregateExec: mode=FinalPartitioned, gby=[#50@0 as #50, #51@1 as #51], aggr=[sum(profit.#52)]
               RepartitionExec: partitioning=Hash([#50@0, #51@1], 4), input_partitions=4
                 AggregateExec: mode=Partial, gby=[#50@0 as #50, #51@1 as #51], aggr=[sum(profit.#52)]
-                  ProjectionExec: expr=[#47@7 as #50, date_part(year, #41@5) as #51, #21@1 * (Some(1),10,0 - #22@2) - #35@4 * #20@0 as #52]
+                  ProjectionExec: expr=[#47@7 as #50, date_part(year, #41@5) as #51, #21@1 * (Some(1),1,0 - #22@2) - #35@4 * #20@0 as #52]
                     RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
                       HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#12@3, #46@0)]
                         HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#16@0, #37@0)], projection=[#20@1, #21@2, #22@3, #12@4, #35@5, #41@7]
@@ -286,9 +286,9 @@ data:
       SortPreservingMergeExec: [#39@2 DESC NULLS LAST], fetch=20
         SortExec: TopK(fetch=20), expr=[#39@2 DESC NULLS LAST], preserve_partitioning=[true]
           ProjectionExec: expr=[#0@0 as #37, #1@1 as #38, sum(lineitem.#22 * Int32(1) - lineitem.#23)@7 as #39, #5@2 as #40, #34@4 as #41, #2@5 as #42, #4@3 as #43, #7@6 as #44]
-            AggregateExec: mode=FinalPartitioned, gby=[#0@0 as #0, #1@1 as #1, #5@2 as #5, #4@3 as #4, #34@4 as #34, #2@5 as #2, #7@6 as #7], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+            AggregateExec: mode=FinalPartitioned, gby=[#0@0 as #0, #1@1 as #1, #5@2 as #5, #4@3 as #4, #34@4 as #34, #2@5 as #2, #7@6 as #7], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
               RepartitionExec: partitioning=Hash([#0@0, #1@1, #5@2, #4@3, #34@4, #2@5, #7@6], 4), input_partitions=4
-                AggregateExec: mode=Partial, gby=[#0@0 as #0, #1@1 as #1, #5@4 as #5, #4@3 as #4, #34@8 as #34, #2@2 as #2, #7@5 as #7], aggr=[sum(lineitem.#22 * Some(1),10,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
+                AggregateExec: mode=Partial, gby=[#0@0 as #0, #1@1 as #1, #5@4 as #5, #4@3 as #4, #34@8 as #34, #2@2 as #2, #7@5 as #7], aggr=[sum(lineitem.#22 * Some(1),1,0 - lineitem.#23) as sum(lineitem.#22 * Int32(1) - lineitem.#23)]
                   HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#33@0, #3@3)], projection=[#0@2, #1@3, #2@4, #4@6, #5@7, #7@8, #22@9, #23@10, #34@1]
                     ProjectionExec: expr=[_0@0 as #33, _1@1 as #34]
                       DataSourceExec: partitions=1, partition_sizes=[1]
@@ -389,11 +389,11 @@ data:
   |-
     == Physical Plan ==
     ProjectionExec: expr=[#25@0 as promo_revenue]
-      ProjectionExec: expr=[Some(10000),5,2 * sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END)@0 / CASE WHEN sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 = Some(0),31,4 THEN CAST(raise_error(Division by zero) AS Decimal128(31, 4)) ELSE sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 END as #25]
-        AggregateExec: mode=Final, gby=[], aggr=[sum(CASE WHEN part.#20 LIKE PROMO% THEN __common_expr_1 ELSE Some(0),21,4 END) as sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+      ProjectionExec: expr=[Some(10000),5,2 * sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END)@0 / CASE WHEN sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 = Some(0),22,4 THEN CAST(raise_error(Division by zero) AS Decimal128(22, 4)) ELSE sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 END as #25]
+        AggregateExec: mode=Final, gby=[], aggr=[sum(CASE WHEN part.#20 LIKE PROMO% THEN __common_expr_1 ELSE Some(0),14,4 END) as sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
           CoalescePartitionsExec
-            AggregateExec: mode=Partial, gby=[], aggr=[sum(CASE WHEN part.#20 LIKE PROMO% THEN __common_expr_1 ELSE Some(0),21,4 END) as sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
-              ProjectionExec: expr=[#5@0 * (Some(1),10,0 - #6@1) as __common_expr_1, #20@2 as #20]
+            AggregateExec: mode=Partial, gby=[], aggr=[sum(CASE WHEN part.#20 LIKE PROMO% THEN __common_expr_1 ELSE Some(0),14,4 END) as sum(CASE WHEN part.#20 LIKE Utf8("PROMO%") THEN lineitem.#5 * Int32(1) - lineitem.#6 WHEN Boolean(true) THEN Int32(0) END), sum(__common_expr_1) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+              ProjectionExec: expr=[#5@0 * (Some(1),1,0 - #6@1) as __common_expr_1, #20@2 as #20]
                 HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#16@0, #1@0)], projection=[#5@3, #6@4, #20@1]
                   ProjectionExec: expr=[_0@0 as #16, _4@1 as #20]
                     DataSourceExec: partitions=1, partition_sizes=[1]
@@ -424,9 +424,9 @@ data:
                   DataSourceExec: partitions=1, partition_sizes=[1]
                 ProjectionExec: expr=[#2@0 as #7, sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 as #8]
                   FilterExec: sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 = scalar_subquery(<pending>)
-                    AggregateExec: mode=FinalPartitioned, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+                    AggregateExec: mode=FinalPartitioned, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
                       RepartitionExec: partitioning=Hash([#2@0], 4), input_partitions=4
-                        AggregateExec: mode=Partial, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+                        AggregateExec: mode=Partial, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
                           RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
                             ProjectionExec: expr=[_2@0 as #2, _5@1 as #5, _6@2 as #6]
                               FilterExec: _10@3 >= 1996-08-01 AND _10@3 < 1996-11-01, projection=[_2@0, _5@1, _6@2]
@@ -436,9 +436,9 @@ data:
             CoalescePartitionsExec
               AggregateExec: mode=Partial, gby=[], aggr=[max(revenue0.#10)]
                 ProjectionExec: expr=[sum(lineitem.#5 * Int32(1) - lineitem.#6)@1 as #10]
-                  AggregateExec: mode=FinalPartitioned, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+                  AggregateExec: mode=FinalPartitioned, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
                     RepartitionExec: partitioning=Hash([#2@0], 4), input_partitions=4
-                      AggregateExec: mode=Partial, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+                      AggregateExec: mode=Partial, gby=[#2@0 as #2], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
                         RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
                           ProjectionExec: expr=[_2@0 as #2, _5@1 as #5, _6@2 as #6]
                             FilterExec: _10@3 >= 1996-08-01 AND _10@3 < 1996-11-01, projection=[_2@0, _5@1, _6@2]
@@ -537,9 +537,9 @@ data:
     == Physical Plan ==
     ProjectionExec: expr=[#25@0 as revenue]
       ProjectionExec: expr=[sum(lineitem.#5 * Int32(1) - lineitem.#6)@0 as #25]
-        AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+        AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
           CoalescePartitionsExec
-            AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.#5 * Some(1),10,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
+            AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.#5 * Some(1),1,0 - lineitem.#6) as sum(lineitem.#5 * Int32(1) - lineitem.#6)]
               HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#16@0, #1@0)], filter=#19@1 = Brand#21 AND #22@3 IN (SET) ([SM CASE, SM BOX, SM PACK, SM PKG]) AND #4@0 >= Some(800),4,2 AND #4@0 <= Some(1800),4,2 AND #21@2 <= 5 OR #19@1 = Brand#13 AND #22@3 IN (SET) ([MED BAG, MED BOX, MED PKG, MED PACK]) AND #4@0 >= Some(2000),4,2 AND #4@0 <= Some(3000),4,2 AND #21@2 <= 10 OR #19@1 = Brand#52 AND #22@3 IN (SET) ([LG CASE, LG BOX, LG PACK, LG PKG]) AND #4@0 >= Some(3000),4,2 AND #4@0 <= Some(4000),4,2 AND #21@2 <= 15, projection=[#5@6, #6@7]
                 ProjectionExec: expr=[_0@0 as #16, _3@1 as #19, _5@2 as #21, _6@3 as #22]
                   FilterExec: _5@2 >= 1 AND (_3@1 = Brand#21 AND _6@3 IN (SET) ([SM CASE, SM BOX, SM PACK, SM PKG]) AND _5@2 <= 5 OR _3@1 = Brand#13 AND _6@3 IN (SET) ([MED BAG, MED BOX, MED PKG, MED PACK]) AND _5@2 <= 10 OR _3@1 = Brand#52 AND _6@3 IN (SET) ([LG CASE, LG BOX, LG PACK, LG PKG]) AND _5@2 <= 15)
diff --git a/python/pysail/tests/spark/function/features/arithmetic_coercion.feature b/python/pysail/tests/spark/function/features/arithmetic_coercion.feature
new file mode 100644
index 0000000000..9b928341b4
--- /dev/null
+++ b/python/pysail/tests/spark/function/features/arithmetic_coercion.feature
@@ -0,0 +1,79 @@
+@arithmetic_coercion
+Feature: Spark type coercion for the +, -, * operators
+
+  # DataFusion's BinaryTypeCoercer does not perform these Spark coercions; Sail
+  # applies them in the arithmetic plan builders (math.rs) so the *result type*
+  # matches Spark 4.1.1. Values and precision/scale are asserted below via typeof.
+  #
+  # Known remaining gap (see the @sail-bug scenario): Spark marks decimal
+  # arithmetic nullable=true (overflow may yield NULL) even for non-null operands,
+  # while a native BinaryExpr inherits its operands' nullability (=> false).
+  # Matching Spark's nullability needs the custom PhysicalExpr follow-up.
+
+  Rule: Decimal with an integer literal narrows the literal to minimal precision
+    Scenario: decimal times a single-digit integer literal
+      When query
+        """
+        SELECT typeof(CAST(2.5 AS DECIMAL(10,2)) * 3) AS t,
+               CAST(2.5 AS DECIMAL(10,2)) * 3 AS r
+        """
+      Then query result
+        | t             | r    |
+        | decimal(12,2) | 7.50 |
+
+    Scenario: decimal plus a single-digit integer literal
+      When query
+        """
+        SELECT typeof(CAST(2.5 AS DECIMAL(10,2)) + 3) AS t,
+               CAST(2.5 AS DECIMAL(10,2)) + 3 AS r
+        """
+      Then query result
+        | t             | r    |
+        | decimal(11,2) | 5.50 |
+
+    Scenario: decimal minus a single-digit integer literal
+      When query
+        """
+        SELECT typeof(CAST(5.5 AS DECIMAL(10,2)) - 2) AS t,
+               CAST(5.5 AS DECIMAL(10,2)) - 2 AS r
+        """
+      Then query result
+        | t             | r    |
+        | decimal(11,2) | 3.50 |
+
+    Scenario: decimal times a three-digit integer literal
+      When query
+        """
+        SELECT typeof(CAST(2.5 AS DECIMAL(10,2)) * 100) AS t,
+               CAST(2.5 AS DECIMAL(10,2)) * 100 AS r
+        """
+      Then query result
+        | t             | r      |
+        | decimal(14,2) | 250.00 |
+
+  Rule: Float or double combined with a decimal promotes to double
+    Scenario: float times decimal returns double
+      When query
+        """
+        SELECT typeof(CAST(1.5 AS FLOAT) * CAST(2.0 AS DECIMAL(10,2))) AS t,
+               CAST(1.5 AS FLOAT) * CAST(2.0 AS DECIMAL(10,2)) AS r
+        """
+      Then query result
+        | t      | r   |
+        | double | 3.0 |
+
+  Rule: Decimal arithmetic is nullable in Spark (known gap — needs custom PhysicalExpr)
+    # Spark marks decimal +, -, * as nullable=true even for non-null operands,
+    # because the operation can overflow to NULL. A native BinaryExpr built in the
+    # plan builder inherits nullability from its operands, so Sail reports false.
+    @sail-bug
+    Scenario: decimal arithmetic reports nullable=true like Spark
+      When query
+        """
+        SELECT CAST(2.5 AS DECIMAL(10,2)) * 3 AS result
+        """
+      Then query schema
+        """
+        root
+         |-- result: decimal(12,2) (nullable = true)
+        """