feat(spark): make pyspark-optional params optional

Match pyspark's optional-parameter surface in the spark namespace:
- make_dt_interval, make_interval: all parts default to zero (int32 0 / lit 0.0)
- str_to_map: pair_delim defaults to ',', key_value_delim defaults to ':'
- round: scale defaults to 0 (HALF_UP rounding to nearest integer)
- shuffle: accepts `seed` kwarg for pyspark parity; raises NotImplementedError
  for non-None values until the Rust binding supports it
- like, ilike: accept `escapeChar` for pyspark parity; same NotImplementedError
  guard; first positional renamed `string` → `str` to match pyspark

ceil/floor `scale=` deferred — the underlying Rust expr_fn is single-arg.

Added a module-level `_ZERO_I32` literal to avoid rebuilding the pyarrow
int32 zero scalar on every call.

Tests: positional-compat coverage for aggregates (`spark.avg(col)` etc.),
defaults-omitted cases for the optional-arg functions, and
NotImplementedError cases for `shuffle(seed=)` and `like/ilike(escapeChar=)`.

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

Author: timsaucer

python/datafusion/functions/spark.py

@@ -32,6 +32,8 @@
 
 from typing import TYPE_CHECKING, Any
 
+import pyarrow as pa
+
 from datafusion._internal import functions as _functions
 from datafusion.expr import Expr, sort_list_to_raw_sort_list
 
@@ -41,6 +43,9 @@
 
 _f = _functions.spark
 
+# Reused int32 literal so optional-arg defaults don't rebuild it per call.
+_ZERO_I32 = Expr.literal(pa.scalar(0, type=pa.int32()))
+
 
 def _filter_raw(filter: Expr | None) -> Any:
     return filter.expr if filter is not None else None
@@ -203,9 +208,12 @@ def array(*cols: Expr) -> Expr:
     return Expr(_f.array(*[c.expr for c in cols]))
 
 
-def shuffle(col: Expr) -> Expr:
+def shuffle(col: Expr, seed: int | None = None) -> Expr:
     """Spark ``shuffle``: returns a random permutation of the input array.
 
+    ``seed`` is accepted for pyspark parity but is not yet wired through the
+    Rust binding; passing a non-``None`` value raises ``NotImplementedError``.
+
     Examples:
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
@@ -217,6 +225,9 @@ def shuffle(col: Expr) -> Expr:
         >>> sorted(r.collect_column("v")[0].as_py())
         [1, 2, 3]
     """
+    if seed is not None:
+        msg = "shuffle(seed=...) is not yet supported by the Spark UDF binding"
+        raise NotImplementedError(msg)
     return Expr(_f.shuffle(col.expr))
 
 
@@ -589,59 +600,78 @@ def last_day(col: Expr) -> Expr:
     return Expr(_f.last_day(col.expr))
 
 
-def make_dt_interval(days: Expr, hours: Expr, mins: Expr, secs: Expr) -> Expr:
+def make_dt_interval(
+    days: Expr | None = None,
+    hours: Expr | None = None,
+    mins: Expr | None = None,
+    secs: Expr | None = None,
+) -> Expr:
     """Spark ``make_dt_interval``: day-time interval from components.
 
+    All parts are optional; omitted parts default to zero, matching pyspark.
+
     Examples:
-        >>> import pyarrow as pa
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
+        >>> r = df.select(dfn.functions.spark.make_dt_interval().alias("v"))
+        >>> r.collect_column("v")[0].as_py()
+        datetime.timedelta(0)
+
+        >>> import pyarrow as pa
         >>> i32 = lambda n: dfn.lit(pa.scalar(n, type=pa.int32()))
         >>> r = df.select(
         ...     dfn.functions.spark.make_dt_interval(
-        ...         i32(1), i32(2), i32(3), dfn.lit(4.5)
+        ...         days=i32(1), hours=i32(2), mins=i32(3), secs=dfn.lit(4.5)
         ...     ).alias("v")
         ... )
         >>> r.collect_column("v")[0].as_py()
         datetime.timedelta(days=1, seconds=7384, microseconds=500000)
     """
