latu

# Using Latu

Latu is a Spark Connect client. It builds a plan locally and the cluster runs it. This file is
the short set of rules that are not guessable from the function names, in the `usage_rules`
convention, so a consuming project can sync it into an agent's context.

Every deliberate departure from PySpark is in `docs/deviations.md`, with why.

## The two shapes

**A builder takes a DataFrame and returns a DataFrame.** It is pure, does no IO, and nothing
reaches the server: `select`, `filter`, `join`, `group_by`, `with_columns`, `sort`, all of them.

**An action runs the query** and returns `{:ok, value} | {:error, %Latu.Error{}}`, with a `!`
twin that raises instead. `collect`, `count`, `show`, `write`, `schema`.

**Every action that can fail has a `!` twin, and a test enforces it.** If you can call `foo/2`,
you can call `foo!/2`. Guessing works.

    {:ok, rows} = Latu.collect(df)
    rows = Latu.collect!(df)

## Coercion: what an atom, a string and a function mean

The same value means different things in different positions.

- **An atom is a column reference.** `:price` is the column named `price`.
- **A string is a column *name* where a name is expected, and a *literal* inside an
  expression.** `Latu.select(df, "price")` selects the column; `F.upper("price")` upcases the
  five-character string.
- **`Latu.Column.col/1`** is the explicit column reference, and **`expr/1`** takes SQL anywhere
  an expression goes: `Latu.select(df, [:id, big: expr("price > 100")])`. In `filter/2` a bare
  string is already SQL.
- **A keyword list projects and names**: `Latu.select(df, total: F.sum(:price))`.

There is no macro DSL and no operator overloading. `Latu.Column.greater(:a, :b)`, not `a > b`.

## Import and alias discipline

    alias Latu.Functions, as: F
    alias Latu.Window, as: W
    import Latu.Column

**Call `Latu` qualified**, the way `Enum` is: `Latu.filter`, `Latu.show`. There is no
`Latu.inspect` and no `Latu.alias`. `Latu.as/2` is the alias.

`lit`, `col`, `expr` and `star` are in `Latu.Column`, **not** `Latu.Functions`. A call to
`F.col/1` is the single most common mistake.

## observe: the metrics come back from the action

`Latu.observe/3` asks the server for metrics alongside the result. A `%Latu.DataFrame{}` is
inert and Latu holds no processes, so the metrics come back **from the action**, through a
`*_with_metrics` twin:

    df = Latu.observe(df, :checks, total: F.count(:id))
    {:ok, rows, info} = Latu.collect_with_metrics(df)
    info.observed  #=> %{checks: %{total: 8}}

Every other action runs the observed frame and does not report: `show`, `collect`, a `join`.
PySpark behaves the same way when nobody reads the `Observation`. The one exception is a plain
**write** (`write`, `save_as_table`, `insert_into`, `write_v2`, `merge`), which **raises** and
names the twin. A write consumes the frame, so its metrics would be produced and dropped.

## Bound the plan, not the action

`Latu.collect/2`, `Latu.to_explorer/2`, `Latu.to_arrow/2` and `Latu.to_nx/2` all bring the
**whole** result back, and none of them takes a row limit. Spark's `collect` takes no arguments
either. To take part of a result, bound the plan:
`df |> Latu.limit(10_000) |> Latu.to_explorer()`, or `Latu.take(df, 10_000)` for rows. For a
result too large to hold, `Latu.stream/2` gives one `Explorer.DataFrame` per Arrow batch and
`Latu.stream_nx/2` one map of tensors.

## Errors tell you what went wrong

A `%Latu.Error{}` from the server carries Spark's own structured detail, with no extra call:

    {:error, error} = Latu.collect(df)

    error.kind         #=> :rpc, Spark refused; :protocol means the server answered out of shape
    error.error_class  #=> "UNRESOLVED_COLUMN.WITH_SUGGESTION"
    error.sql_state    #=> "42703"
    error.classes      #=> ["org.apache.spark.sql.AnalysisException", ...]
    error.parameters   #=> %{"objectName" => "`nope`", ...}

**Match on `error_class`, not on the message.** `Latu.error_details/2` fetches the full cause
chain when you need it; it is a round trip, so it is a call rather than automatic.

## Session config: three reads, and they are not the same read

    Latu.conf!(session, "spark.sql.shuffle.partitions")        # else Spark's default, else nil
    Latu.fetch_conf!(session, "nope")                            # else Spark's default, else ERROR
    Latu.conf!(session, "spark.sql.shuffle.partitions", "200")   # else YOUR default

