pydantable.pyspark

PySpark-shaped façade. See PySpark UI.

pydantable.pyspark

PySpark-shaped :class:DataFrame and :class:DataFrameModel (same engine as core).

This is a facade for familiar names (withColumn, orderBy, …), not a Spark cluster client. See :mod:pydantable.pyspark.sql for functions, types, and :class:~pydantable.window_spec.Window.

Expr

Column expression: operators/methods build a Rust AST with static dtypes.

Source code in python/pydantable/expressions.py

class Expr:  # type: ignore[override]
    """Column expression: operators/methods build a Rust AST with static dtypes."""

    def __init__(self, *, rust_expr: Any):
        self._rust_expr = rust_expr

    @property
    def dtype(self) -> Any:
        return self._rust_expr.dtype

    def referenced_columns(self) -> set[str]:
        return set(self._rust_expr.referenced_columns())

    def alias(self, name: str) -> AliasedExpr:
        """Attach an output column name for use in `select` / `with_columns`."""
        if not isinstance(name, str) or not name:
            raise TypeError("alias(name) expects a non-empty string.")
        return AliasedExpr(name=str(name), expr=self)

    def __repr__(self) -> str:
        cls = type(self).__name__
        refs = sorted(self.referenced_columns())
        ref_s = f" refs={refs!r}" if refs else ""
        ast_s = _rust_expr_ast_snippet(self._rust_expr)
        return f"{cls}(dtype={self.dtype!r}{ref_s} ast={ast_s})"

    def _coerce_other(self, other: Any) -> Expr:
        if isinstance(other, Expr):
            return other
        return Literal(value=other)

    def _binary(self, op_symbol: str, other: Any) -> Expr:
        other_expr = self._coerce_other(other)
        rust_expr = get_expression_runtime().binary_op(
            op_symbol, self._rust_expr, other_expr._rust_expr
        )
        return BinaryOp(rust_expr=rust_expr)

    def _binary_reflected(self, op_symbol: str, other: Any) -> Expr:
        # `other <op> self`
        left_expr = self._coerce_other(other)
        rust_expr = get_expression_runtime().binary_op(
            op_symbol, left_expr._rust_expr, self._rust_expr
        )
        return BinaryOp(rust_expr=rust_expr)

    def _compare(self, op_symbol: str, other: Any) -> Expr:
        other_expr = self._coerce_other(other)
        rust_expr = get_expression_runtime().compare_op(
            op_symbol, self._rust_expr, other_expr._rust_expr
        )
        return CompareOp(rust_expr=rust_expr)

    def cast(self, dtype: Any) -> Expr:
        rust_expr = get_expression_runtime().cast_expr(self._rust_expr, dtype)
        return Expr(rust_expr=rust_expr)

    def is_null(self) -> Expr:
        rust_expr = get_expression_runtime().is_null_expr(self._rust_expr)
        return Expr(rust_expr=rust_expr)

    def is_not_null(self) -> Expr:
        rust_expr = get_expression_runtime().is_not_null_expr(self._rust_expr)
        return Expr(rust_expr=rust_expr)

    def over(
        self,
        partition_by: str | list[str] | tuple[str, ...] | None = None,
        order_by: str | list[str] | tuple[str, ...] | None = None,
    ) -> Expr:
        if partition_by is None and order_by is None:
            return self
        raise TypeError(
            "Expr.over(partition_by=..., order_by=...) is not supported. "
            "Use window functions such as row_number().over(WindowSpec(...)) "
            "or pydantable.window_spec.Window.partitionBy(...).orderBy(...)."
        )

    # Arithmetic
    def __add__(self, other: Any) -> Expr:
        return self._binary("+", other)

    def __sub__(self, other: Any) -> Expr:
        return self._binary("-", other)

    def __mul__(self, other: Any) -> Expr:
        return self._binary("*", other)

    def __truediv__(self, other: Any) -> Expr:
        return self._binary("/", other)

    def __radd__(self, other: Any) -> Expr:
        return self._binary_reflected("+", other)

    def __rsub__(self, other: Any) -> Expr:
        return self._binary_reflected("-", other)

    def __rmul__(self, other: Any) -> Expr:
        return self._binary_reflected("*", other)

    def __rtruediv__(self, other: Any) -> Expr:
        return self._binary_reflected("/", other)

    # Comparisons
    def __eq__(self, other: Any) -> Expr:  # type: ignore[override]
        return self._compare("==", other)

    def __ne__(self, other: Any) -> Expr:  # type: ignore[override]
        return self._compare("!=", other)

    def __lt__(self, other: Any) -> Expr:
        return self._compare("<", other)

    def __le__(self, other: Any) -> Expr:
        return self._compare("<=", other)

    def __gt__(self, other: Any) -> Expr:
        return self._compare(">", other)

    def __ge__(self, other: Any) -> Expr:
        return self._compare(">=", other)

    def isin(self, *values: Any) -> Expr:
        if len(values) == 1 and isinstance(values[0], (list, tuple)):
            vals = list(values[0])
        else:
            vals = list(values)
        rust_expr = get_expression_runtime().expr_in_list(self._rust_expr, vals)
        return Expr(rust_expr=rust_expr)

    def is_in(self, *values: Any) -> Expr:
        """Alias of :meth:`isin` (Polars naming parity)."""
        return self.isin(*values)

    def len(self) -> Expr:
        """String length alias (typed-safe): only valid for ``str`` columns."""
        dt = self.dtype
        origin = get_origin(dt)
        args = get_args(dt)
        if origin is None:
            base = dt
        elif origin is getattr(__import__("typing"), "Union", object()) or str(
            origin
        ).endswith("types.UnionType"):
            non_none = [a for a in args if a is not type(None)]
            base = non_none[0] if len(non_none) == 1 else dt
        else:
            base = dt
        if base is not str:
            raise TypeError("len() is only supported for string columns.")
        return self.char_length()

    def between(self, low: Any, high: Any) -> Expr:
        lo = self._coerce_other(low)
        hi = self._coerce_other(high)
        rust_expr = get_expression_runtime().expr_between(
            self._rust_expr, lo._rust_expr, hi._rust_expr
        )
        return Expr(rust_expr=rust_expr)

    def substr(self, start: Any, length: Any | None = None) -> Expr:
        st = self._coerce_other(start)
        rust = get_expression_runtime()
        if length is None:
            rust_expr = rust.expr_substring(self._rust_expr, st._rust_expr, None)
        else:
            ln = self._coerce_other(length)
            rust_expr = rust.expr_substring(
                self._rust_expr, st._rust_expr, ln._rust_expr
            )
        return Expr(rust_expr=rust_expr)

    def char_length(self) -> Expr:
        rust_expr = get_expression_runtime().expr_string_length(self._rust_expr)
        return Expr(rust_expr=rust_expr)

    def struct_field(self, name: str) -> Expr:
        rust_expr = get_expression_runtime().expr_struct_field(self._rust_expr, name)
        return Expr(rust_expr=rust_expr)

    def struct_json_encode(self) -> Expr:
        """Encode struct cells as JSON text (Polars ``struct.json_encode``)."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_struct_json_encode(self._rust_expr))

    def struct_json_path_match(self, path: str) -> Expr:
        """JSONPath against struct cells (JSON-encode then ``str.json_path_match``).

        Same null/match semantics as :meth:`str_json_path_match` on strings.
        Empty ``path`` raises ``ValueError``.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_struct_json_path_match(self._rust_expr, str(path)),
        )

    def struct_rename_fields(self, names: Sequence[str]) -> Expr:
        """Rename struct subfields in order (one new name per existing field)."""
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_struct_rename_fields(
                self._rust_expr, [str(x) for x in names]
            ),
        )

    def struct_with_fields(self, **fields: Any) -> Expr:
        """Add or replace struct subfields (Polars ``struct.with_fields``).

        Each keyword must be a field name; each value must be an :class:`Expr`.
        """
        if not fields:
            raise TypeError(
                "struct_with_fields() requires at least one keyword field=Expr."
            )
        rust = get_expression_runtime()
        updates: list[tuple[str, Any]] = []
        for k, v in fields.items():
            if not isinstance(v, Expr):
                raise TypeError(
                    f"struct_with_fields({k}=...) expects Expr, got {type(v).__name__}."
                )
            updates.append((str(k), v._rust_expr))
        return Expr(
            rust_expr=rust.expr_struct_with_fields(self._rust_expr, updates),
        )

    # Numeric
    def abs(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_abs(self._rust_expr))

    def round(self, decimals: int = 0) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_round(self._rust_expr, int(decimals)))

    def floor(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_floor(self._rust_expr))

    def ceil(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_ceil(self._rust_expr))

    def cumsum(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_row_accum_cum_sum(self._rust_expr))

    def cumprod(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_row_accum_cum_prod(self._rust_expr))

    def cummin(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_row_accum_cum_min(self._rust_expr))

    def cummax(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_row_accum_cum_max(self._rust_expr))

    def diff(self, periods: int = 1) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_row_accum_diff(self._rust_expr, int(periods)))

    def pct_change(self, periods: int = 1) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_row_accum_pct_change(self._rust_expr, int(periods))
        )

    def clip(self, lower: Any = None, upper: Any = None) -> Expr:
        e: Expr = self
        if lower is not None:
            lo = self._coerce_other(lower)
            e = when(self < lo, lo).otherwise(e)
        if upper is not None:
            hi = self._coerce_other(upper)
            e = when(e > hi, hi).otherwise(e)
        return e

    def replace(self, to_replace: dict[Any, Any]) -> Expr:
        items = list(to_replace.items())
        if not items:
            return self
        if len(items) > 64:
            raise ValueError("replace() supports at most 64 mappings.")
        chain = when(self == Literal(value=items[0][0]), Literal(value=items[0][1]))
        for old, new in items[1:]:
            chain = chain.when(self == Literal(value=old), Literal(value=new))
        return chain.otherwise(self)

    # Strings
    def strip(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_string_unary(self._rust_expr, "strip"))

    def upper(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_string_unary(self._rust_expr, "upper"))

    def lower(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_string_unary(self._rust_expr, "lower"))

    def str_replace(
        self, pattern: str, replacement: str, *, literal: bool = True
    ) -> Expr:
        """Replace matches.

        Default ``literal=True`` is substring replace.
        Use ``literal=False`` for Rust regex (syntax differs from Python ``re``;
        see docs).

        Invalid regex patterns may yield null cells at execution (Polars) rather
        than raise.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_replace(
                self._rust_expr,
                str(pattern),
                str(replacement),
                literal=bool(literal),
            )
        )

    def starts_with(self, prefix: str) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_predicate(
                self._rust_expr, "starts_with", str(prefix)
            )
        )

    def ends_with(self, suffix: str) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_predicate(
                self._rust_expr, "ends_with", str(suffix)
            )
        )

    def str_contains(self, substring: str) -> Expr:
        """True where the string contains ``substring`` (literal, not regex).

        The empty substring matches every non-null string (Polars substring
        ``contains`` semantics).
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_predicate(
                self._rust_expr, "contains", str(substring), literal=True
            )
        )

    def str_contains_pat(self, pattern: str, *, literal: bool = False) -> Expr:
        """Substring or Rust-regex match.

        ``literal=False`` uses the Rust ``regex`` dialect (not Python ``re``).
        Raises ``ValueError`` if ``pattern`` is empty in regex mode.
        Malformed regex may yield null per row at execution; see
        ``SUPPORTED_TYPES`` docs.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_predicate(
                self._rust_expr, "contains", str(pattern), literal=bool(literal)
            )
        )

    def matches(self, pattern: str) -> Expr:
        """Regex match predicate (Rust regex dialect)."""
        if not isinstance(pattern, str) or not pattern:
            raise TypeError("matches(pattern) expects a non-empty string.")
        return self.str_contains_pat(pattern, literal=False)

    def is_empty_str(self) -> Expr:
        """True where string cell is exactly ``\"\"``."""
        return self == ""

    def is_blank_str(self) -> Expr:
        """True where string cell is empty after stripping whitespace."""
        return self.strip().char_length() == 0

    def is_null_or_empty_str(self) -> Expr:
        return self.is_null() | self.is_empty_str()

    def is_not_null_and_not_empty_str(self) -> Expr:
        return self.is_not_null() & ~(self.is_empty_str())

    def str_split(self, delimiter: str) -> Expr:
        """Split string column into ``list[str]`` (per-row).

        Delimiter is literal (not regex). Empty ``delimiter`` follows Polars UTF-8
        split rules.
        Null string cells stay null. Edge cases are documented in
        ``SUPPORTED_TYPES``.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_string_split(self._rust_expr, str(delimiter)))

    def str_reverse(self) -> Expr:
        """Reverse each string (Polars ``str.reverse``).

        Unicode edge cases (e.g. combining marks) follow Polars, not naive
        codepoint reversal. See ``SUPPORTED_TYPES``.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_str_reverse(self._rust_expr))

    def str_pad_start(self, length: int, fill_char: str = " ") -> Expr:
        """Pad start to at least ``length`` characters (character count).

        ``fill_char`` must be exactly one non-empty character; otherwise
        ``ValueError`` at build time.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_str_pad_start(
                self._rust_expr, int(length), str(fill_char)
            )
        )

    def str_pad_end(self, length: int, fill_char: str = " ") -> Expr:
        """Pad end to at least ``length`` characters.

        Same ``fill_char`` rules as :meth:`str_pad_start`.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_str_pad_end(
                self._rust_expr, int(length), str(fill_char)
            )
        )

    def str_zfill(self, length: int) -> Expr:
        """Zero-pad strings to ``length`` (sign handled like Polars ``str.zfill``)."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_str_zfill(self._rust_expr, int(length)))

    def str_extract_regex(self, pattern: str, group_index: int = 1) -> Expr:
        """Extract a regex capture group per row (Rust ``regex`` dialect).

        ``group_index`` 0 is the full match; 1+ are capture groups. Empty
        ``pattern`` raises ``ValueError``. No match or invalid regex may yield
        null; see ``SUPPORTED_TYPES``.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_str_extract_regex(
                self._rust_expr, str(pattern), int(group_index)
            )
        )

    def str_json_path_match(self, path: str) -> Expr:
        """JSONPath against JSON text cells (Polars ``str.json_path_match``).

        Returns a **string** column (serialized match). Malformed JSON or no
        match often yields null at execution time. Empty ``path`` raises
        ``ValueError``.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_str_json_path_match(self._rust_expr, str(path)))

    def str_json_decode(self, dtype: Any) -> Expr:
        """Parse JSON text per row into struct or map (Polars ``str.json_decode``).

        ``dtype`` is a nested model or ``dict[str, T]`` annotation, same style as
        :meth:`cast`. Null string cells yield null. With Polars 0.53, **any
        invalid JSON in the column typically fails execution** at
        :meth:`~pydantable.dataframe.DataFrame.collect` (not a per-row null).
        Map targets use the physical list-of-``{key,value}`` entries; JSON must
        be an **array** such as ``[{"key":"a","value":1}]``, not a bare JSON
        object. Polars execution only; see ``INTERFACE_CONTRACT``.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_str_json_decode(self._rust_expr, dtype))

    def strip_prefix(self, prefix: str) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_unary(
                self._rust_expr, "strip_prefix", str(prefix)
            )
        )

    def strip_suffix(self, suffix: str) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_unary(
                self._rust_expr, "strip_suffix", str(suffix)
            )
        )

    def strip_chars(self, chars: str) -> Expr:
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_string_unary(self._rust_expr, "strip_chars", str(chars))
        )

    # Boolean logic (typed; operands must be boolean expressions)
    def __and__(self, other: Any) -> Expr:
        right = other if isinstance(other, Expr) else self._coerce_other(other)
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_logical_and(self._rust_expr, right._rust_expr))

    def __rand__(self, other: Any) -> Expr:
        left = self._coerce_other(other)
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_logical_and(left._rust_expr, self._rust_expr))

    def __or__(self, other: Any) -> Expr:
        right = other if isinstance(other, Expr) else self._coerce_other(other)
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_logical_or(self._rust_expr, right._rust_expr))

    def __ror__(self, other: Any) -> Expr:
        left = self._coerce_other(other)
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_logical_or(left._rust_expr, self._rust_expr))

    def __invert__(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_logical_not(self._rust_expr))

    # Datetime / date parts (Rust validates column type)
    def dt_year(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "year"))

    def dt_month(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "month"))

    def dt_day(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "day"))

    def dt_hour(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "hour"))

    def dt_minute(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "minute"))

    def dt_second(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "second"))

    def dt_nanosecond(self) -> Expr:
        """Sub-second nanoseconds component (``datetime`` or ``time`` columns)."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "nanosecond"))

    def dt_weekday(self) -> Expr:
        """ISO weekday on ``date`` / ``datetime`` (Mon=1 ... Sun=7, same as Polars).

        Not valid on ``time`` columns (``TypeError`` at build time).
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "weekday"))

    def dt_quarter(self) -> Expr:
        """Calendar quarter 1-4 on ``date`` / ``datetime``.

        Not valid on ``time`` columns (``TypeError`` at build time).
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "quarter"))

    def dt_week(self) -> Expr:
        """ISO 8601 week number 1-53 (``date`` / ``datetime``; Polars ``dt.week``).

        Same definition as Python ``datetime.date.isocalendar().week`` /
        Polars ``dt.week()`` (weeks start Monday; week 1 contains the first
        Thursday of the year). Not valid on ``time`` columns.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "week"))

    def dt_dayofyear(self) -> Expr:
        """Day of year 1-366 on ``date`` / ``datetime`` (Spark ``dayofyear``).

        Matches Polars ``dt.ordinal_day()``. Not valid on ``time`` columns.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "dayofyear"))

    def dt_date(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_datetime_to_date(self._rust_expr))

    def strptime(self, format: str, *, to_datetime: bool = False) -> Expr:
        """Parse strings to ``date`` or ``datetime`` (``strftime`` format string)."""
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_strptime(
                self._rust_expr, str(format), bool(to_datetime)
            ),
        )

    def unix_timestamp(self, unit: str = "seconds") -> Expr:
        """Unix epoch from ``date``/``datetime``; ``unit`` is ``seconds`` or ``ms``."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_unix_timestamp(self._rust_expr, str(unit)))

    def from_unix_time(self, unit: str = "seconds") -> Expr:
        """UTC-naive ``datetime`` from numeric epoch; ``unit`` is ``seconds`` or ``ms``.

        Inverse of :meth:`unix_timestamp` for typical non-null numeric input.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_from_unix_time(self._rust_expr, str(unit)))

    def binary_len(self) -> Expr:
        """Byte length of a ``bytes`` column."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_binary_length(self._rust_expr))

    def map_len(self) -> Expr:
        """Number of entries in a ``dict[str, T]`` map column."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_len(self._rust_expr))

    def map_get(self, key: str) -> Expr:
        """Value for a string key (missing key → null)."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_get(self._rust_expr, str(key)))

    def map_contains_key(self, key: str) -> Expr:
        """Whether the map contains the given string key."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_contains_key(self._rust_expr, str(key)))

    def map_keys(self) -> Expr:
        """List of keys for each map cell."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_keys(self._rust_expr))

    def map_values(self) -> Expr:
        """List of values for each map cell."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_values(self._rust_expr))

    def map_entries(self) -> Expr:
        """List of ``{key, value}`` entry structs for each map cell."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_entries(self._rust_expr))

    def map_from_entries(self) -> Expr:
        """Build ``dict[str, T]`` map cells from ``list[{key, value}]`` entries."""
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_map_from_entries(self._rust_expr))

    def element_at(self, key: str) -> Expr:
        """Alias of :meth:`map_get` for map columns."""
        return self.map_get(key)

    # List columns
    def list_len(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_list_len(self._rust_expr))

    def list_get(self, index: Any) -> Expr:
        rust = get_expression_runtime()
        idx = index if isinstance(index, Expr) else Literal(value=index)
        return Expr(
            rust_expr=rust.expr_list_get(self._rust_expr, idx._rust_expr),
        )

    def list_contains(self, value: Any) -> Expr:
        rust = get_expression_runtime()
        v = value if isinstance(value, Expr) else Literal(value=value)
        return Expr(
            rust_expr=rust.expr_list_contains(self._rust_expr, v._rust_expr),
        )

    def contains_any(self, values: Any) -> Expr:
        """Any of the provided values is contained in each list cell."""
        vals = values
        if isinstance(values, Expr):
            raise TypeError("contains_any(values) expects literal values, not Expr.")
        if not isinstance(values, (list, tuple, set)):
            vals = [values]
        expr: Expr | None = None
        for v in list(vals):
            term = self.list_contains(v)
            expr = term if expr is None else (expr | term)
        if expr is None:
            raise TypeError("contains_any(values) expects at least one value.")
        return expr

    def contains_all(self, values: Any) -> Expr:
        """All of the provided values are contained in each list cell."""
        vals = values
        if isinstance(values, Expr):
            raise TypeError("contains_all(values) expects literal values, not Expr.")
        if not isinstance(values, (list, tuple, set)):
            vals = [values]
        expr: Expr | None = None
        for v in list(vals):
            term = self.list_contains(v)
            expr = term if expr is None else (expr & term)
        if expr is None:
            raise TypeError("contains_all(values) expects at least one value.")
        return expr

    def list_is_empty(self) -> Expr:
        return self.list_len() == 0

    def list_any(self) -> Expr:
        """Any True in a boolean list."""
        return self.list_contains(True)

    def list_all(self) -> Expr:
        """All True in a boolean list."""
        return ~self.list_contains(False)

    def map_is_empty(self) -> Expr:
        return self.map_len() == 0

    def map_has_any_key(self, keys: Any) -> Expr:
        ks = keys
        if isinstance(keys, Expr):
            raise TypeError("map_has_any_key(keys) expects literal keys, not Expr.")
        if not isinstance(keys, (list, tuple, set)):
            ks = [keys]
        expr: Expr | None = None
        for k in list(ks):
            term = self.map_contains_key(str(k))
            expr = term if expr is None else (expr | term)
        if expr is None:
            raise TypeError("map_has_any_key(keys) expects at least one key.")
        return expr

    def list_min(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_list_min(self._rust_expr))

    def list_max(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_list_max(self._rust_expr))

    def list_sum(self) -> Expr:
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_list_sum(self._rust_expr))

    def list_mean(self) -> Expr:
        """Mean of each numeric list cell as ``float``.

        Requires ``list[int]`` or ``list[float]``. Empty lists and null list cells
        yield null.
        """
        rust = get_expression_runtime()
        return Expr(rust_expr=rust.expr_list_mean(self._rust_expr))

    def list_join(self, separator: str, *, ignore_nulls: bool = False) -> Expr:
        """Join each ``list[str]`` cell (Polars ``list.join``).

        Empty lists yield empty strings. ``ignore_nulls`` skips null list
        elements when ``True``. See ``SUPPORTED_TYPES``.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_list_join(
                self._rust_expr, str(separator), ignore_nulls=bool(ignore_nulls)
            )
        )

    def list_sort(
        self,
        *,
        descending: bool = False,
        nulls_last: bool = False,
        maintain_order: bool = False,
    ) -> Expr:
        """Sort each list cell in place (``list[int]``, ``list[float]``, etc.).

        ``descending``, ``nulls_last``, and ``maintain_order`` map to Polars
        ``list.sort`` options. Element-type rules are in ``SUPPORTED_TYPES``.
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_list_sort(
                self._rust_expr,
                descending=bool(descending),
                nulls_last=bool(nulls_last),
                maintain_order=bool(maintain_order),
            )
        )

    def list_unique(self, *, stable: bool = False) -> Expr:
        """Deduplicate list elements per row.

        With ``stable=True``, first-seen order is preserved (Polars
        ``unique_stable``).
        """
        rust = get_expression_runtime()
        return Expr(
            rust_expr=rust.expr_list_unique(self._rust_expr, stable=bool(stable))
        )