-    return Expr(_f.make_dt_interval(days.expr, hours.expr, mins.expr, secs.expr))
+    return Expr(
+        _f.make_dt_interval(
+            (days if days is not None else _ZERO_I32).expr,
+            (hours if hours is not None else _ZERO_I32).expr,
+            (mins if mins is not None else _ZERO_I32).expr,
+            (secs if secs is not None else Expr.literal(0.0)).expr,
+        )
+    )
 
 
 def make_interval(
-    years: Expr,
-    months: Expr,
-    weeks: Expr,
-    days: Expr,
-    hours: Expr,
-    mins: Expr,
-    secs: Expr,
+    years: Expr | None = None,
+    months: Expr | None = None,
+    weeks: Expr | None = None,
+    days: Expr | None = None,
+    hours: Expr | None = None,
+    mins: Expr | None = None,
+    secs: Expr | None = None,
 ) -> Expr:
     """Spark ``make_interval``: interval from year/month/week/day/hour/min/sec parts.
 
+    All parts are optional; omitted parts default to zero, matching pyspark.
+
     Examples:
-        >>> import pyarrow as pa
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
+        >>> r = df.select(dfn.functions.spark.make_interval().alias("v"))
+        >>> r.collect_column("v")[0].as_py().months
+        0
+
+        >>> import pyarrow as pa
         >>> i32 = lambda n: dfn.lit(pa.scalar(n, type=pa.int32()))
-        >>> r = df.select(
-        ...     dfn.functions.spark.make_interval(
-        ...         i32(1), i32(0), i32(0), i32(0),
-        ...         i32(0), i32(0), dfn.lit(0.0)
-        ...     ).alias("v")
-        ... )
+        >>> r = df.select(dfn.functions.spark.make_interval(years=i32(1)).alias("v"))
         >>> r.collect_column("v")[0].as_py().months
         12
     """
     return Expr(
         _f.make_interval(
-            years.expr,
-            months.expr,
-            weeks.expr,
-            days.expr,
-            hours.expr,
-            mins.expr,
-            secs.expr,
+            (years if years is not None else _ZERO_I32).expr,
+            (months if months is not None else _ZERO_I32).expr,
+            (weeks if weeks is not None else _ZERO_I32).expr,
+            (days if days is not None else _ZERO_I32).expr,
+            (hours if hours is not None else _ZERO_I32).expr,
+            (mins if mins is not None else _ZERO_I32).expr,
+            (secs if secs is not None else Expr.literal(0.0)).expr,
         )
     )
 
@@ -984,21 +1014,36 @@ def map_from_entries(col: Expr) -> Expr:
     return Expr(_f.map_from_entries(col.expr))
 
 
-def str_to_map(text: Expr, pair_delim: Expr, key_value_delim: Expr) -> Expr:
+def str_to_map(
+    text: Expr,
+    pair_delim: Expr | None = None,
+    key_value_delim: Expr | None = None,
+) -> Expr:
     """Spark ``str_to_map``: split text into key/value pairs using delimiters.
 
+    Delimiters default to ``","`` and ``":"`` when omitted, matching pyspark.
+
     Examples:
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
+        >>> r = df.select(
+        ...     dfn.functions.spark.str_to_map(dfn.lit("a:1,b:2")).alias("v"))
+        >>> r.collect_column("v")[0].as_py()
+        [('a', '1'), ('b', '2')]
+
         >>> r = df.select(
         ...     dfn.functions.spark.str_to_map(
-        ...         dfn.lit("a:1,b:2"), dfn.lit(","), dfn.lit(":")
+        ...         dfn.lit("a=1;b=2"),
+        ...         pair_delim=dfn.lit(";"),
+        ...         key_value_delim=dfn.lit("="),
         ...     ).alias("v")
         ... )
         >>> r.collect_column("v")[0].as_py()
         [('a', '1'), ('b', '2')]
     """
-    return Expr(_f.str_to_map(text.expr, pair_delim.expr, key_value_delim.expr))
+    pd = pair_delim if pair_delim is not None else Expr.literal(",")
+    kvd = key_value_delim if key_value_delim is not None else Expr.literal(":")
+    return Expr(_f.str_to_map(text.expr, pd.expr, kvd.expr))
 
 
 # ---------------------------------------------------------------------------
@@ -1130,18 +1175,28 @@ def rint(col: Expr) -> Expr:
     return Expr(_f.rint(col.expr))
 
 
-def round(col: Expr, scale: Expr) -> Expr:
+def round(col: Expr, scale: Expr | None = None) -> Expr:
     """Spark ``round``: round to ``scale`` decimal places, HALF_UP rounding.
 
+    ``scale`` defaults to zero when omitted, matching pyspark.
+
     Examples:
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
-        >>> r = df.select(
-        ...     dfn.functions.spark.round(dfn.lit(2.5), dfn.lit(0)).alias("v"))
+        >>> r = df.select(dfn.functions.spark.round(dfn.lit(2.5)).alias("v"))
         >>> r.collect_column("v")[0].as_py()
         3.0
+
+        >>> r = df.select(
+        ...     dfn.functions.spark.round(
+        ...         dfn.lit(2.345), scale=dfn.lit(2)
+        ...     ).alias("v")
+        ... )
+        >>> r.collect_column("v")[0].as_py()
+        2.35
     """
-    return Expr(_f.round(col.expr, scale.expr))
+    scale_expr = scale if scale is not None else _ZERO_I32
+    return Expr(_f.round(col.expr, scale_expr.expr))
 
 
 def unhex(col: Expr) -> Expr:
@@ -1306,9 +1361,16 @@ def elt(*inputs: Expr) -> Expr:
     return Expr(_f.elt(*[i.expr for i in inputs]))
 
 
