feat: implement pmod natively for all numeric types

docs/source/contributor-guide/expression-audits/math_funcs.md

@@ -188,6 +188,11 @@ Internal fused expression that rescales a Decimal128 value (changing scale) and
 
 - Spark 3.4.3, 3.5.8, 4.0.1, 4.1.1 (audited 2026-05-27): `LeafMathExpression(math.Pi, "PI")`; foldable, so Spark `ConstantFolding` rewrites it to a `Literal` before Comet sees the plan. The `CometScalarFunction("pi")` registration is exercised only when `ConstantFolding` is excluded.
 
+## pmod
+
+- Spark 3.4.3, 3.5.8, 4.0.1 (audited 2026-07-24): `Pmod(left, right, evalMode)` signature identical across these versions. `CometPmod` serializes to the `MathExpr pmod` proto (mirroring `Remainder`), carrying the eval mode. The native `spark_pmod` UDF computes `((left % right) + right) % right`; non-ANSI returns NULL on a zero divisor, ANSI raises `DIVIDE_BY_ZERO`. All numeric input types are supported, including decimal (wide decimals use a Decimal256 intermediate, matching modulo).
+- Spark 4.1.1 (audited 2026-07-24): constructor changed to `Pmod(left, right, evalContext: NumericEvalContext)`, but `BinaryArithmetic.evalMode` is still available so no shim is needed. The ANSI zero-divisor error changed to `REMAINDER_BY_ZERO`; Comet's native error is `RemainderByZero`, which matches.
+
 ## positive
 
 - Spark 3.4.3, 3.5.8 (audited 2026-05-27): `UnaryPositive(child)` is a regular expression. There is no Comet serde for `UnaryPositive`, so projections containing `+col` silently disable Comet for the projection on 3.4/3.5.

native/core/src/execution/expressions/arithmetic.rs

@@ -167,7 +167,9 @@ use std::sync::Arc;
 use arrow::datatypes::SchemaRef;
 use datafusion::logical_expr::Operator as DataFusionOperator;
 use datafusion_comet_proto::spark_expression::Expr;
-use datafusion_comet_spark_expr::{create_modulo_expr, create_negate_expr, EvalMode};
+use datafusion_comet_spark_expr::{
+    create_modulo_expr, create_negate_expr, create_pmod_expr, EvalMode,
+};
 
 use crate::execution::{
     expressions::extract_expr,
@@ -251,6 +253,36 @@ impl ExpressionBuilder for RemainderBuilder {
     }
 }
 
+/// Builder for Pmod expressions (uses special positive-modulo function)
+pub struct PmodBuilder;
+
+impl ExpressionBuilder for PmodBuilder {
+    fn build(
+        &self,
+        spark_expr: &Expr,
+        input_schema: SchemaRef,
+        planner: &PhysicalPlanner,
+    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+        let expr = extract_expr!(spark_expr, Pmod);
+        let eval_mode = from_protobuf_eval_mode(expr.eval_mode)?;
+        let left = planner.create_expr(expr.left.as_ref().unwrap(), Arc::clone(&input_schema))?;
+        let right = planner.create_expr(expr.right.as_ref().unwrap(), Arc::clone(&input_schema))?;
+
+        let result = create_pmod_expr(
+            left,
+            right,
+            expr.return_type
+                .as_ref()
+                .map(crate::execution::serde::to_arrow_datatype)
+                .unwrap(),
+            input_schema,
+            eval_mode == EvalMode::Ansi,
+            &planner.session_ctx().state(),
+        );
+        result.map_err(|e| ExecutionError::GeneralError(e.to_string()))
+    }
+}
+
 /// Builder for UnaryMinus expressions (uses special negate function)
 pub struct UnaryMinusBuilder;
 

native/core/src/execution/planner/expression_registry.rs