alias

alias(name)

Attach an output column name for use in select / with_columns.

Source code in python/pydantable/expressions.py

def alias(self, name: str) -> AliasedExpr:
    """Attach an output column name for use in `select` / `with_columns`."""
    if not isinstance(name, str) or not name:
        raise TypeError("alias(name) expects a non-empty string.")
    return AliasedExpr(name=str(name), expr=self)

is_in

is_in(*values)

Alias of :meth:isin (Polars naming parity).

Source code in python/pydantable/expressions.py

def is_in(self, *values: Any) -> Expr:
    """Alias of :meth:`isin` (Polars naming parity)."""
    return self.isin(*values)

len

len()

String length alias (typed-safe): only valid for str columns.

Source code in python/pydantable/expressions.py

def len(self) -> Expr:
    """String length alias (typed-safe): only valid for ``str`` columns."""
    dt = self.dtype
    origin = get_origin(dt)
    args = get_args(dt)
    if origin is None:
        base = dt
    elif origin is getattr(__import__("typing"), "Union", object()) or str(
        origin
    ).endswith("types.UnionType"):
        non_none = [a for a in args if a is not type(None)]
        base = non_none[0] if len(non_none) == 1 else dt
    else:
        base = dt
    if base is not str:
        raise TypeError("len() is only supported for string columns.")
    return self.char_length()

struct_json_encode

struct_json_encode()

Encode struct cells as JSON text (Polars struct.json_encode).

Source code in python/pydantable/expressions.py

def struct_json_encode(self) -> Expr:
    """Encode struct cells as JSON text (Polars ``struct.json_encode``)."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_struct_json_encode(self._rust_expr))

struct_json_path_match

struct_json_path_match(path)

JSONPath against struct cells (JSON-encode then str.json_path_match).

Same null/match semantics as :meth:str_json_path_match on strings. Empty path raises ValueError.

Source code in python/pydantable/expressions.py

def struct_json_path_match(self, path: str) -> Expr:
    """JSONPath against struct cells (JSON-encode then ``str.json_path_match``).

    Same null/match semantics as :meth:`str_json_path_match` on strings.
    Empty ``path`` raises ``ValueError``.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_struct_json_path_match(self._rust_expr, str(path)),
    )

struct_rename_fields

struct_rename_fields(names)

Rename struct subfields in order (one new name per existing field).

Source code in python/pydantable/expressions.py

def struct_rename_fields(self, names: Sequence[str]) -> Expr:
    """Rename struct subfields in order (one new name per existing field)."""
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_struct_rename_fields(
            self._rust_expr, [str(x) for x in names]
        ),
    )

struct_with_fields

struct_with_fields(**fields)

Add or replace struct subfields (Polars struct.with_fields).

Each keyword must be a field name; each value must be an :class:Expr.

Source code in python/pydantable/expressions.py

def struct_with_fields(self, **fields: Any) -> Expr:
    """Add or replace struct subfields (Polars ``struct.with_fields``).

    Each keyword must be a field name; each value must be an :class:`Expr`.
    """
    if not fields:
        raise TypeError(
            "struct_with_fields() requires at least one keyword field=Expr."
        )
    rust = get_expression_runtime()
    updates: list[tuple[str, Any]] = []
    for k, v in fields.items():
        if not isinstance(v, Expr):
            raise TypeError(
                f"struct_with_fields({k}=...) expects Expr, got {type(v).__name__}."
            )
        updates.append((str(k), v._rust_expr))
    return Expr(
        rust_expr=rust.expr_struct_with_fields(self._rust_expr, updates),
    )

str_replace

str_replace(pattern, replacement, *, literal=True)

Replace matches.

Default literal=True is substring replace. Use literal=False for Rust regex (syntax differs from Python re; see docs).

Invalid regex patterns may yield null cells at execution (Polars) rather than raise.

Source code in python/pydantable/expressions.py

def str_replace(
    self, pattern: str, replacement: str, *, literal: bool = True
) -> Expr:
    """Replace matches.

    Default ``literal=True`` is substring replace.
    Use ``literal=False`` for Rust regex (syntax differs from Python ``re``;
    see docs).

    Invalid regex patterns may yield null cells at execution (Polars) rather
    than raise.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_string_replace(
            self._rust_expr,
            str(pattern),
            str(replacement),
            literal=bool(literal),
        )
    )

str_contains

str_contains(substring)

True where the string contains substring (literal, not regex).

The empty substring matches every non-null string (Polars substring contains semantics).

Source code in python/pydantable/expressions.py

def str_contains(self, substring: str) -> Expr:
    """True where the string contains ``substring`` (literal, not regex).

    The empty substring matches every non-null string (Polars substring
    ``contains`` semantics).
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_string_predicate(
            self._rust_expr, "contains", str(substring), literal=True
        )
    )

str_contains_pat

str_contains_pat(pattern, *, literal=False)

Substring or Rust-regex match.

literal=False uses the Rust regex dialect (not Python re). Raises ValueError if pattern is empty in regex mode. Malformed regex may yield null per row at execution; see SUPPORTED_TYPES docs.

Source code in python/pydantable/expressions.py

def str_contains_pat(self, pattern: str, *, literal: bool = False) -> Expr:
    """Substring or Rust-regex match.

    ``literal=False`` uses the Rust ``regex`` dialect (not Python ``re``).
    Raises ``ValueError`` if ``pattern`` is empty in regex mode.
    Malformed regex may yield null per row at execution; see
    ``SUPPORTED_TYPES`` docs.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_string_predicate(
            self._rust_expr, "contains", str(pattern), literal=bool(literal)
        )
    )

matches

matches(pattern)

Regex match predicate (Rust regex dialect).

Source code in python/pydantable/expressions.py

def matches(self, pattern: str) -> Expr:
    """Regex match predicate (Rust regex dialect)."""
    if not isinstance(pattern, str) or not pattern:
        raise TypeError("matches(pattern) expects a non-empty string.")
    return self.str_contains_pat(pattern, literal=False)

is_empty_str

is_empty_str()

True where string cell is exactly "".

Source code in python/pydantable/expressions.py

def is_empty_str(self) -> Expr:
    """True where string cell is exactly ``\"\"``."""
    return self == ""