-def ilike(string: Expr, pattern: Expr) -> Expr:
+def ilike(
+    str: Expr,
+    pattern: Expr,
+    escapeChar: str | None = None,  # noqa: N803
+) -> Expr:
     """Spark ``ilike``: case-insensitive pattern match.
 
+    ``escapeChar`` is accepted for pyspark parity but is not yet wired through
+    the Rust binding; passing a non-``None`` value raises ``NotImplementedError``.
+
     Examples:
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
@@ -1317,7 +1379,10 @@ def ilike(string: Expr, pattern: Expr) -> Expr:
         >>> r.collect_column("v")[0].as_py()
         True
     """
-    return Expr(_f.ilike(string.expr, pattern.expr))
+    if escapeChar is not None:
+        msg = "ilike(escapeChar=...) is not yet supported by the Spark UDF binding"
+        raise NotImplementedError(msg)
+    return Expr(_f.ilike(str.expr, pattern.expr))
 
 
 def length(col: Expr) -> Expr:
@@ -1333,9 +1398,16 @@ def length(col: Expr) -> Expr:
     return Expr(_f.length(col.expr))
 
 
-def like(string: Expr, pattern: Expr) -> Expr:
+def like(
+    str: Expr,
+    pattern: Expr,
+    escapeChar: str | None = None,  # noqa: N803
+) -> Expr:
     """Spark ``like``: case-sensitive pattern match.
 
+    ``escapeChar`` is accepted for pyspark parity but is not yet wired through
+    the Rust binding; passing a non-``None`` value raises ``NotImplementedError``.
+
     Examples:
         >>> ctx = dfn.SessionContext()
         >>> df = ctx.from_pydict({"x": [1]})
@@ -1344,7 +1416,10 @@ def like(string: Expr, pattern: Expr) -> Expr:
         >>> r.collect_column("v")[0].as_py()
         True
     """
-    return Expr(_f.like(string.expr, pattern.expr))
+    if escapeChar is not None:
+        msg = "like(escapeChar=...) is not yet supported by the Spark UDF binding"
+        raise NotImplementedError(msg)
+    return Expr(_f.like(str.expr, pattern.expr))
 
 
 def luhn_check(col: Expr) -> Expr:

python/tests/test_spark_functions.py

@@ -229,6 +229,73 @@ def test_round_half_up():
     assert _val(df, spark.round(lit(2.5), lit(0))) == 3.0
 
 
+# ---------------------------------------------------------------------------
+# Optional parameter defaults / NotImplementedError
+# ---------------------------------------------------------------------------
+
+
+def test_round_scale_default():
+    """spark.round defaults scale to 0."""
+    ctx = SessionContext()
+    df = ctx.from_pydict({"x": [1]})
+    assert _val(df, spark.round(lit(2.5))) == 3.0
+
+
+def test_make_dt_interval_defaults():
+    """spark.make_dt_interval with no args returns a zero day-time interval."""
+    import datetime as dt
+
+    ctx = SessionContext()
+    df = ctx.from_pydict({"x": [1]})
+    assert _val(df, spark.make_dt_interval()) == dt.timedelta(0)
+
+
+def test_make_interval_defaults():
+    """spark.make_interval with no args returns a zero interval."""
+    ctx = SessionContext()
+    df = ctx.from_pydict({"x": [1]})
+    assert _val(df, spark.make_interval()).months == 0
+
+
+def test_str_to_map_defaults():
+    """spark.str_to_map defaults delimiters to ',' and ':'."""
+    ctx = SessionContext()
+    df = ctx.from_pydict({"x": [1]})
+    assert _val(df, spark.str_to_map(lit("a:1,b:2"))) == [("a", "1"), ("b", "2")]
+
+
+def test_shuffle_seed_raises():
+    """spark.shuffle(seed=...) raises NotImplementedError until Rust supports it."""
+    with pytest.raises(NotImplementedError, match="seed"):
+        spark.shuffle(spark.array(lit(1), lit(2)), seed=1)
+
+
+def test_like_escape_raises():
+    """spark.like/ilike escapeChar raises NotImplementedError until Rust supports."""
+    with pytest.raises(NotImplementedError, match="escapeChar"):
+        spark.like(lit("a"), lit("a"), escapeChar="\\")
+    with pytest.raises(NotImplementedError, match="escapeChar"):
+        spark.ilike(lit("a"), lit("a"), escapeChar="\\")
+
+
+def test_aggregate_positional_compat():
+    """Pyspark-style positional calls still work after the rename to ``col``."""
+    ctx = SessionContext()
+    df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
+    out = df.aggregate(
+        [],
+        [
+            spark.avg(col("a")).alias("av"),
+            spark.try_sum(col("a")).alias("ts"),
+            spark.collect_list(col("a")).alias("cl"),
+            spark.collect_set(col("a")).alias("cs"),
+        ],
+    ).collect()
+    rec = pa.Table.from_batches(out)
+    assert rec.column("av")[0].as_py() == 2.0
+    assert rec.column("ts")[0].as_py() == 6.0
+
+
 # ---------------------------------------------------------------------------
 # SQL path via enable_spark_functions
 # ---------------------------------------------------------------------------