@@ -47,6 +47,7 @@ pub enum ExpressionType {
     Divide,
     IntegralDivide,
     Remainder,
+    Pmod,
     UnaryMinus,
 
     // Comparison expressions
@@ -212,6 +213,8 @@ impl ExpressionRegistry {
         );
         self.builders
             .insert(ExpressionType::Remainder, Box::new(RemainderBuilder));
+        self.builders
+            .insert(ExpressionType::Pmod, Box::new(PmodBuilder));
         self.builders
             .insert(ExpressionType::UnaryMinus, Box::new(UnaryMinusBuilder));
     }
@@ -325,6 +328,7 @@ impl ExpressionRegistry {
             Some(ExprStruct::Divide(_)) => Ok(ExpressionType::Divide),
             Some(ExprStruct::IntegralDivide(_)) => Ok(ExpressionType::IntegralDivide),
             Some(ExprStruct::Remainder(_)) => Ok(ExpressionType::Remainder),
+            Some(ExprStruct::Pmod(_)) => Ok(ExpressionType::Pmod),
             Some(ExprStruct::UnaryMinus(_)) => Ok(ExpressionType::UnaryMinus),
 
             Some(ExprStruct::Eq(_)) => Ok(ExpressionType::Eq),

native/proto/src/proto/expr.proto

@@ -92,6 +92,7 @@ message Expr {
     JvmScalarUdf jvm_scalar_udf = 70;
     PreciseTimestampConversion precise_timestamp_conversion = 71;
     Shuffle shuffle = 72;
+    MathExpr pmod = 73;
   }
 
   reserved 20;

native/spark-expr/src/comet_scalar_funcs.rs

@@ -21,7 +21,7 @@ use crate::map_funcs::spark_map_sort;
 use crate::math_funcs::abs::abs;
 use crate::math_funcs::checked_arithmetic::{checked_add, checked_div, checked_mul, checked_sub};
 use crate::math_funcs::log::spark_log;
-use crate::math_funcs::modulo_expr::spark_modulo;
+use crate::math_funcs::modulo_expr::{spark_modulo, spark_pmod};
 use crate::{
     spark_ceil, spark_day_name, spark_decimal_div, spark_decimal_integral_div, spark_floor,
     spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding,
@@ -185,6 +185,9 @@ pub fn create_comet_physical_fun_with_eval_mode(
             let func = Arc::new(spark_modulo);
             make_comet_scalar_udf!("spark_modulo", func, without data_type, fail_on_error)
         }
+        "spark_pmod" => {
+            make_comet_scalar_udf!("spark_pmod", spark_pmod, data_type, fail_on_error)
+        }
         "abs" => {
             let func = Arc::new(abs);
             make_comet_scalar_udf!("abs", func, without data_type)

native/spark-expr/src/lib.rs

@@ -85,8 +85,8 @@ pub use hash_funcs::*;
 pub use json_funcs::{FromJson, ToJson};
 pub use math_funcs::{
     checked_add, checked_div, checked_mul, checked_sub, create_modulo_expr, create_negate_expr,
-    spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_log,
-    spark_make_decimal, spark_round, spark_unhex, spark_unscaled_value, CheckOverflow,
+    create_pmod_expr, spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor,
+    spark_log, spark_make_decimal, spark_round, spark_unhex, spark_unscaled_value, CheckOverflow,
     DecimalRescaleCheckOverflow, NegativeExpr, NormalizeNaNAndZero, WideDecimalBinaryExpr,
     WideDecimalOp,
 };

native/spark-expr/src/math_funcs/mod.rs

@@ -36,7 +36,7 @@ pub use div::spark_decimal_integral_div;
 pub use floor::spark_floor;
 pub use internal::*;
 pub use log::spark_log;
-pub use modulo_expr::create_modulo_expr;
+pub use modulo_expr::{create_modulo_expr, create_pmod_expr};
 pub use negative::{create_negate_expr, NegativeExpr};
 pub use round::spark_round;
 pub use unhex::spark_unhex;