is_blank_str

is_blank_str()

True where string cell is empty after stripping whitespace.

Source code in python/pydantable/expressions.py

def is_blank_str(self) -> Expr:
    """True where string cell is empty after stripping whitespace."""
    return self.strip().char_length() == 0

str_split

str_split(delimiter)

Split string column into list[str] (per-row).

Delimiter is literal (not regex). Empty delimiter follows Polars UTF-8 split rules. Null string cells stay null. Edge cases are documented in SUPPORTED_TYPES.

Source code in python/pydantable/expressions.py

def str_split(self, delimiter: str) -> Expr:
    """Split string column into ``list[str]`` (per-row).

    Delimiter is literal (not regex). Empty ``delimiter`` follows Polars UTF-8
    split rules.
    Null string cells stay null. Edge cases are documented in
    ``SUPPORTED_TYPES``.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_string_split(self._rust_expr, str(delimiter)))

str_reverse

str_reverse()

Reverse each string (Polars str.reverse).

Unicode edge cases (e.g. combining marks) follow Polars, not naive codepoint reversal. See SUPPORTED_TYPES.

Source code in python/pydantable/expressions.py

def str_reverse(self) -> Expr:
    """Reverse each string (Polars ``str.reverse``).

    Unicode edge cases (e.g. combining marks) follow Polars, not naive
    codepoint reversal. See ``SUPPORTED_TYPES``.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_str_reverse(self._rust_expr))

str_pad_start

str_pad_start(length, fill_char=' ')

Pad start to at least length characters (character count).

fill_char must be exactly one non-empty character; otherwise ValueError at build time.

Source code in python/pydantable/expressions.py

def str_pad_start(self, length: int, fill_char: str = " ") -> Expr:
    """Pad start to at least ``length`` characters (character count).

    ``fill_char`` must be exactly one non-empty character; otherwise
    ``ValueError`` at build time.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_str_pad_start(
            self._rust_expr, int(length), str(fill_char)
        )
    )

str_pad_end

str_pad_end(length, fill_char=' ')

Pad end to at least length characters.

Same fill_char rules as :meth:str_pad_start.

Source code in python/pydantable/expressions.py

def str_pad_end(self, length: int, fill_char: str = " ") -> Expr:
    """Pad end to at least ``length`` characters.

    Same ``fill_char`` rules as :meth:`str_pad_start`.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_str_pad_end(
            self._rust_expr, int(length), str(fill_char)
        )
    )

str_zfill

str_zfill(length)

Zero-pad strings to length (sign handled like Polars str.zfill).

Source code in python/pydantable/expressions.py

def str_zfill(self, length: int) -> Expr:
    """Zero-pad strings to ``length`` (sign handled like Polars ``str.zfill``)."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_str_zfill(self._rust_expr, int(length)))

str_extract_regex

str_extract_regex(pattern, group_index=1)

Extract a regex capture group per row (Rust regex dialect).

group_index 0 is the full match; 1+ are capture groups. Empty pattern raises ValueError. No match or invalid regex may yield null; see SUPPORTED_TYPES.

Source code in python/pydantable/expressions.py

def str_extract_regex(self, pattern: str, group_index: int = 1) -> Expr:
    """Extract a regex capture group per row (Rust ``regex`` dialect).

    ``group_index`` 0 is the full match; 1+ are capture groups. Empty
    ``pattern`` raises ``ValueError``. No match or invalid regex may yield
    null; see ``SUPPORTED_TYPES``.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_str_extract_regex(
            self._rust_expr, str(pattern), int(group_index)
        )
    )

str_json_path_match

str_json_path_match(path)

JSONPath against JSON text cells (Polars str.json_path_match).

Returns a string column (serialized match). Malformed JSON or no match often yields null at execution time. Empty path raises ValueError.

Source code in python/pydantable/expressions.py

def str_json_path_match(self, path: str) -> Expr:
    """JSONPath against JSON text cells (Polars ``str.json_path_match``).

    Returns a **string** column (serialized match). Malformed JSON or no
    match often yields null at execution time. Empty ``path`` raises
    ``ValueError``.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_str_json_path_match(self._rust_expr, str(path)))

str_json_decode

str_json_decode(dtype)

Parse JSON text per row into struct or map (Polars str.json_decode).

dtype is a nested model or dict[str, T] annotation, same style as :meth:cast. Null string cells yield null. With Polars 0.53, any invalid JSON in the column typically fails execution at :meth:~pydantable.dataframe.DataFrame.collect (not a per-row null). Map targets use the physical list-of-{key,value} entries; JSON must be an array such as [{"key":"a","value":1}], not a bare JSON object. Polars execution only; see INTERFACE_CONTRACT.

Source code in python/pydantable/expressions.py

def str_json_decode(self, dtype: Any) -> Expr:
    """Parse JSON text per row into struct or map (Polars ``str.json_decode``).

    ``dtype`` is a nested model or ``dict[str, T]`` annotation, same style as
    :meth:`cast`. Null string cells yield null. With Polars 0.53, **any
    invalid JSON in the column typically fails execution** at
    :meth:`~pydantable.dataframe.DataFrame.collect` (not a per-row null).
    Map targets use the physical list-of-``{key,value}`` entries; JSON must
    be an **array** such as ``[{"key":"a","value":1}]``, not a bare JSON
    object. Polars execution only; see ``INTERFACE_CONTRACT``.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_str_json_decode(self._rust_expr, dtype))

dt_nanosecond

dt_nanosecond()

Sub-second nanoseconds component (datetime or time columns).

Source code in python/pydantable/expressions.py

def dt_nanosecond(self) -> Expr:
    """Sub-second nanoseconds component (``datetime`` or ``time`` columns)."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "nanosecond"))

dt_weekday

dt_weekday()

ISO weekday on date / datetime (Mon=1 ... Sun=7, same as Polars).

Not valid on time columns (TypeError at build time).

Source code in python/pydantable/expressions.py

def dt_weekday(self) -> Expr:
    """ISO weekday on ``date`` / ``datetime`` (Mon=1 ... Sun=7, same as Polars).

    Not valid on ``time`` columns (``TypeError`` at build time).
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "weekday"))

dt_quarter

dt_quarter()

Calendar quarter 1-4 on date / datetime.

Not valid on time columns (TypeError at build time).

Source code in python/pydantable/expressions.py

def dt_quarter(self) -> Expr:
    """Calendar quarter 1-4 on ``date`` / ``datetime``.

    Not valid on ``time`` columns (``TypeError`` at build time).
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "quarter"))

dt_week

dt_week()

ISO 8601 week number 1-53 (date / datetime; Polars dt.week).

Same definition as Python datetime.date.isocalendar().week / Polars dt.week() (weeks start Monday; week 1 contains the first Thursday of the year). Not valid on time columns.

Source code in python/pydantable/expressions.py

def dt_week(self) -> Expr:
    """ISO 8601 week number 1-53 (``date`` / ``datetime``; Polars ``dt.week``).

    Same definition as Python ``datetime.date.isocalendar().week`` /
    Polars ``dt.week()`` (weeks start Monday; week 1 contains the first
    Thursday of the year). Not valid on ``time`` columns.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "week"))

dt_dayofyear

dt_dayofyear()

Day of year 1-366 on date / datetime (Spark dayofyear).

Matches Polars dt.ordinal_day(). Not valid on time columns.

Source code in python/pydantable/expressions.py

def dt_dayofyear(self) -> Expr:
    """Day of year 1-366 on ``date`` / ``datetime`` (Spark ``dayofyear``).

    Matches Polars ``dt.ordinal_day()``. Not valid on ``time`` columns.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_temporal_part(self._rust_expr, "dayofyear"))

strptime

strptime(format, *, to_datetime=False)

Parse strings to date or datetime (strftime format string).

Source code in python/pydantable/expressions.py

def strptime(self, format: str, *, to_datetime: bool = False) -> Expr:
    """Parse strings to ``date`` or ``datetime`` (``strftime`` format string)."""
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_strptime(
            self._rust_expr, str(format), bool(to_datetime)
        ),
    )

unix_timestamp

unix_timestamp(unit='seconds')

Unix epoch from date/datetime; unit is seconds or ms.

Source code in python/pydantable/expressions.py

def unix_timestamp(self, unit: str = "seconds") -> Expr:
    """Unix epoch from ``date``/``datetime``; ``unit`` is ``seconds`` or ``ms``."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_unix_timestamp(self._rust_expr, str(unit)))

from_unix_time

from_unix_time(unit='seconds')

UTC-naive datetime from numeric epoch; unit is seconds or ms.

Inverse of :meth:unix_timestamp for typical non-null numeric input.

Source code in python/pydantable/expressions.py

def from_unix_time(self, unit: str = "seconds") -> Expr:
    """UTC-naive ``datetime`` from numeric epoch; ``unit`` is ``seconds`` or ``ms``.

    Inverse of :meth:`unix_timestamp` for typical non-null numeric input.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_from_unix_time(self._rust_expr, str(unit)))

binary_len

binary_len()

Byte length of a bytes column.

Source code in python/pydantable/expressions.py

def binary_len(self) -> Expr:
    """Byte length of a ``bytes`` column."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_binary_length(self._rust_expr))

map_len

map_len()

Number of entries in a dict[str, T] map column.

Source code in python/pydantable/expressions.py

def map_len(self) -> Expr:
    """Number of entries in a ``dict[str, T]`` map column."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_len(self._rust_expr))

map_get

map_get(key)

Value for a string key (missing key → null).

Source code in python/pydantable/expressions.py

def map_get(self, key: str) -> Expr:
    """Value for a string key (missing key → null)."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_get(self._rust_expr, str(key)))

map_contains_key

map_contains_key(key)

Whether the map contains the given string key.

Source code in python/pydantable/expressions.py

def map_contains_key(self, key: str) -> Expr:
    """Whether the map contains the given string key."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_contains_key(self._rust_expr, str(key)))

map_keys

map_keys()

List of keys for each map cell.

Source code in python/pydantable/expressions.py

def map_keys(self) -> Expr:
    """List of keys for each map cell."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_keys(self._rust_expr))

map_values

map_values()

List of values for each map cell.

Source code in python/pydantable/expressions.py

def map_values(self) -> Expr:
    """List of values for each map cell."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_values(self._rust_expr))

map_entries

map_entries()

List of {key, value} entry structs for each map cell.

Source code in python/pydantable/expressions.py

def map_entries(self) -> Expr:
    """List of ``{key, value}`` entry structs for each map cell."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_entries(self._rust_expr))

map_from_entries

map_from_entries()

Build dict[str, T] map cells from list[{key, value}] entries.

Source code in python/pydantable/expressions.py

def map_from_entries(self) -> Expr:
    """Build ``dict[str, T]`` map cells from ``list[{key, value}]`` entries."""
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_map_from_entries(self._rust_expr))

element_at

element_at(key)

Alias of :meth:map_get for map columns.

Source code in python/pydantable/expressions.py

def element_at(self, key: str) -> Expr:
    """Alias of :meth:`map_get` for map columns."""
    return self.map_get(key)

contains_any

contains_any(values)

Any of the provided values is contained in each list cell.

Source code in python/pydantable/expressions.py

def contains_any(self, values: Any) -> Expr:
    """Any of the provided values is contained in each list cell."""
    vals = values
    if isinstance(values, Expr):
        raise TypeError("contains_any(values) expects literal values, not Expr.")
    if not isinstance(values, (list, tuple, set)):
        vals = [values]
    expr: Expr | None = None
    for v in list(vals):
        term = self.list_contains(v)
        expr = term if expr is None else (expr | term)
    if expr is None:
        raise TypeError("contains_any(values) expects at least one value.")
    return expr

contains_all

contains_all(values)

All of the provided values are contained in each list cell.

Source code in python/pydantable/expressions.py

def contains_all(self, values: Any) -> Expr:
    """All of the provided values are contained in each list cell."""
    vals = values
    if isinstance(values, Expr):
        raise TypeError("contains_all(values) expects literal values, not Expr.")
    if not isinstance(values, (list, tuple, set)):
        vals = [values]
    expr: Expr | None = None
    for v in list(vals):
        term = self.list_contains(v)
        expr = term if expr is None else (expr & term)
    if expr is None:
        raise TypeError("contains_all(values) expects at least one value.")
    return expr

list_any

list_any()

Any True in a boolean list.

Source code in python/pydantable/expressions.py

def list_any(self) -> Expr:
    """Any True in a boolean list."""
    return self.list_contains(True)

list_all

list_all()

All True in a boolean list.

Source code in python/pydantable/expressions.py

def list_all(self) -> Expr:
    """All True in a boolean list."""
    return ~self.list_contains(False)

list_mean

list_mean()

Mean of each numeric list cell as float.

Requires list[int] or list[float]. Empty lists and null list cells yield null.

Source code in python/pydantable/expressions.py

def list_mean(self) -> Expr:
    """Mean of each numeric list cell as ``float``.

    Requires ``list[int]`` or ``list[float]``. Empty lists and null list cells
    yield null.
    """
    rust = get_expression_runtime()
    return Expr(rust_expr=rust.expr_list_mean(self._rust_expr))

list_join

list_join(separator, *, ignore_nulls=False)

Join each list[str] cell (Polars list.join).

Empty lists yield empty strings. ignore_nulls skips null list elements when True. See SUPPORTED_TYPES.

Source code in python/pydantable/expressions.py

def list_join(self, separator: str, *, ignore_nulls: bool = False) -> Expr:
    """Join each ``list[str]`` cell (Polars ``list.join``).

    Empty lists yield empty strings. ``ignore_nulls`` skips null list
    elements when ``True``. See ``SUPPORTED_TYPES``.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_list_join(
            self._rust_expr, str(separator), ignore_nulls=bool(ignore_nulls)
        )
    )

list_sort

list_sort(*, descending=False, nulls_last=False, maintain_order=False)

Sort each list cell in place (list[int], list[float], etc.).

descending, nulls_last, and maintain_order map to Polars list.sort options. Element-type rules are in SUPPORTED_TYPES.

Source code in python/pydantable/expressions.py

def list_sort(
    self,
    *,
    descending: bool = False,
    nulls_last: bool = False,
    maintain_order: bool = False,
) -> Expr:
    """Sort each list cell in place (``list[int]``, ``list[float]``, etc.).

    ``descending``, ``nulls_last``, and ``maintain_order`` map to Polars
    ``list.sort`` options. Element-type rules are in ``SUPPORTED_TYPES``.
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_list_sort(
            self._rust_expr,
            descending=bool(descending),
            nulls_last=bool(nulls_last),
            maintain_order=bool(maintain_order),
        )
    )

