[SPARK-61154][CONNECT][PYTHON] Raise VALUE_NOT_BETWEEN for a literal of an int beyond the range of a long, in the Python client as PySpark does

crates/pyspark-rs/src/functions.rs

@@ -205,6 +205,21 @@ pub(crate) fn enum_to_value<'py>(obj: &Bound<'py, PyAny>) -> PyResult<Bound<'py,
     Ok(value)
 }
 
+/// The error of a Python int beyond the range of a long, as PySpark's `LiteralExpression`
+/// raises it: `VALUE_NOT_BETWEEN`, whose maximum is `JVM_SHORT_MAX`, as PySpark writes it.
+pub(crate) fn value_not_between(py: Python<'_>) -> PyErr {
+    crate::column::raise_pyspark(
+        py,
+        "PySparkValueError",
+        "VALUE_NOT_BETWEEN",
+        &[
+            ("arg_name", "value"),
+            ("min", "-9223372036854775808"),
+            ("max", "32767"),
+        ],
+    )
+}
+
 /// Convert a Python object to a Rust Column (handles Column or scalars that become literals).
 pub fn to_column(obj: &Bound<'_, PyAny>) -> PyResult<Column> {
     // Try to get as PyColumn first
@@ -271,6 +286,10 @@ pub fn to_column(obj: &Bound<'_, PyAny>) -> PyResult<Column> {
             },
         )));
     }
+    // An int beyond a long is not a double, as PySpark's literal of it: it is an error.
+    if obj.is_instance_of::<pyo3::types::PyInt>() {
+        return Err(value_not_between(obj.py()));
+    }
 
     if let Ok(f) = obj.extract::<f64>() {
         return Ok(Column::new(Expression::literal(LiteralExpression::double(

crates/pyspark-rs/src/session.rs

@@ -2725,6 +2725,10 @@ pub(crate) fn py_to_value(obj: &Bound<'_, PyAny>) -> PyResult<Value> {
     if let Ok(i) = obj.extract::<i64>() {
         return Ok(Value::Long(i));
     }
+    // An int beyond a long is not a double, as PySpark's literal of it: it is an error.
+    if obj.is_instance_of::<pyo3::types::PyInt>() {
+        return Err(crate::functions::value_not_between(obj.py()));
+    }
 
     // Check for float
     if let Ok(f) = obj.extract::<f64>() {

crates/pyspark-rs/src/stat_arguments.rs

@@ -157,14 +157,28 @@ pub(crate) fn fillna(
     if let Some(dict) = dict {
         let mut pairs = Vec::with_capacity(dict.len());
         for (column, fill) in dict.iter() {
-            pairs.push((column.extract()?, crate::session::py_to_value(&fill)?));
+            pairs.push((column.extract()?, fill_value(&fill)?));
         }
         return Ok(PyDataFrame::new(df.fillna_map(pairs)));
     }
-    Ok(PyDataFrame::new(df.fillna_value(
-        crate::session::py_to_value(value)?,
-        column_refs(&columns),
-    )))
+    Ok(PyDataFrame::new(
+        df.fillna_value(fill_value(value)?, column_refs(&columns)),
+    ))
+}
+
+/// The value of a fill. As PySpark sends it as a proto literal, an int beyond a long is out of
+/// range for the proto, which raises `ValueError`, not the error of a literal.
+fn fill_value(value: &Bound<'_, PyAny>) -> PyResult<spark_connect::row::Value> {
+    if value.is_instance_of::<PyInt>()
+        && !value.is_instance_of::<pyo3::types::PyBool>()
+        && value.extract::<i64>().is_err()
+    {
+        return Err(pyo3::exceptions::PyValueError::new_err(format!(
+            "Value out of range: {}",
+            value.str()?
+        )));
+    }
+    crate::session::py_to_value(value)
 }
 
 /// The columns of `df.describe(*cols)`, as PySpark's: column names, or a list of them.

python/tests/test_dropin_offline.py

@@ -8827,6 +8827,40 @@ def completion_analyzed():
         child.close()
 
 
+def test_int_literals_beyond_a_long_raise_value_not_between_as_pyspark():
+    # As PySpark's LiteralExpression._infer_type: an int beyond the range of a long is an error,
+    # not a double, which would make `col == 2**63` match the rows of the longs next to it. The
+    # maximum is JVM_SHORT_MAX, as PySpark writes it. fillna sends the value as a proto literal,
+    # which raises ValueError for it.
+    from pyspark.errors import PySparkValueError
+    from pyspark.sql import functions as F
+
+    spark = _offline_session()
+    df = spark.range(1)
+    for case in (
+        lambda: F.lit(2**70),
+        lambda: F.lit(-(2**63) - 1),
+        lambda: F.col("id") == 2**63,
+        lambda: F.col("id") > 2**70,
+        lambda: F.col("id").isin([1, 2**70]),
+        lambda: F.col("id").between(1, 2**70),
+        lambda: F.when(F.col("id") > 0, 2**70),
+        lambda: F.array(F.lit(1), F.lit(2**70)),
+        lambda: F.lit([2**70]),
+    ):
+        with pytest.raises(PySparkValueError) as e:
+            case()
+        assert e.value.getCondition() == "VALUE_NOT_BETWEEN"
+        assert e.value.getMessageParameters() == {
+            "arg_name": "value", "min": "-9223372036854775808", "max": "32767"}
+    with pytest.raises(ValueError, match="Value out of range: 1180591620717411303424"):
+        df.fillna(2**70)
+    # The bounds of a long, and a float beyond them, are fine.
+    assert F.lit(2**63 - 1) is not None and F.lit(-(2**63)) is not None
+    assert F.lit(2.0**70) is not None
+    spark.stop()
+
+
 @pytest.mark.parametrize("names", [
     ["z", "a"],
     ["with space", "a.b", "with`tick", "", "123", "class", "雪", "_private"],