`conf/2` is `Map.get`, `fetch_conf/2` is `Map.fetch`. **`conf/3` overrides Spark's own default;
it does not fall back to it.** For a conf Spark defines but nobody set, `conf/2` and
`fetch_conf/2` give you Spark's default and `conf/3` gives you yours. Spark type-checks your
default against the conf.

`set_conf/3`, `set_confs/2` and `unset_conf/2` return `:ok`: the conf lives on the server, the
session struct is unchanged. Values may be a string, number, boolean or atom.

Two more worth knowing:

- **`confs/1` is only what the session has *set*.** A conf at its default is absent from it
  while `conf/2` answers for it.
- **`is_modifiable/2` returning false does not mean `set_conf/3` will fail.** It is false for
  every key Spark does not define, and those are stored happily. What it reliably catches is a
  *static* conf, which is refused.

The session also carries the tuning knobs, all `Latu.connect/2` options: `:retry`
(a `Latu.Retry`, defaulted to PySpark's own policy), `:window_size`, `:keepalive` and
`:keepalive_tolerance`.

## Check your work without a server

**`Latu.Plan` is public and pure**, and a session that was never connected still builds plans.
So a pipeline can be checked for nothing:

    session = Latu.Session.from_url!("sc://localhost:15002")
    df = session |> Latu.range(10) |> Latu.filter(Latu.Column.greater(:id, 3))

    inspect(df)  #=> "#Latu.DataFrame<range → filter>"

That is how Latu tests itself: every golden test compares a locally built plan against
PySpark's bytes. It is also the cheapest feedback loop available when generating Latu code.
Build it, inspect it, then run it.

## Naming

**Spark > Elixir > Polars > dplyr.** Latu uses Spark's own spelling wherever Spark has one, so
if you know the PySpark name, snake_case it and you are usually right. Where Elixir forbids it,
the deviation is recorded:

- `df.alias("a")` → `Latu.as/2` (`alias` is a special form)
- `df.show()` → `Latu.show/2`, returning `:ok` like `File.write/2`
- `col != other` → `Latu.Column.not_equal/2`

## Things Latu deliberately does not do

- **It adds nothing to your supervision tree.** No GenServer, no pool, no application callback
  module. A session is a plain struct; nothing is supervised, nothing is mutated. Two things it
  does hold: the **gRPC channel is a process** (`connect/2` opens it, `disconnect/2` closes it),
  and a **checkpoint is a server-side resource** with `release/1` to free it.
- **It has no client-side type model.** A schema is Spark's own `simpleString`, and a
  `DataType` comes from `Latu.parse_ddl_type/2`, which the server parses.
- **No UDFs in Elixir, and Latu ships no Elixir code to the server.** Spark Connect has no path
  for client-side code. A function already registered on the session is callable by name with
  `Latu.Column.fun("my_udf", [:price])`, whether a SQL UDF, a Hive UDF or a Java class put it
  there. `CREATE FUNCTION` through `Latu.sql/3` registers one, and `Latu.add_jar/3` puts the jar
  there in the first place: bytes you hand it, under a name, session-scoped. Otherwise SQL
  expressions, and the ~500 built-ins in `Latu.Functions`.
- **Structured streaming and MLlib are separate packages, for different reasons.** The test for
  streaming is whether a resource can be honestly bracketed. A checkpoint can
  (`with_checkpoint/3`); a streaming query cannot, because it runs after you stop looking. MLlib
  is separate on surface instead, and it ships as [`latu_ml`](https://hexdocs.pm/latu_ml): a
  server-side model cache, Spark's on-disk model format and its own operator registry.

## Observability

Latu emits `:telemetry` events, and `Latu.Telemetry`'s moduledoc is the list. Two that catch
people:

- **`[:latu, :rpc, :stop]` for `ExecutePlan` measures opening the stream, not draining it**,
  because a result is a lazy stream. For how long a query took, use
  `[:latu, :execute, :stop]`.
- **`[:latu, :execute, :stop]` has an `:abandoned` outcome**, for a stream the caller stopped
  reading. Worth alerting on: the execution keeps running on the server until it times out.

A handler runs in your own process, the one talking to Spark, so a slow handler slows the
query. Metadata is ids only, and the session's token is never in it.

## Stopping a query

`Latu.interrupt/2`, not killing the process. An execution is *reattachable*, so a killed client
leaves the query running on the server until it times out.

    session = Latu.Session.add_tag(session, "report")
    task = Task.async(fn -> df |> Latu.count() end)
    # ... once status/2 shows it running; an interrupt that arrives first matches nothing
    Latu.interrupt(session, tag: "report")
    Task.await(task)