list_unique

list_unique(*, stable=False)

Deduplicate list elements per row.

With stable=True, first-seen order is preserved (Polars unique_stable).

Source code in python/pydantable/expressions.py

def list_unique(self, *, stable: bool = False) -> Expr:
    """Deduplicate list elements per row.

    With ``stable=True``, first-seen order is preserved (Polars
    ``unique_stable``).
    """
    rust = get_expression_runtime()
    return Expr(
        rust_expr=rust.expr_list_unique(self._rust_expr, stable=bool(stable))
    )

Schema

Bases: BaseModel

Base model for DataFrame[YourSchema] column definitions.

Uses extra="forbid" so unexpected fields fail validation at construction.

Source code in python/pydantable/schema/_impl.py

class Schema(BaseModel):
    """Base model for ``DataFrame[YourSchema]`` column definitions.

    Uses ``extra="forbid"`` so unexpected fields fail validation at construction.
    """

    model_config = ConfigDict(extra="forbid")

DataFrame

Bases: DataFrame

Typed table with PySpark method names; runs in-process via the Rust core.

Source code in python/pydantable/pyspark/dataframe.py

class DataFrame(CoreDataFrame):
    """Typed table with PySpark method names; runs in-process via the Rust core."""

    @staticmethod
    def _as_pyspark_df(df: CoreDataFrame) -> DataFrame:
        return cast(
            "DataFrame",
            DataFrame._from_plan(
                root_data=df._root_data,
                root_schema_type=df._root_schema_type,
                current_schema_type=df._current_schema_type,
                rust_plan=df._rust_plan,
            ),
        )

    def withColumn(self, name: str, col: Any) -> DataFrame:
        """Add or replace a column (Spark ``withColumn``)."""
        if not isinstance(col, Expr):
            raise TypeError(
                "withColumn(name, col) expects a typed column Expr. "
                "Hint: use pyspark.sql.functions.lit(...) for literals, or "
                "df['col'] / df.col('col') for existing columns."
            )
        return self._as_pyspark_df(self.with_columns(**{name: col}))

    def withColumns(self, colsMap: Mapping[str, Any]) -> DataFrame:
        """Add or replace multiple columns (Spark ``withColumns``)."""
        cm = dict(colsMap)
        for _k, v in cm.items():
            if not isinstance(v, Expr):
                raise TypeError(
                    "withColumns(colsMap) expects mapping values to be typed Exprs. "
                    "Hint: use pyspark.sql.functions.lit(...) for literals."
                )
        return self._as_pyspark_df(self.with_columns(**cm))

    def withColumnRenamed(self, existing: str, new: str) -> DataFrame:
        """Rename one column (Spark ``withColumnRenamed``)."""
        return self._as_pyspark_df(self.rename({existing: new}))

    def withColumnsRenamed(self, colsMap: Mapping[str, str]) -> DataFrame:
        """Rename multiple columns (Spark ``withColumnsRenamed``)."""
        return self._as_pyspark_df(self.rename(dict(colsMap)))

    def toDF(self, *cols: str) -> DataFrame:
        """Rename all columns in order (Spark ``toDF``)."""
        current = list(self.schema_fields().keys())
        if len(cols) != len(current):
            raise ValueError(f"toDF() expects {len(current)} names, got {len(cols)}.")
        mapping = dict(zip(current, cols, strict=True))
        renamed = self.rename(mapping)
        ordered_types = {name: renamed._current_field_types[name] for name in cols}
        ordered_schema = make_derived_schema_type(
            renamed._root_schema_type, ordered_types
        )
        return self._as_pyspark_df(
            type(self)._from_plan(
                root_data=renamed._root_data,
                root_schema_type=renamed._root_schema_type,
                current_schema_type=ordered_schema,
                rust_plan=renamed._rust_plan,
            )
        )

    def transform(
        self,
        func: Callable[..., CoreDataFrame],
        *args: Any,
        **kwargs: Any,
    ) -> DataFrame:
        """Apply ``func(self, ...)``; must return a :class:`DataFrame`."""
        out = func(self, *args, **kwargs)
        if not isinstance(out, CoreDataFrame):
            raise TypeError("transform(func, ...) expects func to return a DataFrame.")
        return self._as_pyspark_df(out)

    def select_typed(self, *cols: str, **named_exprs: Any) -> DataFrame:
        """Typed projection with computed columns (no SQL-string ``selectExpr``)."""
        if not cols and not named_exprs:
            raise ValueError(
                "select_typed() requires at least one column or expression."
            )
        projected = self.with_columns(**named_exprs) if named_exprs else self
        select_cols = [*cols, *named_exprs.keys()]
        return self._as_pyspark_df(projected.select(*select_cols))

    def where(self, condition: Any) -> DataFrame:
        """Filter rows (Spark ``where`` / ``filter``)."""
        return self.filter(condition)

    def filter(self, condition: Any) -> DataFrame:
        """Filter rows (Spark ``filter``)."""
        return cast("DataFrame", super().filter(condition))

    def select(
        self,
        *cols: Any,
        exclude: Any = None,
        **named: Any,
    ) -> DataFrame:
        """Project columns (Spark ``select``)."""
        return cast("DataFrame", super().select(*cols, exclude=exclude, **named))

    def orderBy(
        self,
        *columns: str,
        ascending: bool | list[bool] | None = None,
    ) -> DataFrame:
        """Sort rows (Spark ``orderBy``)."""
        if not columns:
            raise ValueError("orderBy() requires at least one column name.")
        if ascending is None:
            descending: bool | list[bool] = False
        elif isinstance(ascending, bool):
            descending = [not ascending] * len(columns)
        else:
            if len(ascending) != len(columns):
                raise ValueError("ascending length must match columns length.")
            descending = [not x for x in ascending]
        return cast("DataFrame", super().sort(*columns, descending=descending))

    def limit(self, num: int = 0) -> DataFrame:
        """Take the first ``num`` rows (Spark ``limit``)."""
        if num < 0:
            raise ValueError("limit(n) expects n >= 0.")
        return cast("DataFrame", super().head(num))

    def sample(
        self,
        withReplacement: bool | None = None,
        fraction: float | None = None,
        seed: int | None = None,
    ) -> DataFrame:
        """Sample rows (Spark ``sample``; fraction required in this facade)."""
        if fraction is None:
            raise ValueError("sample(fraction=...) is required (Spark-style).")
        if withReplacement is None:
            withReplacement = False
        if not isinstance(withReplacement, bool):
            raise TypeError("sample(withReplacement=...) must be a bool.")
        return self._as_pyspark_df(
            super().sample(
                fraction=fraction, seed=seed, with_replacement=withReplacement
            )
        )

    def explode(
        self,
        column: Any,
        *,
        outer: bool = False,
        streaming: bool | None = None,
    ) -> DataFrame:
        """Explode **list** columns (Spark ``explode``).

        Use ``outer=True`` for ``explode_outer``.
        """
        return self._as_pyspark_df(
            super().explode(column, outer=outer, streaming=streaming)
        )

    def explode_outer(self, column: Any, *, streaming: bool | None = None) -> DataFrame:
        """Explode list columns with Spark-ish outer null/empty handling (see docs)."""
        return self._as_pyspark_df(super().explode_outer(column, streaming=streaming))

    def explode_all(self, *, streaming: bool | None = None) -> DataFrame:
        """Explode every list-typed column (schema-driven; not a Spark name)."""
        return self._as_pyspark_df(super().explode_all(streaming=streaming))

    def posexplode(
        self,
        column: str,
        *,
        pos: str = "pos",
        value: str | None = None,
        outer: bool = False,
        streaming: bool | None = None,
    ) -> DataFrame:
        """Explode one list column with a **0-based** index (Spark ``posexplode``)."""
        return self._as_pyspark_df(
            super().posexplode(
                column, pos=pos, value=value, outer=outer, streaming=streaming
            )
        )

    def posexplode_outer(
        self,
        column: str,
        *,
        pos: str = "pos",
        value: str | None = None,
        streaming: bool | None = None,
    ) -> DataFrame:
        """``posexplode(..., outer=True)`` alias."""
        return self._as_pyspark_df(
            super().posexplode_outer(column, pos=pos, value=value, streaming=streaming)
        )

    def unnest(self, column: Any, *, streaming: bool | None = None) -> DataFrame:
        """Expand **struct** columns to top-level fields (Spark struct pattern)."""
        return self._as_pyspark_df(super().unnest(column, streaming=streaming))

    def unnest_all(self, *, streaming: bool | None = None) -> DataFrame:
        """Unnest every struct-typed column in the schema."""
        return self._as_pyspark_df(super().unnest_all(streaming=streaming))

    def drop(self, *columns: Any, strict: bool = True) -> DataFrame:
        """Drop columns by name (Spark ``drop``)."""
        return cast("DataFrame", super().drop(*columns, strict=strict))

    def distinct(
        self,
        subset: Sequence[str] | None = None,
        *,
        keep: str = "first",
    ) -> DataFrame:
        """Distinct rows (Spark ``distinct`` / Polars ``unique``)."""
        return cast("DataFrame", super().distinct(subset=subset, keep=keep))

    def dropDuplicates(self, subset: list[str] | None = None) -> DataFrame:
        """Spark ``dropDuplicates``; keep-first (engine-dependent without ordering)."""
        return (
            self.distinct(subset=subset, keep="first")
            if subset is not None
            else self.distinct(keep="first")
        )

    def union(self, other: DataFrame) -> DataFrame:
        """Append rows from ``other`` (vertical concat; schemas must align)."""
        return self._as_pyspark_df(
            CoreDataFrame.concat([self, other], how="vertical"),
        )

    def unionAll(self, other: DataFrame) -> DataFrame:
        """Alias of :meth:`union` (Spark naming)."""
        return self.union(other)

    def unionByName(
        self,
        other: DataFrame,
        *,
        allowMissingColumns: bool = False,
    ) -> DataFrame:
        """Union rows aligning columns by name (reorders ``other`` to match ``self``).

        Column sets must match unless ``allowMissingColumns=True`` (then missing
        fields are filled with nulls; requires compatible nullable dtypes).
        """
        left = list(self._current_field_types.keys())
        right = list(other._current_field_types.keys())
        if set(left) != set(right) and not allowMissingColumns:
            raise ValueError(
                "unionByName requires identical column names on both sides "
                "unless allowMissingColumns=True."
            )
        return self._union_by_name_allow_missing(
            other, allow_missing=allowMissingColumns
        )

    def _union_by_name_allow_missing(
        self, other: DataFrame, *, allow_missing: bool
    ) -> DataFrame:
        left_types = dict(self._current_field_types)
        right_types = dict(other._current_field_types)
        if allow_missing:
            all_names = list(dict.fromkeys([*left_types.keys(), *right_types.keys()]))
        else:
            all_names = list(left_types.keys())

        def _is_optional(ann: Any) -> bool:
            origin = get_origin(ann)
            if origin is None:
                return False
            return (origin is UnionType or origin is Union) and type(None) in get_args(
                ann
            )

        def _strip_optional(ann: Any) -> Any:
            args = tuple(a for a in get_args(ann) if a is not type(None))
            if len(args) == 1:
                return args[0]
            return ann

        def _optionalize(ann: Any) -> Any:
            return ann | None  # PEP604

        def _unify(name: str, lt: Any | None, rt: Any | None) -> Any:
            if lt is None:
                if not allow_missing:
                    raise ValueError(
                        "unionByName requires identical column names on both sides "
                        "unless allowMissingColumns=True."
                    )
                return _optionalize(rt)
            if rt is None:
                if not allow_missing:
                    raise ValueError(
                        "unionByName requires identical column names on both sides "
                        "unless allowMissingColumns=True."
                    )
                return _optionalize(lt)
            if lt == rt:
                return lt

            l_opt = _is_optional(lt)
            r_opt = _is_optional(rt)
            l_base = _strip_optional(lt) if l_opt else lt
            r_base = _strip_optional(rt) if r_opt else rt

            if l_base == r_base:
                return _optionalize(l_base) if (l_opt or r_opt) else l_base
            if {l_base, r_base} == {int, float}:
                return _optionalize(float) if (l_opt or r_opt) else float

            msg = (
                "unionByName has incompatible dtypes for "
                f"column {name!r}: left={lt!r}, right={rt!r}"
            )
            if allow_missing:
                msg = (
                    "unionByName(allowMissingColumns=True) has incompatible dtypes for "
                    + (f"column {name!r}: left={lt!r}, right={rt!r}")
                )
            raise TypeError(msg)

        unified: dict[str, Any] = {}
        for name in all_names:
            unified[name] = _unify(name, left_types.get(name), right_types.get(name))

        def _align(df: CoreDataFrame, side_types: dict[str, Any]) -> CoreDataFrame:
            out = df
            for name in all_names:
                target = unified[name]
                if name not in side_types:
                    out = out.with_columns(**{name: Literal(value=None).cast(target)})
                elif side_types[name] != target:
                    out = out.with_columns(
                        **{
                            name: ColumnRef(name=name, dtype=side_types[name]).cast(
                                target
                            )
                        }
                    )
            return out.select(*all_names)

        left_aligned = _align(self, left_types)
        right_aligned = _align(other, right_types)
        return self._as_pyspark_df(
            CoreDataFrame.concat([left_aligned, right_aligned], how="vertical"),
        )

    def intersect(self, other: DataFrame) -> DataFrame:
        """Rows present in both frames (distinct row keys; same schema required)."""
        keys = list(self._current_field_types.keys())
        if keys != list(other._current_field_types.keys()):
            raise ValueError("intersect() requires identical schemas.")
        inner = super().join(other, on=keys, how="inner")
        return self._as_pyspark_df(inner.distinct())

    def subtract(self, other: DataFrame) -> DataFrame:
        """Anti join on all columns (schemas must match)."""
        keys = list(self._current_field_types.keys())
        if keys != list(other._current_field_types.keys()):
            raise ValueError("subtract() requires identical schemas.")
        return self._as_pyspark_df(super().join(other, on=keys, how="anti"))

    def except_(self, other: DataFrame) -> DataFrame:
        """Distinct set difference (Spark ``except`` / SQL ``EXCEPT DISTINCT``)."""
        return self.subtract(other).distinct()

    # Python keyword compatibility: expose `.except(...)` name too.
    except__doc__ = "Alias for except_ (Spark except)."

    def exceptAll(self, other: DataFrame) -> DataFrame:
        """Multiset difference (Spark ``EXCEPT ALL``)."""
        if self._current_field_types != other._current_field_types:
            raise ValueError("exceptAll() requires identical schemas.")
        use_streaming = (
            bool(self._engine_streaming_default)
            if self._engine_streaming_default is not None
            else False
        )
        out_data, schema_desc = self._engine.execute_except_all(
            self._rust_plan,
            self._root_data,
            other._rust_plan,
            other._root_data,
            as_python_lists=True,
            streaming=use_streaming,
        )
        derived_fields = self._field_types_from_descriptors(schema_desc)
        derived_schema_type = make_derived_schema_type(
            self._current_schema_type, derived_fields
        )
        rust_plan = self._engine.make_plan(derived_fields)
        return self._as_pyspark_df(
            self._from_plan(
                root_data=out_data,
                root_schema_type=derived_schema_type,
                current_schema_type=derived_schema_type,
                rust_plan=rust_plan,
                engine=self._engine,
            )
        )

    def intersectAll(self, other: DataFrame) -> DataFrame:
        """Multiset intersection (Spark ``INTERSECT ALL``)."""
        if self._current_field_types != other._current_field_types:
            raise ValueError("intersectAll() requires identical schemas.")
        use_streaming = (
            bool(self._engine_streaming_default)
            if self._engine_streaming_default is not None
            else False
        )
        out_data, schema_desc = self._engine.execute_intersect_all(
            self._rust_plan,
            self._root_data,
            other._rust_plan,
            other._root_data,
            as_python_lists=True,
            streaming=use_streaming,
        )
        derived_fields = self._field_types_from_descriptors(schema_desc)
        derived_schema_type = make_derived_schema_type(
            self._current_schema_type, derived_fields
        )
        rust_plan = self._engine.make_plan(derived_fields)
        return self._as_pyspark_df(
            self._from_plan(
                root_data=out_data,
                root_schema_type=derived_schema_type,
                current_schema_type=derived_schema_type,
                rust_plan=rust_plan,
                engine=self._engine,
            )
        )

    def join(
        self,
        other: CoreDataFrame,
        *,
        on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        left_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        right_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        how: str = "inner",
        suffix: str = "_right",
        coalesce: bool | None = None,
        validate: str | None = None,
        join_nulls: bool | None = None,
        maintain_order: bool | str | None = None,
        streaming: bool | None = None,
        keepRightJoinKeys: bool = False,
        keepLeftJoinKeys: bool = False,
    ) -> DataFrame:
        """Join two frames (Spark-shaped wrapper over core join).

        Supports Spark-ish join modes like ``left_semi``/``left_anti`` and
        ``right_semi``/``right_anti`` (aliases over core ``semi``/``anti``), plus
        Spark-style validate shorthands: ``validate='1:1'|'1:m'|'m:1'|'m:m'``.
        """
        if not isinstance(keepRightJoinKeys, bool):
            raise TypeError("join(keepRightJoinKeys=...) expects a bool.")
        if not isinstance(keepLeftJoinKeys, bool):
            raise TypeError("join(keepLeftJoinKeys=...) expects a bool.")
        if on is not None and (left_on is not None or right_on is not None):
            raise ValueError(
                "join() use either on=... or left_on=/right_on=..., not both."
            )

        how_norm = str(how).strip().lower()
        how_aliases = {
            "outer": "outer",
            "full_outer": "full",
            "full": "full",
            "right_outer": "right",
            "left_outer": "left",
        }
        how_norm = how_aliases.get(how_norm, how_norm)
        if how_norm == "left_semi":
            how_norm = "semi"
        elif how_norm == "left_anti":
            how_norm = "anti"
        elif how_norm == "right_semi":
            how_norm = "right_semi"
        elif how_norm == "right_anti":
            how_norm = "right_anti"

        supported_hows = {
            "inner",
            "left",
            "right",
            "outer",
            "full",
            "cross",
            "semi",
            "anti",
            "left_semi",
            "left_anti",
            "right_semi",
            "right_anti",
            "left_outer",
            "right_outer",
            "full_outer",
        }
        if how_norm not in {"right_semi", "right_anti"} and how_norm not in {
            "inner",
            "left",
            "right",
            "outer",
            "full",
            "cross",
            "semi",
            "anti",
        }:
            raise ValueError(
                "join(how=...) must be one of: "
                + ", ".join(repr(x) for x in sorted(supported_hows))
            )

        def _resolve_key_arg(
            arg: str | ColumnRef | Sequence[str | ColumnRef] | None, *, arg_name: str
        ) -> str | list[str] | None:
            if arg is None:
                return None
            if isinstance(arg, str):
                return arg
            if isinstance(arg, ColumnRef):
                referenced = arg.referenced_columns()
                if len(referenced) != 1:
                    raise TypeError(
                        f"join({arg_name}=...) ColumnRef must reference exactly "
                        f"one column; referenced_columns={sorted(referenced)!r}"
                    )
                return next(iter(referenced))
            raw = list(arg)
            out: list[str] = []
            for k in raw:
                if isinstance(k, str):
                    out.append(k)
                elif isinstance(k, ColumnRef):
                    referenced = k.referenced_columns()
                    if len(referenced) != 1:
                        raise TypeError(
                            f"join({arg_name}=...) ColumnRef must reference exactly "
                            f"one column; referenced_columns={sorted(referenced)!r}"
                        )
                    out.append(next(iter(referenced)))
                else:
                    raise TypeError(
                        f"join({arg_name}=...) expects str, ColumnRef, or a "
                        "sequence of str|ColumnRef."
                    )
            if len(set(out)) != len(out):
                raise ValueError(
                    f"join({arg_name}=...) must not contain duplicate keys."
                )
            return out

        on_names = _resolve_key_arg(on, arg_name="on")
        left_names = _resolve_key_arg(left_on, arg_name="left_on")
        right_names = _resolve_key_arg(right_on, arg_name="right_on")

        used_on = on_names is not None
        used_lr = left_names is not None or right_names is not None
        if how_norm != "cross":
            if used_on:
                pass
            elif used_lr:
                if left_names is None or right_names is None:
                    raise ValueError(
                        "join() requires both left_on=... and right_on=... when on=... "
                        "is not set."
                    )
                left_list = (
                    [left_names] if isinstance(left_names, str) else list(left_names)
                )
                right_list = (
                    [right_names] if isinstance(right_names, str) else list(right_names)
                )
                if len(left_list) != len(right_list):
                    raise ValueError(
                        "join(left_on=..., right_on=...) key lists must match length."
                    )
            else:
                raise ValueError(
                    "join() requires on=... or left_on=.../right_on=... for "
                    "non-cross joins."
                )

        if how_norm in ("right_semi", "right_anti"):
            swapped_how = "semi" if how_norm == "right_semi" else "anti"
            if used_on:
                swapped = other.join(
                    self,
                    on=on_names,
                    how=swapped_how,
                    suffix=suffix,
                    coalesce=coalesce,
                    validate=validate,
                    join_nulls=join_nulls,
                    maintain_order=maintain_order,
                    streaming=streaming,
                )
            else:
                # swap sides: keys swap too
                swapped = other.join(
                    self,
                    left_on=right_names,
                    right_on=left_names,
                    how=swapped_how,
                    suffix=suffix,
                    coalesce=coalesce,
                    validate=validate,
                    join_nulls=join_nulls,
                    maintain_order=maintain_order,
                    streaming=streaming,
                )
            # right-only output is already guaranteed by semi/anti on swapped join.
            _ = keepLeftJoinKeys  # reserved for future parity knobs
            return self._as_pyspark_df(swapped)

        joined = super().join(
            other,
            on=on_names,
            left_on=left_names,
            right_on=right_names,
            how=how_norm,
            suffix=suffix,
            coalesce=coalesce,
            validate=validate,
            join_nulls=join_nulls,
            maintain_order=maintain_order,
            streaming=streaming,
        )

        # Spark-ish USING behavior: when joining on same-named keys, keep one copy
        # of each join key by default.
        if used_on and on_names is not None and not keepRightJoinKeys:
            # Core join can still produce suffixed duplicates depending on rules;
            # drop any right-side key duplicates if present.
            keys = [on_names] if isinstance(on_names, str) else list(on_names)
            drop_cols: list[str] = []
            for k in keys:
                rk = f"{k}{suffix}"
                if rk in joined._current_field_types:
                    drop_cols.append(rk)
            if drop_cols:
                joined = joined.drop(*drop_cols)
        elif (
            used_lr
            and not keepRightJoinKeys
            and left_names is not None
            and right_names is not None
        ):
            left_list = (
                [left_names] if isinstance(left_names, str) else list(left_names)
            )
            right_list = (
                [right_names] if isinstance(right_names, str) else list(right_names)
            )
            drop_cols: list[str] = []
            for lk, rk in zip(left_list, right_list, strict=True):
                if lk == rk:
                    cand = f"{rk}{suffix}"
                    if cand in joined._current_field_types:
                        drop_cols.append(cand)
            if drop_cols:
                joined = joined.drop(*drop_cols)

        return self._as_pyspark_df(joined)

    def group_by(
        self,
        *keys: str | ColumnRef,
        maintain_order: bool = False,
        drop_nulls: bool = True,
    ) -> PySparkGroupedDataFrame:
        inner = super().group_by(
            *keys, maintain_order=maintain_order, drop_nulls=drop_nulls
        )
        return PySparkGroupedDataFrame(
            inner._df,
            inner._keys,
            maintain_order=inner._maintain_order,
            drop_nulls=inner._drop_nulls,
        )

    def groupBy(
        self,
        *keys: str | ColumnRef,
        maintain_order: bool = False,
        drop_nulls: bool = True,
    ) -> PySparkGroupedDataFrame:
        """Spark alias of :meth:`group_by`."""
        return self.group_by(
            *keys, maintain_order=maintain_order, drop_nulls=drop_nulls
        )

    def sort(
        self,
        *columns: str,
        ascending: bool | list[bool] | None = None,
    ) -> DataFrame:
        """Sort by column names (Spark ``sort`` / ``orderBy``)."""
        return self.orderBy(*columns, ascending=ascending)

    def crossJoin(self, other: DataFrame) -> DataFrame:
        """Cross join (Spark ``crossJoin``)."""
        return self._as_pyspark_df(super().join(other, how="cross"))

    def count(self) -> int:
        """Row count via ``global_row_count()`` (Spark-style action)."""
        d = self.select(global_row_count()).to_dict()
        col = next(iter(d.keys()))
        return int(d[col][0])

    def fillna(
        self,
        value: Any = None,
        *,
        subset: str | list[str] | None = None,
    ) -> DataFrame:
        """Spark name for :meth:`fill_null`."""
        if isinstance(subset, str):
            sub: str | list[str] | None = subset
        elif subset is None:
            sub = None
        else:
            sub = list(subset)
        return self._as_pyspark_df(self.fill_null(value, subset=sub))

    def dropna(
        self,
        how: str = "any",
        *,
        thresh: int | None = None,
        subset: str | list[str] | None = None,
    ) -> DataFrame:
        """Spark name for :meth:`drop_nulls` (``thresh`` maps to ``threshold``)."""
        if isinstance(subset, str):
            sub: str | list[str] | None = subset
        elif subset is None:
            sub = None
        else:
            sub = list(subset)
        return self._as_pyspark_df(
            self.drop_nulls(sub, how=how, threshold=thresh),
        )

    @property
    def na(self) -> DataFrameNaFunctions:
        return DataFrameNaFunctions(self)

    def printSchema(self, level: int | None = None) -> None:
        """Print schema tree (Spark ``printSchema``; ``level`` accepted for parity)."""
        _ = level
        print("root")
        for line in _format_schema_type_lines(self.schema, ""):
            print(line)

    def explain(
        self,
        extended: bool | None = None,
        mode: str | None = None,
        *,
        format: TypingLiteral["text", "json"] = "text",
    ) -> None:
        """Print logical plan (Spark ``explain``; ``extended`` / ``mode`` reserved)."""
        _ = extended
        _ = mode
        out = super().explain(format=format)
        if isinstance(out, dict):
            import json

            print(json.dumps(out, indent=2, sort_keys=True))
        else:
            print(out)

    def describe(self) -> str:
        """Same string as core :meth:`~pydantable.dataframe.DataFrame.describe`."""
        return super().describe()

    def toPandas(self):
        """Materialize to a ``pandas.DataFrame`` (requires ``pandas`` installed)."""
        try:
            import pandas as pd
        except ImportError as e:
            raise ImportError("toPandas() requires pandas (pip install pandas).") from e
        return pd.DataFrame(self.to_dict())

    def show(
        self,
        n: int = 20,
        truncate: bool = True,
        vertical: bool = False,
    ) -> None:
        """Print up to ``n`` rows (via :meth:`head` + :meth:`to_dict`).

        **Cost:** bounded materialization. Not a distributed Spark runtime.
        """
        h = self.head(int(n))
        data = h.to_dict()
        if vertical:
            cols = list(data.keys())
            if not cols:
                print("(empty)")
                return
            nrows = len(next(iter(data.values())))
            for i in range(nrows):
                print(f"- record {i}")
                for c in cols:
                    print(f"  {c}: {data[c][i]!r}")
            return
        print(_text_show_table(data, truncate=truncate))

    def summary(self) -> str:
        """Spark-style name for :meth:`describe` (numeric columns; materializes)."""
        return self.describe()

    @property
    def schema(self) -> StructType:
        fields = [
            StructField(n, annotation_to_data_type(t))
            for n, t in self._current_field_types.items()
        ]
        return StructType(fields)

    def __getitem__(self, key: str | list[str]) -> Any:
        if isinstance(key, str):
            return self.col(key)
        if isinstance(key, list):
            if not key:
                raise ValueError("Column list must be non-empty.")
            return self.select(*key)
        raise TypeError(
            "DataFrame indexing supports a single column name (str) or list[str]."
        )

withColumn

withColumn(name, col)

Add or replace a column (Spark withColumn).

Source code in python/pydantable/pyspark/dataframe.py

def withColumn(self, name: str, col: Any) -> DataFrame:
    """Add or replace a column (Spark ``withColumn``)."""
    if not isinstance(col, Expr):
        raise TypeError(
            "withColumn(name, col) expects a typed column Expr. "
            "Hint: use pyspark.sql.functions.lit(...) for literals, or "
            "df['col'] / df.col('col') for existing columns."
        )
    return self._as_pyspark_df(self.with_columns(**{name: col}))

withColumns

withColumns(colsMap)

Add or replace multiple columns (Spark withColumns).

Source code in python/pydantable/pyspark/dataframe.py

def withColumns(self, colsMap: Mapping[str, Any]) -> DataFrame:
    """Add or replace multiple columns (Spark ``withColumns``)."""
    cm = dict(colsMap)
    for _k, v in cm.items():
        if not isinstance(v, Expr):
            raise TypeError(
                "withColumns(colsMap) expects mapping values to be typed Exprs. "
                "Hint: use pyspark.sql.functions.lit(...) for literals."
            )
    return self._as_pyspark_df(self.with_columns(**cm))

withColumnRenamed

withColumnRenamed(existing, new)

Rename one column (Spark withColumnRenamed).

Source code in python/pydantable/pyspark/dataframe.py

def withColumnRenamed(self, existing: str, new: str) -> DataFrame:
    """Rename one column (Spark ``withColumnRenamed``)."""
    return self._as_pyspark_df(self.rename({existing: new}))

withColumnsRenamed

withColumnsRenamed(colsMap)

Rename multiple columns (Spark withColumnsRenamed).

Source code in python/pydantable/pyspark/dataframe.py

def withColumnsRenamed(self, colsMap: Mapping[str, str]) -> DataFrame:
    """Rename multiple columns (Spark ``withColumnsRenamed``)."""
    return self._as_pyspark_df(self.rename(dict(colsMap)))

toDF

toDF(*cols)

Rename all columns in order (Spark toDF).

Source code in python/pydantable/pyspark/dataframe.py

def toDF(self, *cols: str) -> DataFrame:
    """Rename all columns in order (Spark ``toDF``)."""
    current = list(self.schema_fields().keys())
    if len(cols) != len(current):
        raise ValueError(f"toDF() expects {len(current)} names, got {len(cols)}.")
    mapping = dict(zip(current, cols, strict=True))
    renamed = self.rename(mapping)
    ordered_types = {name: renamed._current_field_types[name] for name in cols}
    ordered_schema = make_derived_schema_type(
        renamed._root_schema_type, ordered_types
    )
    return self._as_pyspark_df(
        type(self)._from_plan(
            root_data=renamed._root_data,
            root_schema_type=renamed._root_schema_type,
            current_schema_type=ordered_schema,
            rust_plan=renamed._rust_plan,
        )
    )

transform

transform(func, *args, **kwargs)

Apply func(self, ...); must return a :class:DataFrame.

Source code in python/pydantable/pyspark/dataframe.py

def transform(
    self,
    func: Callable[..., CoreDataFrame],
    *args: Any,
    **kwargs: Any,
) -> DataFrame:
    """Apply ``func(self, ...)``; must return a :class:`DataFrame`."""
    out = func(self, *args, **kwargs)
    if not isinstance(out, CoreDataFrame):
        raise TypeError("transform(func, ...) expects func to return a DataFrame.")
    return self._as_pyspark_df(out)

select_typed

select_typed(*cols, **named_exprs)

Typed projection with computed columns (no SQL-string selectExpr).

Source code in python/pydantable/pyspark/dataframe.py

def select_typed(self, *cols: str, **named_exprs: Any) -> DataFrame:
    """Typed projection with computed columns (no SQL-string ``selectExpr``)."""
    if not cols and not named_exprs:
        raise ValueError(
            "select_typed() requires at least one column or expression."
        )
    projected = self.with_columns(**named_exprs) if named_exprs else self
    select_cols = [*cols, *named_exprs.keys()]
    return self._as_pyspark_df(projected.select(*select_cols))

where

where(condition)

Filter rows (Spark where / filter).

Source code in python/pydantable/pyspark/dataframe.py

def where(self, condition: Any) -> DataFrame:
    """Filter rows (Spark ``where`` / ``filter``)."""
    return self.filter(condition)

filter

filter(condition)

Filter rows (Spark filter).

Source code in python/pydantable/pyspark/dataframe.py

def filter(self, condition: Any) -> DataFrame:
    """Filter rows (Spark ``filter``)."""
    return cast("DataFrame", super().filter(condition))

select

select(*cols, exclude=None, **named)

Project columns (Spark select).

Source code in python/pydantable/pyspark/dataframe.py

def select(
    self,
    *cols: Any,
    exclude: Any = None,
    **named: Any,
) -> DataFrame:
    """Project columns (Spark ``select``)."""
    return cast("DataFrame", super().select(*cols, exclude=exclude, **named))

orderBy

orderBy(*columns, ascending=None)

Sort rows (Spark orderBy).

Source code in python/pydantable/pyspark/dataframe.py

def orderBy(
    self,
    *columns: str,
    ascending: bool | list[bool] | None = None,
) -> DataFrame:
    """Sort rows (Spark ``orderBy``)."""
    if not columns:
        raise ValueError("orderBy() requires at least one column name.")
    if ascending is None:
        descending: bool | list[bool] = False
    elif isinstance(ascending, bool):
        descending = [not ascending] * len(columns)
    else:
        if len(ascending) != len(columns):
            raise ValueError("ascending length must match columns length.")
        descending = [not x for x in ascending]
    return cast("DataFrame", super().sort(*columns, descending=descending))

limit

limit(num=0)

Take the first num rows (Spark limit).

Source code in python/pydantable/pyspark/dataframe.py

def limit(self, num: int = 0) -> DataFrame:
    """Take the first ``num`` rows (Spark ``limit``)."""
    if num < 0:
        raise ValueError("limit(n) expects n >= 0.")
    return cast("DataFrame", super().head(num))

sample

sample(withReplacement=None, fraction=None, seed=None)

Sample rows (Spark sample; fraction required in this facade).

Source code in python/pydantable/pyspark/dataframe.py

def sample(
    self,
    withReplacement: bool | None = None,
    fraction: float | None = None,
    seed: int | None = None,
) -> DataFrame:
    """Sample rows (Spark ``sample``; fraction required in this facade)."""
    if fraction is None:
        raise ValueError("sample(fraction=...) is required (Spark-style).")
    if withReplacement is None:
        withReplacement = False
    if not isinstance(withReplacement, bool):
        raise TypeError("sample(withReplacement=...) must be a bool.")
    return self._as_pyspark_df(
        super().sample(
            fraction=fraction, seed=seed, with_replacement=withReplacement
        )
    )

explode

explode(column, *, outer=False, streaming=None)

Explode list columns (Spark explode).

Use outer=True for explode_outer.

Source code in python/pydantable/pyspark/dataframe.py

def explode(
    self,
    column: Any,
    *,
    outer: bool = False,
    streaming: bool | None = None,
) -> DataFrame:
    """Explode **list** columns (Spark ``explode``).

    Use ``outer=True`` for ``explode_outer``.
    """
    return self._as_pyspark_df(
        super().explode(column, outer=outer, streaming=streaming)
    )

explode_outer

explode_outer(column, *, streaming=None)

Explode list columns with Spark-ish outer null/empty handling (see docs).

Source code in python/pydantable/pyspark/dataframe.py

def explode_outer(self, column: Any, *, streaming: bool | None = None) -> DataFrame:
    """Explode list columns with Spark-ish outer null/empty handling (see docs)."""
    return self._as_pyspark_df(super().explode_outer(column, streaming=streaming))

explode_all

explode_all(*, streaming=None)

Explode every list-typed column (schema-driven; not a Spark name).

Source code in python/pydantable/pyspark/dataframe.py

def explode_all(self, *, streaming: bool | None = None) -> DataFrame:
    """Explode every list-typed column (schema-driven; not a Spark name)."""
    return self._as_pyspark_df(super().explode_all(streaming=streaming))

posexplode

posexplode(column, *, pos='pos', value=None, outer=False, streaming=None)

Explode one list column with a 0-based index (Spark posexplode).

Source code in python/pydantable/pyspark/dataframe.py

def posexplode(
    self,
    column: str,
    *,
    pos: str = "pos",
    value: str | None = None,
    outer: bool = False,
    streaming: bool | None = None,
) -> DataFrame:
    """Explode one list column with a **0-based** index (Spark ``posexplode``)."""
    return self._as_pyspark_df(
        super().posexplode(
            column, pos=pos, value=value, outer=outer, streaming=streaming
        )
    )

posexplode_outer

posexplode_outer(column, *, pos='pos', value=None, streaming=None)

posexplode(..., outer=True) alias.

Source code in python/pydantable/pyspark/dataframe.py

def posexplode_outer(
    self,
    column: str,
    *,
    pos: str = "pos",
    value: str | None = None,
    streaming: bool | None = None,
) -> DataFrame:
    """``posexplode(..., outer=True)`` alias."""
    return self._as_pyspark_df(
        super().posexplode_outer(column, pos=pos, value=value, streaming=streaming)
    )

unnest

unnest(column, *, streaming=None)

Expand struct columns to top-level fields (Spark struct pattern).

Source code in python/pydantable/pyspark/dataframe.py

def unnest(self, column: Any, *, streaming: bool | None = None) -> DataFrame:
    """Expand **struct** columns to top-level fields (Spark struct pattern)."""
    return self._as_pyspark_df(super().unnest(column, streaming=streaming))

unnest_all

unnest_all(*, streaming=None)

Unnest every struct-typed column in the schema.

Source code in python/pydantable/pyspark/dataframe.py

def unnest_all(self, *, streaming: bool | None = None) -> DataFrame:
    """Unnest every struct-typed column in the schema."""
    return self._as_pyspark_df(super().unnest_all(streaming=streaming))

drop

drop(*columns, strict=True)

Drop columns by name (Spark drop).

Source code in python/pydantable/pyspark/dataframe.py

def drop(self, *columns: Any, strict: bool = True) -> DataFrame:
    """Drop columns by name (Spark ``drop``)."""
    return cast("DataFrame", super().drop(*columns, strict=strict))

distinct

distinct(subset=None, *, keep='first')

Distinct rows (Spark distinct / Polars unique).

Source code in python/pydantable/pyspark/dataframe.py

def distinct(
    self,
    subset: Sequence[str] | None = None,
    *,
    keep: str = "first",
) -> DataFrame:
    """Distinct rows (Spark ``distinct`` / Polars ``unique``)."""
    return cast("DataFrame", super().distinct(subset=subset, keep=keep))

dropDuplicates

dropDuplicates(subset=None)

Spark dropDuplicates; keep-first (engine-dependent without ordering).

Source code in python/pydantable/pyspark/dataframe.py

def dropDuplicates(self, subset: list[str] | None = None) -> DataFrame:
    """Spark ``dropDuplicates``; keep-first (engine-dependent without ordering)."""
    return (
        self.distinct(subset=subset, keep="first")
        if subset is not None
        else self.distinct(keep="first")
    )

union

union(other)

Append rows from other (vertical concat; schemas must align).

Source code in python/pydantable/pyspark/dataframe.py

def union(self, other: DataFrame) -> DataFrame:
    """Append rows from ``other`` (vertical concat; schemas must align)."""
    return self._as_pyspark_df(
        CoreDataFrame.concat([self, other], how="vertical"),
    )

unionAll

unionAll(other)

Alias of :meth:union (Spark naming).

Source code in python/pydantable/pyspark/dataframe.py

def unionAll(self, other: DataFrame) -> DataFrame:
    """Alias of :meth:`union` (Spark naming)."""
    return self.union(other)

unionByName

unionByName(other, *, allowMissingColumns=False)

Union rows aligning columns by name (reorders other to match self).

Column sets must match unless allowMissingColumns=True (then missing fields are filled with nulls; requires compatible nullable dtypes).

Source code in python/pydantable/pyspark/dataframe.py

def unionByName(
    self,
    other: DataFrame,
    *,
    allowMissingColumns: bool = False,
) -> DataFrame:
    """Union rows aligning columns by name (reorders ``other`` to match ``self``).

    Column sets must match unless ``allowMissingColumns=True`` (then missing
    fields are filled with nulls; requires compatible nullable dtypes).
    """
    left = list(self._current_field_types.keys())
    right = list(other._current_field_types.keys())
    if set(left) != set(right) and not allowMissingColumns:
        raise ValueError(
            "unionByName requires identical column names on both sides "
            "unless allowMissingColumns=True."
        )
    return self._union_by_name_allow_missing(
        other, allow_missing=allowMissingColumns
    )

intersect

intersect(other)

Rows present in both frames (distinct row keys; same schema required).

Source code in python/pydantable/pyspark/dataframe.py

def intersect(self, other: DataFrame) -> DataFrame:
    """Rows present in both frames (distinct row keys; same schema required)."""
    keys = list(self._current_field_types.keys())
    if keys != list(other._current_field_types.keys()):
        raise ValueError("intersect() requires identical schemas.")
    inner = super().join(other, on=keys, how="inner")
    return self._as_pyspark_df(inner.distinct())

subtract

subtract(other)

Anti join on all columns (schemas must match).

Source code in python/pydantable/pyspark/dataframe.py

def subtract(self, other: DataFrame) -> DataFrame:
    """Anti join on all columns (schemas must match)."""
    keys = list(self._current_field_types.keys())
    if keys != list(other._current_field_types.keys()):
        raise ValueError("subtract() requires identical schemas.")
    return self._as_pyspark_df(super().join(other, on=keys, how="anti"))

except_

except_(other)

Distinct set difference (Spark except / SQL EXCEPT DISTINCT).

Source code in python/pydantable/pyspark/dataframe.py

def except_(self, other: DataFrame) -> DataFrame:
    """Distinct set difference (Spark ``except`` / SQL ``EXCEPT DISTINCT``)."""
    return self.subtract(other).distinct()

exceptAll

exceptAll(other)

Multiset difference (Spark EXCEPT ALL).

Source code in python/pydantable/pyspark/dataframe.py

def exceptAll(self, other: DataFrame) -> DataFrame:
    """Multiset difference (Spark ``EXCEPT ALL``)."""
    if self._current_field_types != other._current_field_types:
        raise ValueError("exceptAll() requires identical schemas.")
    use_streaming = (
        bool(self._engine_streaming_default)
        if self._engine_streaming_default is not None
        else False
    )
    out_data, schema_desc = self._engine.execute_except_all(
        self._rust_plan,
        self._root_data,
        other._rust_plan,
        other._root_data,
        as_python_lists=True,
        streaming=use_streaming,
    )
    derived_fields = self._field_types_from_descriptors(schema_desc)
    derived_schema_type = make_derived_schema_type(
        self._current_schema_type, derived_fields
    )
    rust_plan = self._engine.make_plan(derived_fields)
    return self._as_pyspark_df(
        self._from_plan(
            root_data=out_data,
            root_schema_type=derived_schema_type,
            current_schema_type=derived_schema_type,
            rust_plan=rust_plan,
            engine=self._engine,
        )
    )

intersectAll

intersectAll(other)

Multiset intersection (Spark INTERSECT ALL).

Source code in python/pydantable/pyspark/dataframe.py

def intersectAll(self, other: DataFrame) -> DataFrame:
    """Multiset intersection (Spark ``INTERSECT ALL``)."""
    if self._current_field_types != other._current_field_types:
        raise ValueError("intersectAll() requires identical schemas.")
    use_streaming = (
        bool(self._engine_streaming_default)
        if self._engine_streaming_default is not None
        else False
    )
    out_data, schema_desc = self._engine.execute_intersect_all(
        self._rust_plan,
        self._root_data,
        other._rust_plan,
        other._root_data,
        as_python_lists=True,
        streaming=use_streaming,
    )
    derived_fields = self._field_types_from_descriptors(schema_desc)
    derived_schema_type = make_derived_schema_type(
        self._current_schema_type, derived_fields
    )
    rust_plan = self._engine.make_plan(derived_fields)
    return self._as_pyspark_df(
        self._from_plan(
            root_data=out_data,
            root_schema_type=derived_schema_type,
            current_schema_type=derived_schema_type,
            rust_plan=rust_plan,
            engine=self._engine,
        )
    )

join

join(other, *, on=None, left_on=None, right_on=None, how='inner', suffix='_right', coalesce=None, validate=None, join_nulls=None, maintain_order=None, streaming=None, keepRightJoinKeys=False, keepLeftJoinKeys=False)

Join two frames (Spark-shaped wrapper over core join).

Supports Spark-ish join modes like left_semi/left_anti and right_semi/right_anti (aliases over core semi/anti), plus Spark-style validate shorthands: validate='1:1'|'1:m'|'m:1'|'m:m'.

Source code in python/pydantable/pyspark/dataframe.py

def join(
    self,
    other: CoreDataFrame,
    *,
    on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
    left_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
    right_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
    how: str = "inner",
    suffix: str = "_right",
    coalesce: bool | None = None,
    validate: str | None = None,
    join_nulls: bool | None = None,
    maintain_order: bool | str | None = None,
    streaming: bool | None = None,
    keepRightJoinKeys: bool = False,
    keepLeftJoinKeys: bool = False,
) -> DataFrame:
    """Join two frames (Spark-shaped wrapper over core join).

    Supports Spark-ish join modes like ``left_semi``/``left_anti`` and
    ``right_semi``/``right_anti`` (aliases over core ``semi``/``anti``), plus
    Spark-style validate shorthands: ``validate='1:1'|'1:m'|'m:1'|'m:m'``.
    """
    if not isinstance(keepRightJoinKeys, bool):
        raise TypeError("join(keepRightJoinKeys=...) expects a bool.")
    if not isinstance(keepLeftJoinKeys, bool):
        raise TypeError("join(keepLeftJoinKeys=...) expects a bool.")
    if on is not None and (left_on is not None or right_on is not None):
        raise ValueError(
            "join() use either on=... or left_on=/right_on=..., not both."
        )

    how_norm = str(how).strip().lower()
    how_aliases = {
        "outer": "outer",
        "full_outer": "full",
        "full": "full",
        "right_outer": "right",
        "left_outer": "left",
    }
    how_norm = how_aliases.get(how_norm, how_norm)
    if how_norm == "left_semi":
        how_norm = "semi"
    elif how_norm == "left_anti":
        how_norm = "anti"
    elif how_norm == "right_semi":
        how_norm = "right_semi"
    elif how_norm == "right_anti":
        how_norm = "right_anti"

    supported_hows = {
        "inner",
        "left",
        "right",
        "outer",
        "full",
        "cross",
        "semi",
        "anti",
        "left_semi",
        "left_anti",
        "right_semi",
        "right_anti",
        "left_outer",
        "right_outer",
        "full_outer",
    }
    if how_norm not in {"right_semi", "right_anti"} and how_norm not in {
        "inner",
        "left",
        "right",
        "outer",
        "full",
        "cross",
        "semi",
        "anti",
    }:
        raise ValueError(
            "join(how=...) must be one of: "
            + ", ".join(repr(x) for x in sorted(supported_hows))
        )

    def _resolve_key_arg(
        arg: str | ColumnRef | Sequence[str | ColumnRef] | None, *, arg_name: str
    ) -> str | list[str] | None:
        if arg is None:
            return None
        if isinstance(arg, str):
            return arg
        if isinstance(arg, ColumnRef):
            referenced = arg.referenced_columns()
            if len(referenced) != 1:
                raise TypeError(
                    f"join({arg_name}=...) ColumnRef must reference exactly "
                    f"one column; referenced_columns={sorted(referenced)!r}"
                )
            return next(iter(referenced))
        raw = list(arg)
        out: list[str] = []
        for k in raw:
            if isinstance(k, str):
                out.append(k)
            elif isinstance(k, ColumnRef):
                referenced = k.referenced_columns()
                if len(referenced) != 1:
                    raise TypeError(
                        f"join({arg_name}=...) ColumnRef must reference exactly "
                        f"one column; referenced_columns={sorted(referenced)!r}"
                    )
                out.append(next(iter(referenced)))
            else:
                raise TypeError(
                    f"join({arg_name}=...) expects str, ColumnRef, or a "
                    "sequence of str|ColumnRef."
                )
        if len(set(out)) != len(out):
            raise ValueError(
                f"join({arg_name}=...) must not contain duplicate keys."
            )
        return out

    on_names = _resolve_key_arg(on, arg_name="on")
    left_names = _resolve_key_arg(left_on, arg_name="left_on")
    right_names = _resolve_key_arg(right_on, arg_name="right_on")

    used_on = on_names is not None
    used_lr = left_names is not None or right_names is not None
    if how_norm != "cross":
        if used_on:
            pass
        elif used_lr:
            if left_names is None or right_names is None:
                raise ValueError(
                    "join() requires both left_on=... and right_on=... when on=... "
                    "is not set."
                )
            left_list = (
                [left_names] if isinstance(left_names, str) else list(left_names)
            )
            right_list = (
                [right_names] if isinstance(right_names, str) else list(right_names)
            )
            if len(left_list) != len(right_list):
                raise ValueError(
                    "join(left_on=..., right_on=...) key lists must match length."
                )
        else:
            raise ValueError(
                "join() requires on=... or left_on=.../right_on=... for "
                "non-cross joins."
            )

    if how_norm in ("right_semi", "right_anti"):
        swapped_how = "semi" if how_norm == "right_semi" else "anti"
        if used_on:
            swapped = other.join(
                self,
                on=on_names,
                how=swapped_how,
                suffix=suffix,
                coalesce=coalesce,
                validate=validate,
                join_nulls=join_nulls,
                maintain_order=maintain_order,
                streaming=streaming,
            )
        else:
            # swap sides: keys swap too
            swapped = other.join(
                self,
                left_on=right_names,
                right_on=left_names,
                how=swapped_how,
                suffix=suffix,
                coalesce=coalesce,
                validate=validate,
                join_nulls=join_nulls,
                maintain_order=maintain_order,
                streaming=streaming,
            )
        # right-only output is already guaranteed by semi/anti on swapped join.
        _ = keepLeftJoinKeys  # reserved for future parity knobs
        return self._as_pyspark_df(swapped)

    joined = super().join(
        other,
        on=on_names,
        left_on=left_names,
        right_on=right_names,
        how=how_norm,
        suffix=suffix,
        coalesce=coalesce,
        validate=validate,
        join_nulls=join_nulls,
        maintain_order=maintain_order,
        streaming=streaming,
    )

    # Spark-ish USING behavior: when joining on same-named keys, keep one copy
    # of each join key by default.
    if used_on and on_names is not None and not keepRightJoinKeys:
        # Core join can still produce suffixed duplicates depending on rules;
        # drop any right-side key duplicates if present.
        keys = [on_names] if isinstance(on_names, str) else list(on_names)
        drop_cols: list[str] = []
        for k in keys:
            rk = f"{k}{suffix}"
            if rk in joined._current_field_types:
                drop_cols.append(rk)
        if drop_cols:
            joined = joined.drop(*drop_cols)
    elif (
        used_lr
        and not keepRightJoinKeys
        and left_names is not None
        and right_names is not None
    ):
        left_list = (
            [left_names] if isinstance(left_names, str) else list(left_names)
        )
        right_list = (
            [right_names] if isinstance(right_names, str) else list(right_names)
        )
        drop_cols: list[str] = []
        for lk, rk in zip(left_list, right_list, strict=True):
            if lk == rk:
                cand = f"{rk}{suffix}"
                if cand in joined._current_field_types:
                    drop_cols.append(cand)
        if drop_cols:
            joined = joined.drop(*drop_cols)

    return self._as_pyspark_df(joined)

groupBy

groupBy(*keys, maintain_order=False, drop_nulls=True)

Spark alias of :meth:group_by.

Source code in python/pydantable/pyspark/dataframe.py

def groupBy(
    self,
    *keys: str | ColumnRef,
    maintain_order: bool = False,
    drop_nulls: bool = True,
) -> PySparkGroupedDataFrame:
    """Spark alias of :meth:`group_by`."""
    return self.group_by(
        *keys, maintain_order=maintain_order, drop_nulls=drop_nulls
    )

sort

sort(*columns, ascending=None)

Sort by column names (Spark sort / orderBy).

Source code in python/pydantable/pyspark/dataframe.py

def sort(
    self,
    *columns: str,
    ascending: bool | list[bool] | None = None,
) -> DataFrame:
    """Sort by column names (Spark ``sort`` / ``orderBy``)."""
    return self.orderBy(*columns, ascending=ascending)

crossJoin

crossJoin(other)

Cross join (Spark crossJoin).

Source code in python/pydantable/pyspark/dataframe.py

def crossJoin(self, other: DataFrame) -> DataFrame:
    """Cross join (Spark ``crossJoin``)."""
    return self._as_pyspark_df(super().join(other, how="cross"))

count

count()

Row count via global_row_count() (Spark-style action).

Source code in python/pydantable/pyspark/dataframe.py

def count(self) -> int:
    """Row count via ``global_row_count()`` (Spark-style action)."""
    d = self.select(global_row_count()).to_dict()
    col = next(iter(d.keys()))
    return int(d[col][0])

fillna

fillna(value=None, *, subset=None)

Spark name for :meth:fill_null.

Source code in python/pydantable/pyspark/dataframe.py

def fillna(
    self,
    value: Any = None,
    *,
    subset: str | list[str] | None = None,
) -> DataFrame:
    """Spark name for :meth:`fill_null`."""
    if isinstance(subset, str):
        sub: str | list[str] | None = subset
    elif subset is None:
        sub = None
    else:
        sub = list(subset)
    return self._as_pyspark_df(self.fill_null(value, subset=sub))

dropna

dropna(how='any', *, thresh=None, subset=None)

Spark name for :meth:drop_nulls (thresh maps to threshold).

Source code in python/pydantable/pyspark/dataframe.py

def dropna(
    self,
    how: str = "any",
    *,
    thresh: int | None = None,
    subset: str | list[str] | None = None,
) -> DataFrame:
    """Spark name for :meth:`drop_nulls` (``thresh`` maps to ``threshold``)."""
    if isinstance(subset, str):
        sub: str | list[str] | None = subset
    elif subset is None:
        sub = None
    else:
        sub = list(subset)
    return self._as_pyspark_df(
        self.drop_nulls(sub, how=how, threshold=thresh),
    )

printSchema

printSchema(level=None)

Print schema tree (Spark printSchema; level accepted for parity).

Source code in python/pydantable/pyspark/dataframe.py

def printSchema(self, level: int | None = None) -> None:
    """Print schema tree (Spark ``printSchema``; ``level`` accepted for parity)."""
    _ = level
    print("root")
    for line in _format_schema_type_lines(self.schema, ""):
        print(line)

explain

explain(extended=None, mode=None, *, format='text')

Print logical plan (Spark explain; extended / mode reserved).

Source code in python/pydantable/pyspark/dataframe.py

def explain(
    self,
    extended: bool | None = None,
    mode: str | None = None,
    *,
    format: TypingLiteral["text", "json"] = "text",
) -> None:
    """Print logical plan (Spark ``explain``; ``extended`` / ``mode`` reserved)."""
    _ = extended
    _ = mode
    out = super().explain(format=format)
    if isinstance(out, dict):
        import json

        print(json.dumps(out, indent=2, sort_keys=True))
    else:
        print(out)

describe

describe()

Same string as core :meth:~pydantable.dataframe.DataFrame.describe.

Source code in python/pydantable/pyspark/dataframe.py

def describe(self) -> str:
    """Same string as core :meth:`~pydantable.dataframe.DataFrame.describe`."""
    return super().describe()

toPandas

toPandas()

Materialize to a pandas.DataFrame (requires pandas installed).

Source code in python/pydantable/pyspark/dataframe.py

def toPandas(self):
    """Materialize to a ``pandas.DataFrame`` (requires ``pandas`` installed)."""
    try:
        import pandas as pd
    except ImportError as e:
        raise ImportError("toPandas() requires pandas (pip install pandas).") from e
    return pd.DataFrame(self.to_dict())

show

show(n=20, truncate=True, vertical=False)

Print up to n rows (via :meth:head + :meth:to_dict).

Cost: bounded materialization. Not a distributed Spark runtime.

Source code in python/pydantable/pyspark/dataframe.py

def show(
    self,
    n: int = 20,
    truncate: bool = True,
    vertical: bool = False,
) -> None:
    """Print up to ``n`` rows (via :meth:`head` + :meth:`to_dict`).

    **Cost:** bounded materialization. Not a distributed Spark runtime.
    """
    h = self.head(int(n))
    data = h.to_dict()
    if vertical:
        cols = list(data.keys())
        if not cols:
            print("(empty)")
            return
        nrows = len(next(iter(data.values())))
        for i in range(nrows):
            print(f"- record {i}")
            for c in cols:
                print(f"  {c}: {data[c][i]!r}")
        return
    print(_text_show_table(data, truncate=truncate))

summary

summary()

Spark-style name for :meth:describe (numeric columns; materializes).

Source code in python/pydantable/pyspark/dataframe.py

def summary(self) -> str:
    """Spark-style name for :meth:`describe` (numeric columns; materializes)."""
    return self.describe()

DataFrameModel

Bases: DataFrameModel

Class-based container using :class:DataFrame for Spark-shaped methods.

Source code in python/pydantable/pyspark/dataframe.py

class DataFrameModel(CoreDataFrameModel):
    """Class-based container using :class:`DataFrame` for Spark-shaped methods."""

    _dataframe_cls = DataFrame

    def withColumn(self, name: str, col: Any) -> DataFrameModel:
        return cast(
            "DataFrameModel", self._from_dataframe(self._df.withColumn(name, col))
        )

    def withColumns(self, colsMap: Mapping[str, Any]) -> DataFrameModel:
        return cast(
            "DataFrameModel", self._from_dataframe(self._df.withColumns(colsMap))
        )

    def withColumnRenamed(self, existing: str, new: str) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.withColumnRenamed(existing, new)),
        )

    def withColumnsRenamed(self, colsMap: Mapping[str, str]) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.withColumnsRenamed(colsMap)),
        )

    def toDF(self, *cols: str) -> DataFrameModel:
        return cast("DataFrameModel", self._from_dataframe(self._df.toDF(*cols)))

    def transform(
        self,
        func: Callable[..., CoreDataFrameModel],
        *args: Any,
        **kwargs: Any,
    ) -> DataFrameModel:
        out = func(self, *args, **kwargs)
        if not isinstance(out, CoreDataFrameModel):
            raise TypeError(
                "transform(func, ...) expects func to return a DataFrameModel."
            )
        return cast("DataFrameModel", self._from_dataframe(out._df))

    def select_typed(self, *cols: str, **named_exprs: Any) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.select_typed(*cols, **named_exprs)),
        )

    def where(self, condition: Any) -> DataFrameModel:
        return cast("DataFrameModel", self._from_dataframe(self._df.where(condition)))

    def filter(self, condition: Any) -> DataFrameModel:
        return cast("DataFrameModel", self._from_dataframe(self._df.filter(condition)))

    def select(self, *cols: str | ColumnRef | Expr, **named: Any) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.select(*cols, **named)),
        )

    def orderBy(
        self,
        *columns: str,
        ascending: bool | list[bool] | None = None,
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.orderBy(*columns, ascending=ascending)),
        )

    def limit(self, num: int) -> DataFrameModel:
        return cast("DataFrameModel", self._from_dataframe(self._df.limit(num)))

    def sample(
        self,
        withReplacement: bool | None = None,
        fraction: float | None = None,
        seed: int | None = None,
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.sample(
                    withReplacement=withReplacement,
                    fraction=fraction,
                    seed=seed,
                )
            ),
        )

    def explode(
        self,
        columns: Any,
        *,
        outer: bool = False,
        streaming: bool | None = None,
    ) -> DataFrameModel:
        """Explode **list** columns (Spark ``explode``).

        Use ``outer=True`` for ``explode_outer``.
        """
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.explode(columns, outer=outer, streaming=streaming)
            ),
        )

    def explode_outer(
        self, columns: Any, *, streaming: bool | None = None
    ) -> DataFrameModel:
        """Explode lists with Spark-ish *outer* null/empty handling (see docs)."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.explode_outer(columns, streaming=streaming)),
        )

    def explode_all(self, *, streaming: bool | None = None) -> DataFrameModel:
        """Explode every list-typed column in the schema (schema-driven convenience)."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.explode_all(streaming=streaming)),
        )

    def posexplode(
        self,
        column: str,
        *,
        pos: str = "pos",
        value: str | None = None,
        outer: bool = False,
        streaming: bool | None = None,
    ) -> DataFrameModel:
        """Explode one list with **0-based** positions (Spark ``posexplode``)."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.posexplode(
                    column, pos=pos, value=value, outer=outer, streaming=streaming
                )
            ),
        )

    def posexplode_outer(
        self,
        column: str,
        *,
        pos: str = "pos",
        value: str | None = None,
        streaming: bool | None = None,
    ) -> DataFrameModel:
        """``posexplode(..., outer=True)`` alias."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.posexplode_outer(
                    column, pos=pos, value=value, streaming=streaming
                )
            ),
        )

    def unnest(self, columns: Any, *, streaming: bool | None = None) -> DataFrameModel:
        """Expand **struct** columns to top-level fields (Spark analogue)."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.unnest(columns, streaming=streaming)),
        )

    def unnest_all(self, *, streaming: bool | None = None) -> DataFrameModel:
        """Unnest every struct-typed column in the schema."""
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.unnest_all(streaming=streaming)),
        )

    def drop(self, *cols: str | ColumnRef) -> DataFrameModel:
        return cast("DataFrameModel", self._from_dataframe(self._df.drop(*cols)))

    def distinct(
        self,
        subset: Sequence[str] | None = None,
        *,
        keep: str = "first",
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.distinct(subset=subset, keep=keep)),
        )

    def dropDuplicates(self, subset: list[str] | None = None) -> DataFrameModel:
        return (
            self.distinct(subset=subset, keep="first")
            if subset is not None
            else self.distinct(keep="first")
        )

    def union(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast("DataFrameModel", self._from_dataframe(self._df.union(od)))

    def unionAll(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        return self.union(other)

    def unionByName(
        self,
        other: DataFrameModel | DataFrame,
        *,
        allowMissingColumns: bool = False,
    ) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.unionByName(od, allowMissingColumns=allowMissingColumns)
            ),
        )

    def join(
        self,
        other: DataFrameModel | DataFrame,
        *,
        on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        left_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        right_on: str | ColumnRef | Sequence[str | ColumnRef] | None = None,
        how: str = "inner",
        suffix: str = "_right",
        coalesce: bool | None = None,
        validate: str | None = None,
        join_nulls: bool | None = None,
        maintain_order: bool | str | None = None,
        streaming: bool | None = None,
        keepRightJoinKeys: bool = False,
        keepLeftJoinKeys: bool = False,
    ) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.join(
                    od,
                    on=on,
                    left_on=left_on,
                    right_on=right_on,
                    how=how,
                    suffix=suffix,
                    coalesce=coalesce,
                    validate=validate,
                    join_nulls=join_nulls,
                    maintain_order=maintain_order,
                    streaming=streaming,
                    keepRightJoinKeys=keepRightJoinKeys,
                    keepLeftJoinKeys=keepLeftJoinKeys,
                )
            ),
        )

    def intersect(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.intersect(od)),
        )

    def subtract(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.subtract(od)),
        )

    def except_(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        """Distinct set difference (Spark ``except`` / SQL ``EXCEPT DISTINCT``)."""
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast("DataFrameModel", self._from_dataframe(self._df.except_(od)))

    def exceptAll(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.exceptAll(od)),
        )

    def intersectAll(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.intersectAll(od)),
        )

    def group_by(
        self,
        *keys: Any,
        maintain_order: bool = False,
        drop_nulls: bool = True,
    ) -> PySparkGroupedDataFrameModel:
        inner = self._df.group_by(
            *keys, maintain_order=maintain_order, drop_nulls=drop_nulls
        )
        if not isinstance(inner, PySparkGroupedDataFrame):
            raise TypeError("group_by did not return PySparkGroupedDataFrame.")
        return PySparkGroupedDataFrameModel(inner, type(self))

    def groupBy(
        self,
        *keys: Any,
        maintain_order: bool = False,
        drop_nulls: bool = True,
    ) -> PySparkGroupedDataFrameModel:
        return self.group_by(
            *keys, maintain_order=maintain_order, drop_nulls=drop_nulls
        )

    def sort(
        self,
        *columns: str,
        ascending: bool | list[bool] | None = None,
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.sort(*columns, ascending=ascending)),
        )

    def crossJoin(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
        od = other._df if isinstance(other, DataFrameModel) else other
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.crossJoin(od)),
        )

    def count(self) -> int:
        return self._df.count()

    def fillna(
        self,
        value: Any = None,
        *,
        subset: str | list[str] | None = None,
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(self._df.fillna(value, subset=subset)),
        )

    def dropna(
        self,
        how: str = "any",
        *,
        thresh: int | None = None,
        subset: str | list[str] | None = None,
    ) -> DataFrameModel:
        return cast(
            "DataFrameModel",
            self._from_dataframe(
                self._df.dropna(how=how, thresh=thresh, subset=subset),
            ),
        )

    @property
    def na(self) -> DataFrameModelNaFunctions:
        return DataFrameModelNaFunctions(self)

    def printSchema(self, level: int | None = None) -> None:
        self._df.printSchema(level=level)

    def explain(
        self,
        extended: bool | None = None,
        mode: str | None = None,
        *,
        format: TypingLiteral["text", "json"] = "text",
    ) -> None:
        self._df.explain(extended=extended, mode=mode, format=format)

    def describe(self) -> str:
        return self._df.describe()

    def toPandas(self):
        return self._df.toPandas()

    def show(
        self,
        n: int = 20,
        truncate: bool = True,
        vertical: bool = False,
    ) -> None:
        """Print rows (delegates to :class:`DataFrame`)."""
        self._df.show(n=n, truncate=truncate, vertical=vertical)

    def summary(self) -> str:
        return self._df.summary()

    @property
    def schema(self) -> StructType:
        return self._df.schema

    def __getitem__(self, key: str | list[str]) -> Any:
        return self._df[key]  # type: ignore[index]

explode

explode(columns, *, outer=False, streaming=None)

Explode list columns (Spark explode).

Use outer=True for explode_outer.

Source code in python/pydantable/pyspark/dataframe.py

def explode(
    self,
    columns: Any,
    *,
    outer: bool = False,
    streaming: bool | None = None,
) -> DataFrameModel:
    """Explode **list** columns (Spark ``explode``).

    Use ``outer=True`` for ``explode_outer``.
    """
    return cast(
        "DataFrameModel",
        self._from_dataframe(
            self._df.explode(columns, outer=outer, streaming=streaming)
        ),
    )

explode_outer

explode_outer(columns, *, streaming=None)

Explode lists with Spark-ish outer null/empty handling (see docs).

Source code in python/pydantable/pyspark/dataframe.py

def explode_outer(
    self, columns: Any, *, streaming: bool | None = None
) -> DataFrameModel:
    """Explode lists with Spark-ish *outer* null/empty handling (see docs)."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(self._df.explode_outer(columns, streaming=streaming)),
    )

explode_all

explode_all(*, streaming=None)

Explode every list-typed column in the schema (schema-driven convenience).

Source code in python/pydantable/pyspark/dataframe.py

def explode_all(self, *, streaming: bool | None = None) -> DataFrameModel:
    """Explode every list-typed column in the schema (schema-driven convenience)."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(self._df.explode_all(streaming=streaming)),
    )

posexplode

posexplode(column, *, pos='pos', value=None, outer=False, streaming=None)

Explode one list with 0-based positions (Spark posexplode).

Source code in python/pydantable/pyspark/dataframe.py

def posexplode(
    self,
    column: str,
    *,
    pos: str = "pos",
    value: str | None = None,
    outer: bool = False,
    streaming: bool | None = None,
) -> DataFrameModel:
    """Explode one list with **0-based** positions (Spark ``posexplode``)."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(
            self._df.posexplode(
                column, pos=pos, value=value, outer=outer, streaming=streaming
            )
        ),
    )

posexplode_outer

posexplode_outer(column, *, pos='pos', value=None, streaming=None)

posexplode(..., outer=True) alias.

Source code in python/pydantable/pyspark/dataframe.py

def posexplode_outer(
    self,
    column: str,
    *,
    pos: str = "pos",
    value: str | None = None,
    streaming: bool | None = None,
) -> DataFrameModel:
    """``posexplode(..., outer=True)`` alias."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(
            self._df.posexplode_outer(
                column, pos=pos, value=value, streaming=streaming
            )
        ),
    )

unnest

unnest(columns, *, streaming=None)

Expand struct columns to top-level fields (Spark analogue).

Source code in python/pydantable/pyspark/dataframe.py

def unnest(self, columns: Any, *, streaming: bool | None = None) -> DataFrameModel:
    """Expand **struct** columns to top-level fields (Spark analogue)."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(self._df.unnest(columns, streaming=streaming)),
    )

unnest_all

unnest_all(*, streaming=None)

Unnest every struct-typed column in the schema.

Source code in python/pydantable/pyspark/dataframe.py

def unnest_all(self, *, streaming: bool | None = None) -> DataFrameModel:
    """Unnest every struct-typed column in the schema."""
    return cast(
        "DataFrameModel",
        self._from_dataframe(self._df.unnest_all(streaming=streaming)),
    )

except_

except_(other)

Distinct set difference (Spark except / SQL EXCEPT DISTINCT).

Source code in python/pydantable/pyspark/dataframe.py

def except_(self, other: DataFrameModel | DataFrame) -> DataFrameModel:
    """Distinct set difference (Spark ``except`` / SQL ``EXCEPT DISTINCT``)."""
    od = other._df if isinstance(other, DataFrameModel) else other
    return cast("DataFrameModel", self._from_dataframe(self._df.except_(od)))

show

show(n=20, truncate=True, vertical=False)

Print rows (delegates to :class:DataFrame).

Source code in python/pydantable/pyspark/dataframe.py

def show(
    self,
    n: int = 20,
    truncate: bool = True,
    vertical: bool = False,
) -> None:
    """Print rows (delegates to :class:`DataFrame`)."""
    self._df.show(n=n, truncate=truncate, vertical=vertical)

SqlDataFrame

Bases: SqlDataFrame, DataFrame

Lazy-SQL backend plus PySpark-shaped API (withColumn, orderBy, …).

Source code in python/pydantable/pyspark/sql_dataframe.py

class SqlDataFrame(CoreSqlDataFrame, PySparkDataFrame):
    """Lazy-SQL backend plus PySpark-shaped API (``withColumn``, ``orderBy``, …)."""

    @staticmethod
    def _as_pyspark_df(df: CoreDataFrame) -> SqlDataFrame:
        return cast(
            "SqlDataFrame",
            SqlDataFrame._from_plan(
                root_data=df._root_data,
                root_schema_type=df._root_schema_type,
                current_schema_type=df._current_schema_type,
                rust_plan=df._rust_plan,
                engine=df._engine,
            ),
        )

    def toDF(self, *cols: str) -> SqlDataFrame:
        current = list(self.schema_fields().keys())
        if len(cols) != len(current):
            raise ValueError(f"toDF() expects {len(current)} names, got {len(cols)}.")
        mapping = dict(zip(current, cols, strict=True))
        renamed = self.rename(mapping)
        ordered_types = {name: renamed._current_field_types[name] for name in cols}
        ordered_schema = make_derived_schema_type(
            renamed._root_schema_type, ordered_types
        )
        return self._as_pyspark_df(
            type(self)._from_plan(
                root_data=renamed._root_data,
                root_schema_type=renamed._root_schema_type,
                current_schema_type=ordered_schema,
                rust_plan=renamed._rust_plan,
                engine=self._engine,
            ),
        )

SqlDataFrameModel

Bases: SqlDataFrameModel, DataFrameModel

Lazy-SQL backend plus PySpark-shaped :class:DataFrameModel methods.

Source code in python/pydantable/pyspark/sql_dataframe.py

class SqlDataFrameModel(CoreSqlDataFrameModel, PySparkDataFrameModel):
    """Lazy-SQL backend plus PySpark-shaped :class:`DataFrameModel` methods."""

    _dataframe_cls = SqlDataFrame

    @classmethod
    def concat(
        cls,
        dfs: Sequence[DataFrameModel[Any]],
        *,
        how: str = "vertical",
    ) -> DataFrameModel[Any]:
        if len(dfs) < 2:
            raise ValueError("concat() requires at least two DataFrameModel inputs.")
        if not all(isinstance(df, DataFrameModel) for df in dfs):
            raise TypeError("concat() expects a sequence of DataFrameModel objects.")
        return cls._from_dataframe(
            SqlDataFrame.concat([df._df for df in dfs], how=how),
        )