# `Latu`
[🔗](https://github.com/zero-one-group/latu/blob/v0.8.0/lib/latu.ex#L1)

A native Elixir DataFrame API for Apache Spark, over Spark Connect.

**Latu adds nothing to your supervision tree.** It defines no GenServer, Agent, Supervisor,
Registry or pool, and its `mix.exs` declares no application callback module, so adding it
to your dependencies starts nothing at all. `Latu.Session` is a plain struct; where it
lives is your application's business, not Latu's.

    {:ok, session} = Latu.connect("sc://localhost:15002")
    Latu.disconnect(session)

Two things it does hold, both in structs you keep: **the channel is a process** — `connect/2`
opens it, `disconnect/2` closes it, and an execution's response stream is owned by the
connection, not by whoever reads it: reading it to the end or stopping early both release it,
server side and locally, but a consumer whose process is *killed* unwinds nothing, so that
leaves the execution running until `interrupt/2` or the server's timeout, and its buffer with
it — and **a checkpoint is a server-side resource**, freed by `release/1` or scoped by
`with_checkpoint/3`, because there is no finalizer to free it for you. Latu hands out
resources and never keeps them; nothing is tracked between calls.

A streaming query is the third: `write_stream/2` hands back a `%Latu.StreamingQuery{}`, a
server-side object addressed by id, stopped by `Latu.StreamingQuery.stop/1` or scoped by
`with_stream/3`. Nothing here owns one; a supervised query is a recipe, not a module. MLlib is
separate on different grounds, as `docs/decisions.md` has it.

This module is the verbs — plus the few expression builders that take a DataFrame rather than
a column: `col/2` for a tagged reference, and `scalar/1` and `exists/1` for a subquery over
another frame. Everything else expression-shaped comes from three modules:

    import Latu.Column              # operators, predicates, casts, sort keys, over/2
    alias Latu.Functions, as: F     # Spark's ~500 functions, under Spark's own names
    alias Latu.Window, as: W        # window specifications

The catalog — databases, tables and views — lives on `Latu.Catalog`, as it lives on
`spark.catalog` in PySpark. Caching a *table* by name is there too
(`Latu.Catalog.cache_table/2`); caching the frame in your hand is `cache/1` here.

`Latu.Column` is small and gets composed by hand, so it is imported. This module is called
qualified — `Latu.filter`, `Latu.show` — the way `Enum` is; nothing in it collides with
`Latu.Column` or with `Kernel`, so `import Latu` compiles if a REPL wants it, but the verbs
read better with their module on. The other two are aliased: their names collide with the
verbs on purpose, exactly as Spark's do (`Latu.count/1` and `F.count/1` are different
things), and `F.` is how you find one of five hundred functions with tab completion.

# `add_jar`

```elixir
@spec add_jar(Latu.Session.t(), String.t(), binary()) ::
  :ok | {:error, Latu.Error.t()}
```

Put a jar on the session, so a class in it resolves by name.

    :ok = Latu.add_jar(session, "udfs.jar", File.read!("priv/udfs.jar"))
    {:ok, _} = Latu.sql(session, "CREATE FUNCTION my_udf AS 'com.example.MyUdf'")
    Latu.select(df, [Latu.Column.fun("my_udf", [:price])])

Latu still ships no code of its own and does no local file IO: a jar is bytes you hand it,
under a name. What the server does with them is `sparkContext.addJar`, so whatever the jar
registers is then callable the way any other registered function is — by name through
`Latu.Column.fun/3`, or through `CREATE FUNCTION` and `sql/3`.

`name` is the artifact's identity rather than a hash of its bytes. Re-sending **identical**
bytes under a name the session already holds is a no-op; different bytes under that name are
refused, and a jar cannot be replaced in a live session — a new name is the only way to a new
version. `Latu.sql(session, "LIST JARS")` is what a session holds.

Session-scoped, and gone when the session ends. It reaches the driver's classloader, which is
what resolving a function name needs; code that has to run in a **task** wants the jar on the
cluster's own classpath instead.

# `add_jar!`

```elixir
@spec add_jar!(Latu.Session.t(), String.t(), binary()) :: :ok
```

Like `add_jar/3`, raising on failure.

# `agg`

```elixir
@spec agg(Latu.GroupedData.t() | Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Apply aggregates, giving a DataFrame back.

    df |> Latu.group_by(:suburb) |> Latu.agg(total: F.sum(:price))
    Latu.agg(df, total: F.sum(:price))

Takes a grouped frame or a plain DataFrame; without grouping it aggregates the whole thing.

# `approx_quantile`

```elixir
@spec approx_quantile(
  Latu.DataFrame.t(),
  [String.t() | atom()] | String.t() | atom(),
  [number()],
  number(),
  keyword()
) :: {:ok, [float()] | [[float()]]} | {:error, Latu.Error.t()}
```

Approximate quantiles. An action: it runs when called.

    Latu.approx_quantile(df, :score, [0.0, 0.5, 1.0], 0.01)
    #=> {:ok, [10.0, 30.0, 70.0]}

    Latu.approx_quantile(df, [:score, :weight], [0.5], 0.01)
    #=> {:ok, [[30.0], [3.0]]}

One name gives one flat list, a list of names gives a list per column — PySpark's asymmetry,
kept because it is the shape callers expect. `relative_error` is the accuracy Spark may trade
away for speed; 0.0 asks for exact quantiles, which is expensive. **Nulls and NaNs are
ignored here** — the opposite of `cov/4` and `corr/4`, and Spark's own documented rule — and a
column with no values left gives an empty list.

## Options

  * `:progress` — as `collect/2` describes. Defaults to `nil`.

# `approx_quantile!`

```elixir
@spec approx_quantile!(
  Latu.DataFrame.t(),
  [String.t() | atom()] | String.t() | atom(),
  [number()],
  number(),
  keyword()
) :: [float()] | [[float()]]
```

Like `approx_quantile/5`, raising on failure.

# `as`

```elixir
@spec as(Latu.DataFrame.t(), String.t() | atom()) :: Latu.DataFrame.t()
```

Name the DataFrame, so its columns can be qualified as `name.column`.

Spark calls this `alias`, which Elixir cannot use as a function name.

# `cache`

```elixir
@spec cache(Latu.DataFrame.t()) ::
  {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

`persist/2` at Spark's default level, which is what `cache` means everywhere.

# `cache!`

```elixir
@spec cache!(Latu.DataFrame.t()) :: Latu.DataFrame.t()
```

Like `cache/1`, raising on failure. Returns the DataFrame, so it pipes.

# `checkpoint`

```elixir
@spec checkpoint(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

Materialise the frame on the server and hand back a frame that reads the result.

    {:ok, base} = Latu.checkpoint(expensive)
    # ... many queries over `base`, each skipping the work above it ...
    :ok = Latu.release(base)

Cuts the plan: everything above the checkpoint is computed once.

## Options

  * `:eager` — do the work now rather than at the next action. Defaults to `true`.
  * `:local` — use executor storage instead of a reliable location. Defaults to `false`.
    Faster and needs no checkpoint directory, but the data dies with the executor.
  * `:storage_level` — a name `storage_level/1` knows: `:none`, `:disk_only`, `:disk_only_2`,
    `:disk_only_3`, `:memory_only`, `:memory_only_2`, `:memory_and_disk`,
    `:memory_and_disk_2`, `:memory_and_disk_deser`, `:off_heap`. Defaults to `nil`.
    **Accepted with `local: true` only** — the server reads the level on that branch alone,
    so passing one without it is refused rather than dropped in silence.

## Examples

    {:ok, base} = Latu.checkpoint(expensive)
    {:ok, base} = Latu.checkpoint(expensive, local: true)
    {:ok, base} = Latu.checkpoint(expensive, local: true, storage_level: :memory_only)
    {:ok, base} = Latu.checkpoint(expensive, eager: false)

**The reliable form needs a checkpoint directory on the server**, and the command carries no
path: the directory is `spark.checkpoint.dir`, read once at startup (`SparkContext`'s own
setter is not reachable over Connect). Without it the server answers "Checkpoint directory
has not been set". `local: true` needs no directory.

**This is the one resource in Latu with a release call of its own** — `cache/1`, a temp view
and a clone allocate server state too, but the session's end is the only thing that frees
them — and it is why `release/1` and `with_checkpoint/3` exist. Latu holds no processes and
has no finalizer, so nothing frees a checkpoint for you; what does bound it is the session,
since the server drops its cached relations when the session ends. **Prefer
`with_checkpoint/3`** unless you need the frame to outlive one function — in a REPL you
usually do, which is why the plain form exists.
`docs/decisions.md` has the argument, including why PySpark's finalizer is not available here.

# `checkpoint!`

```elixir
@spec checkpoint!(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Like `checkpoint/2`, raising on failure and returning the frame.

# `clone_session`

```elixir
@spec clone_session(
  Latu.Session.t(),
  keyword()
) :: {:ok, Latu.Session.t()} | {:error, Latu.Error.t()}
```

Fork the session on the server, returning the clone.

The clone starts with the original's configuration and state, and is isolated from it
afterwards: a temp view registered in one is invisible to the other. Good for a REPL — fork,
make a mess, throw the fork away. Like `status/2`, it needs a session the server already
knows — one that has run at least one thing.

    scratch = Latu.clone_session!(session)
    scratch |> Latu.table("orders") |> Latu.create_temp_view!("candidates")
    Latu.release_session!(scratch)

**The clone shares the channel**, because a Spark Connect session is server-side state keyed
by an id and not a connection. So `disconnect/2` on either one closes the transport for both;
`release_session/2` on the clone ends only the clone.

## Options

  * `:session_id` — the clone's own id. Must be a UUID, and raises if it is not. Defaults to
    a freshly generated one.

# `clone_session!`

```elixir
@spec clone_session!(
  Latu.Session.t(),
  keyword()
) :: Latu.Session.t()
```

Like `clone_session/2`, raising on failure.

# `coalesce`

```elixir
@spec coalesce(Latu.DataFrame.t(), pos_integer()) :: Latu.DataFrame.t()
```

Fewer partitions without a shuffle.

# `col`

```elixir
@spec col(Latu.DataFrame.t(), String.t() | atom()) :: Latu.Plan.expression()
```

A reference to one of this DataFrame's columns, tagged with its identity.

See `Latu.DataFrame.col/2`. For an untagged reference, `Latu.Column.col/1`.

# `col_regex`

```elixir
@spec col_regex(Latu.DataFrame.t(), String.t()) :: Latu.Plan.expression()
```

Columns whose names match a Java regex.

    Latu.select(df, Latu.col_regex(df, "`.*_id`"))

Spark wants the pattern in backticks. Always relative to a frame, as PySpark's `colRegex`
is — the pattern matches that frame's column names, so it takes the frame rather than
standing alone like `Latu.Column.col/1`.

Note that a plain `"prefix.*"` is **not** a regex to Spark: `select/2` sends it as an
ordinary attribute and the analyzer expands it as a qualified star. This is the explicit
form, and PySpark has no implicit one either.

# `collect`

```elixir
@spec collect(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, [map()]} | {:error, Latu.Error.t()}
```

All the rows, as maps with atom keys.

An action: the whole result comes to your machine. `stream/2` is the lazy form, and
`to_explorer/2` the columnar one.

## Options

  * `:keys` — how column names come back. Defaults to `:atoms`. Use `:strings` when the
    names come out of dynamic SQL: atom keys are bounded by the columns you have ever
    selected, and an unbounded one is an atom-table leak.
  * `:progress` — a 1-arity function called with a `Latu.Progress` as the query runs.
    Defaults to `nil`. Every action that reaches the server takes it, the `*_with_metrics`
    twins included; `glimpse/2` is the one exception, and says why.

## Examples

    Latu.range(session, 2) |> Latu.collect()
    #=> {:ok, [%{id: 0}, %{id: 1}]}

    Latu.collect(df, keys: :strings)
    #=> {:ok, [%{"id" => 0}]}

    Latu.collect(df, progress: &IO.inspect/1)

See `Latu.DataFrame.collect/2`.

# `collect!`

```elixir
@spec collect!(
  Latu.DataFrame.t(),
  keyword()
) :: [map()]
```

Like `collect/2`, raising on failure.

# `collect_with_metrics`

```elixir
@spec collect_with_metrics(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, [map()], Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`collect/2`, and the metrics `observe/3` asked for.

## Options

`collect/2`'s: `:keys` and `:progress`.

See `observe/3`.

# `collect_with_metrics!`

```elixir
@spec collect_with_metrics!(
  Latu.DataFrame.t(),
  keyword()
) :: {[map()], Latu.ExecutionInfo.t()}
```

Like `collect_with_metrics/2`, raising on failure and returning `{rows, metrics}`.

# `columns`

```elixir
@spec columns(Latu.DataFrame.t()) :: {:ok, [String.t()]} | {:error, Latu.Error.t()}
```

The column names.

    Latu.columns(df)  #=> {:ok, ["id", "name"]}

# `columns!`

```elixir
@spec columns!(Latu.DataFrame.t()) :: [String.t()]
```

Like `columns/1`, raising on failure.

# `conf`

```elixir
@spec conf(Latu.Session.t(), String.t()) ::
  {:ok, String.t() | nil} | {:error, Latu.Error.t()}
```

A session config value, or nil where Spark has nothing to give.

    Latu.conf!(session, "spark.sql.shuffle.partitions")   #=> "4"
    Latu.conf!(session, "spark.sql.ansi.enabled")         #=> "false", Spark's own default
    Latu.conf!(session, "no.such.conf")                   #=> nil

A key that is set gives its value; a key Spark knows but nobody has set gives **Spark's own
default**; a key Spark does not know at all gives nil — `Map.get/2`'s shape. Spark's
`GetOption` arm, which PySpark never sends. `fetch_conf/2` is the same read with an error for
the unknown key, and `conf/3` is the one that supplies its own default instead of Spark's.

Values are strings, as Spark's runtime conf holds them.

# `conf`

```elixir
@spec conf(
  Latu.Session.t(),
  String.t(),
  String.t() | number() | boolean() | atom() | nil
) ::
  {:ok, String.t() | nil} | {:error, Latu.Error.t()}
```

A session config value, falling back to `default` rather than to Spark's own default.

    Latu.conf!(session, "spark.sql.shuffle.partitions", "200")
    Latu.conf!(session, "my.app.setting", nil)

**This overrides Spark's default, it does not follow it.** `GetWithDefault` reads only what
the session has actually set; for anything unset you get `default` back, even where Spark has
a documented default of its own. That is the point of it — Spark's own comment for this path
says "useful when the default value defined by Apache Spark is not the desired one" — and it
is why `conf/2` exists for the other reading.

A `nil` default is PySpark's `spark.conf.get(key, None)` and comes back as nil. Anything else
follows `set_conf/3`'s coercion, and Spark type-checks it against the conf it belongs to, so
a default of `"soon"` for an integer conf is refused rather than returned.

# `conf!`

```elixir
@spec conf!(Latu.Session.t(), String.t()) :: String.t() | nil
```

Like `conf/2`, raising on failure.

# `conf!`

```elixir
@spec conf!(
  Latu.Session.t(),
  String.t(),
  String.t() | number() | boolean() | atom() | nil
) ::
  String.t() | nil
```

Like `conf/3`, raising on failure.

# `confs`

```elixir
@spec confs(
  Latu.Session.t(),
  keyword()
) ::
  {:ok, %{required(String.t()) => String.t() | nil}} | {:error, Latu.Error.t()}
```

Every config the session has **set**, as a map.

    Latu.confs!(session)
    Latu.confs!(session, prefix: "spark.sql.")

Not the registry: Spark's `getAll` reads the session's own settings, so a conf sitting at its
default is absent here even though `conf/2` answers for it. Expect a handful of entries on a
fresh session — whatever the server was started with — not hundreds.

## Options

  * `:prefix` — return only keys starting with this string, filtered on the server. Defaults
    to `nil`, meaning every set conf. Spark strips the prefix off the keys it returns; Latu
    puts it back, so the keys here are always whole keys you can hand to `conf/2`.

# `confs!`

```elixir
@spec confs!(
  Latu.Session.t(),
  keyword()
) :: %{required(String.t()) => String.t() | nil}
```

Like `confs/2`, raising on failure.

# `connect`

```elixir
@spec connect(
  String.t() | Latu.Session.t() | nil,
  keyword()
) :: {:ok, Latu.Session.t()} | {:error, Latu.Error.t()}
```

Connect to a Spark Connect server.

Takes an `sc://` URL or an unconnected `Latu.Session`. With no arguments, reads
`SPARK_REMOTE`.

    {:ok, session} = Latu.connect("sc://localhost:15002")
    {:ok, session} = Latu.connect("sc://prod:15002/;use_ssl=true", timeout: :infinity)

## Options

These are `Latu.Session.from_url/2`'s, and they override the URL — so they apply to a URL
and passing them alongside an already-built session raises. Every one is also a field of
`%Latu.Session{}`, which is where their defaults live.

  * `:session_id` — the id Spark keys server-side state on. A UUID; one is generated.
  * `:user_id`, `:user_name` — the identity the server sees. Also settable in the URL.
  * `:client_type` — the user agent string. Defaults to Latu's own.
  * `:timeout` — per-RPC deadline in milliseconds, or `:infinity`. Defaults to `60_000`.
  * `:connect_timeout` — deadline for establishing the channel. Defaults to `10_000`.
  * `:tags` — execution tags every query inherits, so `interrupt/2` can find them later.
    Defaults to `[]`; `Latu.Session.add_tag/2` adds one afterwards.
  * `:window_size` — HTTP/2 flow-control window in bytes. Defaults to `134_217_728` (128 MiB).
    A throughput setting, not a memory limit: it does not cap what the transport buffers for
    a consumer that reads slowly (`stream/2`).
  * `:keepalive` — HTTP/2 ping interval in milliseconds. Defaults to `60_000`.
  * `:keepalive_tolerance` — pings missed before the channel is considered dead. Defaults
    to `2`.
  * `:retry` — a `Latu.Retry` policy every RPC retries under. Defaults to `%Latu.Retry{}`.

Why these numbers, and why the knobs live on the session rather than in application config,
is in `docs/decisions.md`.

# `connect!`

```elixir
@spec connect!(
  String.t() | Latu.Session.t() | nil,
  keyword()
) :: Latu.Session.t()
```

Like `connect/2`, raising on failure.

# `copy_to_fs`

```elixir
@spec copy_to_fs(Latu.Session.t(), String.t(), binary()) ::
  :ok | {:error, Latu.Error.t()}
```

Copy bytes to a file on the cluster's filesystem.

    bytes = Explorer.DataFrame.dump_parquet!(frame)
    :ok = Latu.copy_to_fs(session, "/data/orders.parquet", bytes)
    orders = Latu.read(session, format: "parquet", path: "/data/orders.parquet")

PySpark's `copyFromLocalToFs`. The bytes travel as an artifact and the server writes them with
Hadoop's `FileSystem`, so they land on the driver's **default filesystem**: HDFS or object
storage on a cluster, the driver's own disk on a single machine. `path` is absolute and
carries no scheme; the default filesystem decides where `/data` is. An existing file is
overwritten, and nothing stays in the session.

The other way to hand data over is `create_dataframe/3`, which carries the bytes inside the
plan. This one leaves a file behind that any later session can `read/2`, and it is the route
when the file and the server are not on the same disk, which is every container.

**A destination on the driver's local disk is refused by default**, because it would let a
client overwrite any file the driver can. A single-machine server allows it with
`spark.sql.artifact.copyFromLocalToFs.allowDestLocal`, set on the session with `set_conf/3`
or on the server at start-up; a cluster whose default filesystem is not local needs nothing.

# `copy_to_fs!`

```elixir
@spec copy_to_fs!(Latu.Session.t(), String.t(), binary()) :: :ok
```

Like `copy_to_fs/3`, raising on failure.

# `corr`

```elixir
@spec corr(Latu.DataFrame.t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  {:ok, float()} | {:error, Latu.Error.t()}
```

Correlation of two numeric columns. An action: it runs when called.

    Latu.corr(df, :score, :weight)  #=> {:ok, 0.98}

**A null counts as zero here too**, exactly as in `cov/4`, and for the same reason — so this
and `F.corr/2` disagree on data with nulls. A correlation Spark cannot compute comes back as
`NaN` rather than an error.

## Options

  * `:method` — the correlation method. Defaults to `:pearson`, and Spark has no other; it
    is an option only because PySpark's signature has one.
  * `:progress` — as `collect/2` describes. Defaults to `nil`.

# `corr!`

```elixir
@spec corr!(Latu.DataFrame.t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  float()
```

Like `corr/4`, raising on failure.

# `count`

```elixir
@spec count(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, non_neg_integer()} | {:error, Latu.Error.t()}
```

Rows per group (lazy), or how many rows there are (an action).

    df |> Latu.group_by(:suburb) |> Latu.count()   # a DataFrame with a count column
    df |> Latu.count()                             # {:ok, 10}

Spark's own overloading: `GroupedData.count` is a transformation, `DataFrame.count` runs
the query. The structs disambiguate.

## Options

  * `:progress` — a 1-arity function called with a `Latu.Progress` as the query runs.
    Defaults to `nil`. See `collect/2`.

On a grouped frame the count is lazy and takes no options at all — passing any raises,
because there is no execution to watch.

See `Latu.GroupedData.count/1` and `Latu.DataFrame.count/1`.

# `count!`

```elixir
@spec count!(
  Latu.DataFrame.t(),
  keyword()
) :: non_neg_integer()
```

Like `count/2`, raising on failure. On a grouped frame it is `count/1` itself — a lazy count
cannot fail, and the `!` is accepted so the pair reads the same after a `group_by/2`.

# `count_with_metrics`

```elixir
@spec count_with_metrics(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, non_neg_integer(), Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`count/2`, and the metrics `observe/3` asked for.

## Options

`count/2`'s: `:progress`.

See `observe/3`.

# `count_with_metrics!`

```elixir
@spec count_with_metrics!(
  Latu.DataFrame.t(),
  keyword()
) :: {non_neg_integer(), Latu.ExecutionInfo.t()}
```

Like `count_with_metrics/2`, raising on failure and returning `{count, metrics}`.

# `cov`

```elixir
@spec cov(Latu.DataFrame.t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  {:ok, float()} | {:error, Latu.Error.t()}
```

Sample covariance of two numeric columns. An action: it runs when called.

    Latu.cov(df, :score, :weight)  #=> {:ok, 12.5}

**A null counts as zero here; it does not drop the row.** Spark wraps each column in
`when(isnull(c), 0.0)` before aggregating (`StatFunctions.calculateCovImpl`), so this and
`F.covar_samp/2` — which ignores a row where either value is null — give **different answers
on the same data**. Both are Spark's. Use the aggregate if you want the nulls gone.

## Options

  * `:progress` — as `collect/2` describes. Defaults to `nil`.

# `cov!`

```elixir
@spec cov!(Latu.DataFrame.t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  float()
```

Like `cov/4`, raising on failure.

# `create_dataframe`

```elixir
@spec create_dataframe(Latu.Session.t(), term(), keyword()) ::
  {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

A DataFrame from local data — `collect/2`'s inverse. An action: the data ships to the
server when called.

Takes a list of row maps (columns sorted by key, PySpark's rule for dicts), column data (a
keyword list in declared order, or a map sorted by key), or an `Explorer.DataFrame`. The
data travels as one Arrow IPC stream inside the plan while it fits the server's
`localRelationCacheThreshold` (64 MiB by default); past that it is chunked and cached as
session artifacts, and the plan references the hashes — PySpark's own escalation.

## Options

  * `:schema` — a string the server casts the data to: DDL (`"id INT, name STRING"`) or
    Spark's JSON schema form. Defaults to `nil`, meaning infer from the data. **Empty data
    needs one**, since there is nothing to infer from. There is no client-side schema model,
    here or in `read/2`.

**A list column cannot go up.** Explorer writes Arrow's `LargeList` and the server's reader
has no arm for it, so an `array<...>` column is refused at the first action, not here,
whatever `schema:` says
(`docs/deviations.md`); `copy_to_fs/3` and a Parquet file are the route for one.

**A schema is matched to the data by name**, whatever order either is in:
`[%{id: 1, jan: 10.0}]` with `schema: "jan DOUBLE, id INT"` puts each value under its own
name. When no data name is in the schema, the schema renames by position —
`create_dataframe(session, [n: [1]], schema: "id INT")` — and a schema naming some columns but
not others is refused, naming them. The server parses the schema as the frame is built (one
round trip), so a malformed one fails here rather than at the first action. Every column
ships nullable — Explorer writes Arrow fields so — and the server refuses to cast one to a
`NOT NULL` field (`NULLABLE_COLUMN_OR_FIELD`); leave nullability to the default.

## Examples

    {:ok, df} = Latu.create_dataframe(session, [%{id: 1, name: "Ada"}, %{id: 2, name: "Bo"}])

    Latu.create_dataframe!(session, id: [1, 2], name: ["Ada", "Bo"])

    Latu.create_dataframe(session, [%{id: 1}], schema: "id INT")
    Latu.create_dataframe(session, [], schema: "id INT, name STRING")

# `create_dataframe!`

```elixir
@spec create_dataframe!(Latu.Session.t(), term(), keyword()) :: Latu.DataFrame.t()
```

Like `create_dataframe/3`, raising on failure.

# `create_temp_view`

```elixir
@spec create_temp_view(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}
```

Register the DataFrame as a temporary view, visible to `sql/3`. An action.

    :ok = Latu.create_temp_view(df, "people", replace: true)
    Latu.sql!(session, "SELECT count(*) FROM people")

## Options

  * `:replace` — swap an existing view rather than raising. Defaults to `false`.
  * `:global` — register in the `global_temp` database, visible across sessions rather than
    just this one. Defaults to `false`.

One call for PySpark's four `create*TempView` methods (`docs/deviations.md`).
`Latu.Catalog.drop_temp_view/2` is the inverse.

## Examples

    :ok = Latu.create_temp_view(df, "people")
    :ok = Latu.create_temp_view(df, "people", replace: true)
    :ok = Latu.create_temp_view(df, "people", global: true, replace: true)

# `create_temp_view!`

```elixir
@spec create_temp_view!(Latu.DataFrame.t(), String.t() | atom(), keyword()) :: :ok
```

Like `create_temp_view/3`, raising on failure.

# `cross_join`

```elixir
@spec cross_join(Latu.DataFrame.t(), Latu.DataFrame.t()) :: Latu.DataFrame.t()
```

Every pairing of the two frames: `join/3` with `how: :cross` and no condition.

Spark's own separate method, kept because the intent is worth spelling out — a cross join by
accident is a very different query from one on purpose.

# `crosstab`

```elixir
@spec crosstab(Latu.DataFrame.t(), String.t() | atom(), String.t() | atom()) ::
  Latu.DataFrame.t()
```

A contingency table of two columns.

    Latu.crosstab(df, :team, :grade)

The first column of the result is named `col1_col2` and holds `col1`'s distinct values; the
remaining columns are `col2`'s distinct values, and the cells are counts. Spark's naming, and
a null becomes the string `"null"` there.

# `cube`

```elixir
@spec cube(Latu.DataFrame.t(), term()) :: Latu.GroupedData.t()
```

Group by every combination of these columns.

# `describe`

```elixir
@spec describe(Latu.DataFrame.t(), [String.t() | atom()] | String.t() | atom()) ::
  Latu.DataFrame.t()
```

`summary/2`'s fixed five — count, mean, stddev, min, max — over the columns named.

    Latu.describe(df)
    Latu.describe(df, [:score, :weight])

A separate Spark relation from `summary/2`, not an option on it.

# `disconnect`

```elixir
@spec disconnect(
  Latu.Session.t(),
  keyword()
) :: {:ok, Latu.Session.t()} | {:error, Latu.Error.t()}
```

Close the channel. Idempotent.

## Options

  * `:release` — end the session on the server first, rather than leaving Spark to time it
    out. Defaults to `false`; `release_session/2` is the direct call, and `docs/decisions.md`
    has why the default stayed put.

## Examples

    {:ok, session} = Latu.disconnect(session)
    {:ok, session} = Latu.disconnect(session, release: true)

# `disconnect!`

```elixir
@spec disconnect!(
  Latu.Session.t(),
  keyword()
) :: Latu.Session.t()
```

Like `disconnect/2`, raising on failure.

# `distinct`

```elixir
@spec distinct(Latu.DataFrame.t(), term(), keyword()) :: Latu.DataFrame.t()
```

Drop duplicate rows, by these columns or by all of them.

    Latu.distinct(df)
    Latu.distinct(df, [:suburb])
    Latu.distinct(events, [:id], within_watermark: true)

## Options

  * `:within_watermark` — on a streaming frame with a watermark, PySpark's
    `dropDuplicatesWithinWatermark`: a duplicate is one that arrives within the watermark
    delay, so the state Spark keeps is bounded instead of growing forever. Defaults to
    `false`. Write the columns first, `[]` for all of them.

# `drop`

```elixir
@spec drop(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Remove columns.

    Latu.drop(df, [:x, "y"])

See `Latu.DataFrame.drop/2`.

# `drop_na`

```elixir
@spec drop_na(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Drop rows by how many non-null values they carry.

## Options

  * `:how` — `:any` (the default) drops a row with any null in it; `:all` drops only rows
    that are null all the way across. Ignored when `:min_non_nulls` is given.
  * `:min_non_nulls` — keep rows carrying at least this many non-null values, overriding
    `:how`, as PySpark's `thresh` overrides its. Spark's own name for the wire field, where
    PySpark abbreviates (`docs/deviations.md`).
  * `:subset` — the columns to judge on: one name, or a list of them. Defaults to `[]`,
    which means every column.

## Examples

    Latu.drop_na(df)                       # any null anywhere
    Latu.drop_na(df, how: :all)            # only rows that are null all the way across
    Latu.drop_na(df, how: :any)            # the default, written out
    Latu.drop_na(df, min_non_nulls: 3)     # keep rows with at least three
    Latu.drop_na(df, subset: [:score])     # judge on these columns only
    Latu.drop_na(df, subset: :score)       # one column needs no list

# `dtypes`

```elixir
@spec dtypes(Latu.DataFrame.t()) ::
  {:ok, [{String.t(), String.t()}]} | {:error, Latu.Error.t()}
```

Name and type per column, as pairs — PySpark's `df.dtypes`.

    Latu.dtypes(df)  #=> {:ok, [{"id", "bigint"}, {"name", "string"}]}

# `dtypes!`

```elixir
@spec dtypes!(Latu.DataFrame.t()) :: [{String.t(), String.t()}]
```

Like `dtypes/1`, raising on failure.

# `error_details`

```elixir
@spec error_details(Latu.Session.t(), Latu.Error.t()) ::
  {:ok, Latu.Error.t()} | {:error, Latu.Error.t()}
```

Fill in an error's full server-side cause chain.

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

    Enum.map(error.causes, & &1.message)
    #=> ["Job aborted due to stage failure: ...", "/ by zero"]

**Most of what you want is already on the error**, with no round trip: the Spark error class,
the SQLSTATE, the JVM class hierarchy, the message parameters and, when the server is
configured to send one, a stack trace all arrive in the gRPC trailers. See `Latu.Error`.
This adds what they do not carry: the **chain of causes**, root cause last, each with its
own frames.

**One case is handled without a call.** The gRPC status carries the server's message cut at
2048 characters. When an error arrives cut, Latu fetches the detail before handing the error
back, so `message` is whole and `causes` is filled. That is one bounded round trip, on an
error path, and only when the message shows the cut. Every other error is an explicit call,
as here. PySpark fetches on every error; a Latu action returns `{:error, _}` for expected
refusals too, and a round trip on each of those would be a poor trade.

An error with no `error_id` — anything that did not come from the server — comes back
unchanged rather than as a failure.

**The server hands an error's detail over exactly once**, and forgets it: its own handler
invalidates the id as it answers. Latu keeps whatever it already has when a fetch comes back
empty, so calling this twice is safe and the second call is simply a wasted round trip.

# `error_details!`

```elixir
@spec error_details!(Latu.Session.t(), Latu.Error.t()) :: Latu.Error.t()
```

Like `error_details/2`, raising on failure.

# `except`

```elixir
@spec except(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) :: Latu.DataFrame.t()
```

Rows in the first and not the second, distinct unless `all: true`. Matches by position.

## Options

  * `:all` — keep duplicates, pairing them up as `EXCEPT ALL` does. Defaults to `false`.
  * `:by_name`, `:allow_missing_columns` — accepted, but setting either to `true` raises:
    they apply to `:union` only, as in PySpark.

## Examples

    Latu.except(df, other)
    Latu.except(df, other, all: true)

PySpark spells these `subtract` and `exceptAll`. See `Latu.DataFrame.except/3`.

# `exists`

```elixir
@spec exists(Latu.DataFrame.t()) :: Latu.Plan.expression()
```

A predicate that holds when this DataFrame has any rows.

    Latu.filter(orders, Latu.exists(open_alerts))

See `Latu.DataFrame.exists/1`. `Latu.Functions.exists/2` is the array function, not this.

# `explain`

```elixir
@spec explain(
  Latu.DataFrame.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

Print the plan Spark would run, and return `:ok`.

    Latu.explain!(df)
    Latu.explain!(df, mode: :formatted)

## Options

  * `:mode` — how much plan to print. Defaults to `:simple`.
    * `:simple` — the physical plan alone.
    * `:extended` — parsed, analysed, optimised and physical plans.
    * `:codegen` — the physical plan plus the generated code.
    * `:cost` — the optimised plan with statistics, where they have been computed.
    * `:formatted` — a split view: the plan outline, then per-node detail.

PySpark spells the same thing two ways — `explain(True)` and `explain(mode="extended")` —
and refuses both together; there is one spelling here. `explain_string/2` returns it instead
of printing.

# `explain!`

```elixir
@spec explain!(
  Latu.DataFrame.t(),
  keyword()
) :: :ok
```

Like `explain/2`, raising on failure.

# `explain_string`

```elixir
@spec explain_string(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}
```

The plan as a string, where `explain/2` prints it.

## Options

`explain/2`'s: `:mode`.

# `explain_string!`

```elixir
@spec explain_string!(
  Latu.DataFrame.t(),
  keyword()
) :: String.t()
```

Like `explain_string/2`, raising on failure.

# `fetch_conf`

```elixir
@spec fetch_conf(Latu.Session.t(), String.t()) ::
  {:ok, String.t()} | {:error, Latu.Error.t()}
```

`conf/2` with an error for a key Spark does not know.

    Latu.fetch_conf!(session, "spark.sql.shuffle.partitions")   #=> "4"
    Latu.fetch_conf(session, "no.such.conf")                    #=> {:error, %Latu.Error{}}

Set value, else Spark's own default, else `SQL_CONF_NOT_FOUND` — Spark's `Get` arm and
PySpark's `spark.conf.get(key)`. `Map.fetch/2` to `conf/2`'s `Map.get/2`: the read for when a
typo'd key should say so.

# `fetch_conf!`

```elixir
@spec fetch_conf!(Latu.Session.t(), String.t()) :: String.t()
```

Like `fetch_conf/2`, raising on failure.

# `fill_na`

```elixir
@spec fill_na(Latu.DataFrame.t(), term(), keyword()) :: Latu.DataFrame.t()
```

Fill nulls with a value.

## Options

  * `:subset` — the columns to fill: one name, or a list of them. Defaults to `[]`, which
    means every column the value's type fits.

## Examples

    Latu.fill_na(df, 0)                          # every column the value's type fits
    Latu.fill_na(df, 0, subset: [:score])        # only these
    Latu.fill_na(df, score: 0, team: "unknown")  # a value per column

**The type is the filter, not an error.** Spark fills only the columns whose type matches the
value, so filling a string column with a number does nothing at all — quietly. Name the
columns, or pass pairs, if you want to be sure.

# `filter`

```elixir
@spec filter(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Keep the rows the condition holds for.

    Latu.filter(df, "id > 3")

See `Latu.DataFrame.filter/2`.

# `first`

```elixir
@spec first(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, map() | nil} | {:error, Latu.Error.t()}
```

The first row, or nil when there are none.

## Options

`collect/2`'s: `:keys` and `:progress`.

## Examples

    Latu.first(df)                  #=> {:ok, %{id: 0}}
    Latu.first(Latu.limit(df, 0))   #=> {:ok, nil}

# `first!`

```elixir
@spec first!(
  Latu.DataFrame.t(),
  keyword()
) :: map() | nil
```

Like `first/2`, raising on failure.

# `freq_items`

```elixir
@spec freq_items(
  Latu.DataFrame.t(),
  [String.t() | atom()] | String.t() | atom(),
  keyword()
) ::
  Latu.DataFrame.t()
```

Frequent items, one array column of candidates per column named.

The algorithm is Karp, Schenker and Papadimitriou's, so the result may contain false
positives — Spark says so too.

## Options

  * `:support` — the minimum frequency a value must reach to be a candidate, as a fraction
    of the rows. Defaults to `0.01`, as in PySpark.

## Examples

    Latu.freq_items(df, [:score, :team])
    Latu.freq_items(df, :score, support: 0.4)

# `glimpse`

```elixir
@spec glimpse(
  Latu.DataFrame.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

A transposed preview: one line per column, with its type and its first few values.

    Latu.glimpse(df)
    #=> Rows: at least 10
    #=> Columns: 3
    #=> $ id    <bigint> 0, 1, 2, 3, 4, 5, 6, 7, 8, 9
    #=> $ name  <string> "Ada", "Bo", nil, "Grace", nil, "Alan", …
    #=> $ score <double> 1.5, 2.5, 3.5, 4.5, 5.5, 6.5, 7.5, 8.5, …

Reads far better than `show/2` on a wide frame, which is most real frames. Two round trips:
the schema, then a `limit`ed collect. Values are rendered with `inspect/1`, so a string is
quoted and a null is `nil`.

**`Rows:` is exact only when it is free.** A sample that comes back short of `num_rows` has
proved there was nothing more to give, so the count is known; otherwise it reads
`at least 10`, because a real count on a Spark frame is a full scan. `count: true` pays for
the exact number, and is the only unbounded thing here.

## Options

  * `:num_rows` — values shown per column. Defaults to `10`.
  * `:width` — where a line is cut. Defaults to `80`; `:infinity` cuts nothing.
  * `:count` — pay for the exact row count. Defaults to `false`.

There is deliberately no `:progress` — everything glimpse reads is bounded except
`count: true`, and `count/2` takes a handler if you want to watch that.

## Examples

    Latu.glimpse(df)
    Latu.glimpse(df, num_rows: 3, width: :infinity)
    Latu.glimpse(df, count: true)

Spark has no method like this; the name, the `$` lines and the transposed shape are dplyr's
and Polars'. `docs/deviations.md`.

# `glimpse!`

```elixir
@spec glimpse!(
  Latu.DataFrame.t(),
  keyword()
) :: :ok
```

Like `glimpse/2`, raising on failure.

# `group_by`

```elixir
@spec group_by(Latu.DataFrame.t(), term()) :: Latu.GroupedData.t()
```

Group rows, giving a `Latu.GroupedData` that `agg/2` turns back into a DataFrame.

    df |> Latu.group_by(:suburb) |> Latu.agg(total: Latu.Column.fun("sum", [:price]))

See `Latu.DataFrame.group_by/2`.

# `grouping_sets`

```elixir
@spec grouping_sets(Latu.DataFrame.t(), [[term()]], term()) :: Latu.GroupedData.t()
```

Group by an explicit list of grouping sets — SQL's `GROUPING SETS`.

    df
    |> Latu.grouping_sets([[:suburb, :year], [:suburb], []], [:suburb, :year])
    |> Latu.agg(total: F.sum(:price))

The fifth `Aggregate.GroupType`, and the one `rollup/2` and `cube/2` are special cases of:
say exactly which combinations you want instead of every prefix or every subset. **An empty
set is the grand total** and is legal — most of the point of the verb.

The second argument is the grouping columns, which Spark needs in order to resolve the sets;
PySpark takes them the same way, as `groupingSets(sets, *cols)`.

# `head`

```elixir
@spec head(Latu.DataFrame.t(), non_neg_integer() | keyword(), keyword()) ::
  {:ok, map() | nil} | {:ok, [map()]} | {:error, Latu.Error.t()}
```

`first/2` under PySpark's other name: one row or nil, not a list. With a count it is
`take/3`: a list, even for one row. Both shapes are PySpark's.

## Options

`collect/2`'s: `:keys` and `:progress`. They go in the second argument when there is no
count, and in the third when there is.

## Examples

    Latu.head(df)                     #=> {:ok, %{id: 0}}
    Latu.head(df, 2)                  #=> {:ok, [%{id: 0}, %{id: 1}]}
    Latu.head(df, keys: :strings)     #=> {:ok, %{"id" => 0}}
    Latu.head(df, 2, keys: :strings)  #=> {:ok, [%{"id" => 0}, %{"id" => 1}]}

# `head!`

```elixir
@spec head!(Latu.DataFrame.t(), non_neg_integer() | keyword(), keyword()) ::
  map() | nil | [map()]
```

Like `head/3`, raising on failure.

# `hint`

```elixir
@spec hint(Latu.DataFrame.t(), String.t() | atom(), [term()]) :: Latu.DataFrame.t()
```

Attach a planner hint.

    Latu.hint(df, "broadcast")
    Latu.hint(df, "merge")
    Latu.hint(df, "repartition", [4, "suburb"])

Join hints are `BROADCAST`, `MERGE`, `SHUFFLE_HASH` and `SHUFFLE_REPLICATE_NL`; partitioning
hints are `COALESCE`, `REPARTITION` and `REPARTITION_BY_RANGE`. Spark matches the name
case-insensitively and **ignores a hint it does not recognise**, so a typo is silent.

Parameters follow Latu's usual coercion — a binary is a string literal, an atom is a column
reference — which matches PySpark even though it calls `F.lit` on every parameter, because
`lit` of a column returns the column unchanged. So `hint(df, "repartition", [4, "suburb"])`
and `hint(df, "repartition", [4, :suburb])` are different plans, and both are valid.

# `input_files`

```elixir
@spec input_files(Latu.DataFrame.t()) ::
  {:ok, [String.t()]} | {:error, Latu.Error.t()}
```

The files this frame reads, as the server resolved them. Empty for a computed frame.

# `input_files!`

```elixir
@spec input_files!(Latu.DataFrame.t()) :: [String.t()]
```

Like `input_files/1`, raising on failure.

# `insert_into`

```elixir
@spec insert_into(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}
```

Insert into an existing table, by position. An action.

PySpark's `insertInto`, including its position-based column matching: the frame's column
*names* are ignored and its column *order* has to match the table's.

## Options

  * `:overwrite` — replace the table's contents rather than adding to them. Defaults to
    `nil`, which sends no mode at all and leaves Spark on append; `false` sends append
    explicitly. This is `write/2`'s `:mode` under PySpark's boolean spelling, and the only
    key here — a writer option has nowhere to go on an insert.

## Examples

    Latu.insert_into(df, "people")
    Latu.insert_into(df, "people", overwrite: true)

# `insert_into!`

```elixir
@spec insert_into!(Latu.DataFrame.t(), String.t() | atom(), keyword()) :: :ok
```

Like `insert_into/3`, raising on failure.

# `insert_into_with_metrics`

```elixir
@spec insert_into_with_metrics(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`insert_into/3`, and the metrics `observe/3` asked for. See `observe/3`.

# `insert_into_with_metrics!`

```elixir
@spec insert_into_with_metrics!(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()
```

Like `insert_into_with_metrics/3`, raising on failure.

# `interrupt`

```elixir
@spec interrupt(
  Latu.Session.t(),
  keyword()
) :: {:ok, [String.t()]} | {:error, Latu.Error.t()}
```

Cancel executions on the server, returning the operation ids it interrupted.

    Latu.interrupt(session)                      # everything this session is running
    Latu.interrupt(session, tag: "exploration")  # everything carrying the tag
    Latu.interrupt(session, operation_id: id)    # one execution

**The process running the query cannot call this**, because it is blocked in the query. That
is what tags are for: tag the session, run the work from a `Task` or another shell, and
interrupt by tag from anywhere.

    session = Latu.Session.add_tag(session, "exploration")
    task = Task.async(fn -> session |> Latu.range(5_000_000_000) |> Latu.count() end)
    # ... later, once status/2 shows it running: an interrupt that reaches the server before
    # the ExecutePlan does matches nothing, and {:ok, []} is the answer.
    Latu.interrupt(session, tag: "exploration")
    Task.await(task)

**This, rather than killing the process.** A Latu execution is reattachable — the promise
that a client may vanish and come back — so a killed client leaves the query *running on the
server*, holding cluster resources until the detached timeout expires. Interrupt it, then let
the task finish: Latu releases the execution as the stream ends, error or not.

An empty list means nothing matched, which is not an error — an execution that has already
finished is nothing to cancel. Unlike `status/2` and `clone_session/2`, this works on a
session the server has never heard of: Spark's interrupt handler creates one rather than
looking one up, so interrupting a session that has run nothing *makes* it real. Harmless.

## Options

At most one, and it chooses the scope. With neither, the scope is the whole session.

  * `:tag` — cancel every execution carrying this tag. Tags come from `:tags` on
    `connect/2` or from `Latu.Session.add_tag/2`.
  * `:operation_id` — cancel one execution, by the id `status/2` reports.

# `interrupt!`

```elixir
@spec interrupt!(
  Latu.Session.t(),
  keyword()
) :: [String.t()]
```

Like `interrupt/2`, raising on failure.

# `intersect`

```elixir
@spec intersect(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) ::
  Latu.DataFrame.t()
```

Rows in both, distinct unless `all: true`. Matches by position.

## Options

  * `:all` — keep duplicates, pairing them up as `INTERSECT ALL` does. Defaults to `false`,
    which is what `intersect` means. PySpark spells the other one `intersectAll`.
  * `:by_name`, `:allow_missing_columns` — accepted, but setting either to `true` raises:
    they apply to `:union` only, as in PySpark.

## Examples

    Latu.intersect(df, other)
    Latu.intersect(df, other, all: true)

See `Latu.DataFrame.intersect/3`.

# `is_empty`

```elixir
@spec is_empty(Latu.DataFrame.t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}
```

Whether the frame has no rows.

A one-row limit, counted server-side. PySpark spells it as an empty projection plus
`take(1)`; the answer is the same (`docs/deviations.md`).

# `is_empty!`

```elixir
@spec is_empty!(Latu.DataFrame.t()) :: boolean()
```

Like `is_empty/1`, raising on failure.

# `is_local`

```elixir
@spec is_local(Latu.DataFrame.t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}
```

Whether Spark can run this plan without a cluster — `spark.range(5)` cannot.

# `is_local!`

```elixir
@spec is_local!(Latu.DataFrame.t()) :: boolean()
```

Like `is_local/1`, raising on failure.

# `is_modifiable`

```elixir
@spec is_modifiable(Latu.Session.t(), String.t()) ::
  {:ok, boolean()} | {:error, Latu.Error.t()}
```

Whether Spark will let this session change that config.

    Latu.is_modifiable!(session, "spark.sql.ansi.enabled")   #=> true
    Latu.is_modifiable!(session, "spark.sql.warehouse.dir")  #=> false, it is static
    Latu.is_modifiable!(session, "my.app.setting")           #=> false, Spark never heard of it

**False does not mean `set_conf/3` will fail.** Spark answers true only for a config it
*defines* and that is not static, so an unregistered key is false and `set_conf/3` will
happily store it anyway. The one this reliably predicts is the static conf, which is refused.
Spark's own spelling is `isModifiable`.

# `is_modifiable!`

```elixir
@spec is_modifiable!(Latu.Session.t(), String.t()) :: boolean()
```

Like `is_modifiable/2`, raising on failure.

# `is_streaming`

```elixir
@spec is_streaming(Latu.DataFrame.t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}
```

Whether the frame is a streaming source. `read/2` and `table/3` build one with
`is_streaming: true`, `table_changes/3` with its own flag.

# `is_streaming!`

```elixir
@spec is_streaming!(Latu.DataFrame.t()) :: boolean()
```

Like `is_streaming/1`, raising on failure.

# `join`

```elixir
@spec join(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) :: Latu.DataFrame.t()
```

Join two DataFrames.

Lazy: nothing is sent until an action. Both frames must come from the same session.

## Options

  * `:on` — what to join on: a column name, a list of names, or a condition expression.
    Defaults to `nil`. **Names and a condition are not the same join.** Names become Spark's
    `using_columns`, which matches on equality *and collapses the duplicate column*; a
    condition leaves both columns in the result. With no `:on` at all and `how: :cross`, it
    is a cross join; with no `:on` and any other `:how`, Spark decides, which is rarely what
    you meant.
  * `:how` — the join type. Defaults to `:inner`. One of:
    * `:inner` — rows matching on both sides.
    * `:cross` — every pair, so normally with no `:on`.
    * `:full` — every row from both, nulls where there is no match. Spark's `full_outer`.
    * `:left` — every left row. Spark's `left_outer`.
    * `:right` — every right row. Spark's `right_outer`.
    * `:semi` — left rows that have a match, and only the left columns. Spark's `left_semi`.
    * `:anti` — left rows that have **no** match, left columns only. Spark's `left_anti`.

Spark's aliases (`"outer"`, `"leftouter"`, `"left_anti"`, …) are not accepted: one spelling
per join type, and an unknown one raises naming the seven. See `docs/deviations.md`.

## Examples

    # match on a shared name; one `customer_id` column comes back
    Latu.join(orders, customers, on: :customer_id)

    # several names
    Latu.join(orders, customers, on: [:region, :customer_id])

    # a condition; both id columns survive, so qualify them downstream
    Latu.join(orders, customers, on: Latu.Column.expr("o.id = c.id"), how: :left)

    # orders with no matching customer
    Latu.join(orders, customers, on: :customer_id, how: :anti)

    Latu.join(sizes, colours, how: :cross)

See `Latu.DataFrame.join/3`.

# `join_as_of`

```elixir
@spec join_as_of(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) ::
  Latu.DataFrame.t()
```

An as-of join: match each left row with the *nearest* right row instead of an equal one.

## Options

  * `:left_as_of` — **required.** The left frame's ordering column: the thing being matched
    *nearest on*, usually a timestamp. A bare name is tagged to the frame it belongs to, so
    the same column name on both sides is unambiguous; pass an expression to build one
    yourself.
  * `:right_as_of` — **required.** The same, on the right frame.
  * `:on` — an equality key applied alongside the as-of match, as `join/3` takes it: names
    become `USING`, an expression becomes a condition. Defaults to `nil`. This is the
    "match within a group" key — symbol, device, account.
  * `:how` — the join type, from `join/3`'s set: `:inner` (the default), `:cross`, `:full`,
    `:left`, `:right`, `:semi`, `:anti`.
  * `:direction` — which way to look for the nearest row. Defaults to `:backward`.
    * `:backward` — the last right row at or before the left one.
    * `:forward` — the first right row at or after it.
    * `:nearest` — whichever is closer in either direction.
  * `:tolerance` — how far a match may be. An **expression**, so an interval is
    `Latu.Column.expr("INTERVAL 1 DAY")` rather than a bare number. Defaults to `nil`,
    meaning no bound.
  * `:allow_exact_matches` — whether an exactly equal row counts as a match. Defaults to
    `true`; `false` makes `:backward` strictly-before and `:forward` strictly-after.

## Examples

    # each trade gets the quote in force at the time, within two seconds, per symbol
    Latu.join_as_of(trades, quotes,
      left_as_of: :time,
      right_as_of: :time,
      on: :symbol,
      tolerance: Latu.Column.expr("INTERVAL 2 SECONDS"),
      direction: :backward
    )

    # keep every left row even where nothing matched
    Latu.join_as_of(trades, quotes, left_as_of: :time, right_as_of: :time, how: :left)

**Joining a frame to itself needs `as/2` on each side.** Both as-of references would
otherwise carry the same plan id and Spark answers `AMBIGUOUS_COLUMN_REFERENCE` — its own
rule for any self-join, with aliasing as the documented fix:

    Latu.join_as_of(Latu.as(df, "l"), Latu.as(df, "r"), left_as_of: :t, right_as_of: :t)

**PySpark keeps this private** as `DataFrame._joinAsOf`, exposed only through
pandas-on-Spark's `merge_asof`. The relation is part of the protocol, so Latu ships it as a
verb; `docs/deviations.md`.

# `lateral_join`

```elixir
@spec lateral_join(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) ::
  Latu.DataFrame.t()
```

A lateral join: the right side may reference the left's columns, row by row.

## Options

  * `:on` — a **condition** only; this relation has no `USING` form, so a bare column name
    is not accepted here as it is on `join/3`. Defaults to `nil`.
  * `:how` — the join type. Defaults to `:inner`; `:left` and `:cross` are the only others.
    Those three are what Spark's `LateralJoinType` accepts, and anything else — including
    `join/3`'s `:full`, `:semi` and `:anti` — is refused here rather than at the server.

## Examples

    Latu.lateral_join(orders, Latu.select_expr(orders, ["explode(items) as item"]))
    Latu.lateral_join(left, right, on: greater(:id, 1), how: :left)

# `limit`

```elixir
@spec limit(Latu.DataFrame.t(), non_neg_integer()) :: Latu.DataFrame.t()
```

Keep at most `count` rows.

# `merge`

```elixir
@spec merge(
  Latu.MergeInto.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

Run the merge.

An action: everything before it was inert. Returns `:ok`, since a merge reports no rows —
`merge_with_metrics/2` is the form that returns what an `observe/3` in the source counted,
which is how you find out how many rows a merge touched without a second pass.

## Options

  * `:progress` — as `collect/2` describes. Defaults to `nil`.

# `merge!`

```elixir
@spec merge!(
  Latu.MergeInto.t(),
  keyword()
) :: :ok
```

Like `merge/2`, raising on failure.

# `merge_into`

```elixir
@spec merge_into(Latu.DataFrame.t(), String.t() | atom(), term(), keyword()) ::
  Latu.MergeInto.t()
```

Start a merge: upsert this frame into a target table.

    source
    |> Latu.as("s")
    |> Latu.merge_into("people", expr("people.id = s.id"))
    |> Latu.when_matched(:update, set: [name: col("s.name")])
    |> Latu.when_not_matched(:insert_all)
    |> Latu.merge()

The frame is the *source*; `table` is the target, and `condition` is what decides whether a
source row matches a target row. Both sides are in scope in the condition, so the names need
qualifying — the target by its table name, the source by `as/2` (`Latu.as(df, "s")`), which
is what makes `people.id = s.id` resolve.

Nothing is sent until `merge/2`. This returns a `Latu.MergeInto`, which is inert data: the
clauses are `when_matched/3`, `when_not_matched/3` and `when_not_matched_by_source/3`, and
at least one is required.

## Options

  * `:schema_evolution` — let the merge add columns the source has and the target does not.
    Defaults to `false`. PySpark's `withSchemaEvolution()`.

The three clause lists are also keys of the underlying struct, but they are built by the
`when_*` verbs rather than passed here.

**The target must be a table that supports row-level operations** — an Iceberg or Delta
table, not a plain catalog table. Spark's own built-in sources do not, and refuse the merge
at analysis; the plan is the same either way, which is why Latu ships the verb.

# `merge_with_metrics`

```elixir
@spec merge_with_metrics(
  Latu.MergeInto.t(),
  keyword()
) :: {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`merge/2`, and the metrics an `observe/3` in the source plan asked for.

    {:ok, info} =
      source
      |> Latu.as("s")
      |> Latu.observe(:audit, rows: F.count(lit(1)))
      |> Latu.merge_into("people", expr("people.id = s.id"))
      |> Latu.when_not_matched(:insert_all)
      |> Latu.merge_with_metrics()

    info.observed  #=> %{audit: %{rows: 3}}

The fifth writer path with a metrics twin, and PySpark threads its observations through this
one too.

## Options

  * `:progress` — as `collect/2` describes. Defaults to `nil`.

# `merge_with_metrics!`

```elixir
@spec merge_with_metrics!(
  Latu.MergeInto.t(),
  keyword()
) :: Latu.ExecutionInfo.t()
```

Like `merge_with_metrics/2`, raising on failure.

# `metadata_column`

```elixir
@spec metadata_column(Latu.DataFrame.t(), String.t() | atom()) ::
  Latu.Plan.expression()
```

A hidden metadata column — `_metadata` on a file source, and whatever a source adds.

    Latu.select(df, path: Latu.Column.expr("_metadata.file_path"))
    Latu.select(df, meta: Latu.metadata_column(df, "_metadata"))

Not in the frame's schema, so `columns/1` does not list it and a bare `col/2` will not
resolve it; this sets the flag that asks for one.

# `nearest_by_join`

```elixir
@spec nearest_by_join(Latu.DataFrame.t(), Latu.DataFrame.t(), term(), keyword()) ::
  Latu.DataFrame.t()
```

A nearest-neighbour join: rank the right side per left row and keep the best few.

The third argument is the **ranking expression** — the distance or similarity being ranked
on. Both frames are in scope in it, so the names need qualifying with `as/2`.

## Options

Three of the four are required, and all four are validated before anything is sent,
mirroring Spark's own `NearestByJoinValidation` as PySpark does — so the message reads the
same wherever it fires.

  * `:num_results` — **required.** How many right rows to keep per left row. An integer from
    1 to 100,000.
  * `:mode` — **required.** How the ranking is computed.
    * `:approx` — an approximate search; cheaper on a large right side.
    * `:exact` — an exhaustive one.
  * `:direction` — **required.** Which end of the ranking is best.
    * `:distance` — smaller ranks better.
    * `:similarity` — larger ranks better.
  * `:how` — the join type. Defaults to `:inner`; `:left` is the only other, and keeps left
    rows that matched nothing.

## Examples

    Latu.nearest_by_join(queries, base, Latu.Column.expr("abs(q.x - b.x)"),
      num_results: 3,
      mode: :approx,
      direction: :distance
    )

    Latu.nearest_by_join(queries, base, Latu.Column.expr("cos_sim(q.v, b.v)"),
      num_results: 10,
      mode: :exact,
      direction: :similarity,
      how: :left
    )

# `observe`

```elixir
@spec observe(
  Latu.DataFrame.t(),
  String.t() | atom(),
  keyword() | [Latu.Plan.expression()]
) ::
  Latu.DataFrame.t()
```

Observe aggregates over a frame without changing what it returns.

The frame comes back unchanged; the metrics come back from the action, through one of the
eight `*_with_metrics` twins — `collect_with_metrics/2`, `count_with_metrics/2`,
`to_explorer_with_metrics/2`, `write_with_metrics/2`, `save_as_table_with_metrics/3`,
`insert_into_with_metrics/3`, `write_v2_with_metrics/3` and `merge_with_metrics/2` — as a
`Latu.ExecutionInfo`.

    df = Latu.observe(df, :quality, rows: F.count(:id), worst: F.min(:price))
    {:ok, info} = Latu.write_with_metrics(df, path: "/out")

    info.observed  #=> %{quality: %{rows: 1000, worst: -2.5}}

Any *other* action still runs and simply does not report — `show/2` on an observed frame
shows it, `collect/2` collects it, a `join/3` over it is an ordinary join — exactly as
PySpark does when nobody reads the `Observation`. The one exception is a plain **write**
(`write/2`, `save_as_table/3`, `insert_into/3`, `write_v2/3`, `merge/2`): a write consumes the
frame, so its metrics would be produced and dropped with nothing to show for it, and it
**raises**, naming the twin. `docs/decisions.md` (M11.1, narrowed at M12.6).

Name each metric with a keyword list: the names are how you read the values back, not
decoration. Spark requires *aggregate* expressions here and refuses a bare column when it
analyses the plan.

PySpark spells this `df.observe(observation, *exprs)`, where the `Observation` is a mutable
handle the client writes into as responses arrive. Latu holds no processes and its frames are
inert, so the metrics ride back with the result instead — see `docs/deviations.md`.

# `offset`

```elixir
@spec offset(Latu.DataFrame.t(), non_neg_integer()) :: Latu.DataFrame.t()
```

Skip the first `count` rows.

# `order_by`

```elixir
@spec order_by(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

`sort/2`, spelled Spark's other way.

# `parse`

```elixir
@spec parse(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Parse a frame of strings into a structured frame.

## Options

  * `:format` — **required.** `:json`, `:csv` or `:xml`. Anything else raises naming the
    three. XML is in Spark 4.2's enum; whether your server implements it is its own question.
  * `:schema` — a type from `parse_ddl_type/2`. Defaults to `nil`, meaning infer.
  * `:options` — the reader options that format takes, as a keyword list. Defaults to `[]`.

## Examples

    df |> Latu.parse(format: :json, options: [multiLine: true])
    df |> Latu.parse(format: :csv, schema: Latu.parse_ddl_type!(session, "id INT, name STRING"))

PySpark spells this `spark.read.json(df)`, overloading the reader on its argument, and parses
a DDL schema **client-side** to fill the field. Latu names the relation instead and asks the
server for the type, which is the same route `to/2` takes.

# `parse_ddl`

```elixir
@spec parse_ddl(Latu.Session.t(), String.t()) ::
  {:ok, [Latu.Result.field()]} | {:error, Latu.Error.t()}
```

What a DDL schema string means to the server, in `schema/1`'s shape.

    Latu.parse_ddl!(session, "id INT, tags ARRAY<STRING>")
    #=> [%{name: "id", type: "int", nullable: true},
    #=>  %{name: "tags", type: "array<string>", nullable: true}]

Latu sends schemas as strings (`read/2`, `create_dataframe/3`), so this is how you check one
without running anything. PySpark keeps the same arm private.

# `parse_ddl!`

```elixir
@spec parse_ddl!(Latu.Session.t(), String.t()) :: [Latu.Result.field()]
```

Like `parse_ddl/2`, raising on failure.

# `parse_ddl_type`

```elixir
@spec parse_ddl_type(Latu.Session.t(), String.t()) ::
  {:ok, Latu.Plan.data_type()} | {:error, Latu.Error.t()}
```

The `DataType` message the server parses a DDL string into.

    {:ok, type} = Latu.parse_ddl_type(session, "id BIGINT, name STRING")
    df = Latu.to(df, type)

`parse_ddl/2` *reports* the same parse as data; this hands back Spark's own type message,
which is what `to/2` needs. `ToSchema.schema` is a `DataType` with no string alternative:
on 4.2.0, `DataTypeProtoConverter.toCatalystType` has no `UNPARSED` case, so a DDL string in
that field is refused.

This is the one place a generated protobuf struct is part of Latu's public surface, and it is
deliberate: Latu holds no client-side type model (`docs/decisions.md`, M8.1 and M10.1), so
the only honest way to name a target schema is to let the server name it. Treat the value as
opaque — pass it to `to/2`, do not read it.

# `parse_ddl_type!`

```elixir
@spec parse_ddl_type!(Latu.Session.t(), String.t()) :: Latu.Plan.data_type()
```

Like `parse_ddl_type/2`, raising on failure.

# `persist`

```elixir
@spec persist(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

Ask the server to cache this frame, and hand it back.

Over Connect this is a round trip, where classic Spark's is a driver-local call that cannot
fail — hence the tuple, with `persist!/2` for the pipe. The caching itself is still lazy: a
success means the server registered the query, not that anything is materialised.

## Options

  * `:level` — one of Spark's own storage levels. Defaults to `:memory_and_disk_deser`, as
    in Scala since 3.0.
    * `:none` — no storage.
    * `:disk_only`, `:disk_only_2`, `:disk_only_3` — disk, at 1, 2 or 3 replicas.
    * `:memory_only`, `:memory_only_2` — memory, serialized, at 1 or 2 replicas.
    * `:memory_and_disk`, `:memory_and_disk_2` — memory, spilling to disk.
    * `:memory_and_disk_deser` — the same, deserialized. The default.
    * `:off_heap` — off-heap memory, spilling to disk.

## Examples

    df = Latu.cache!(df)
    df |> Latu.persist!(level: :disk_only) |> Latu.count()
    {:ok, df} = Latu.persist(df, level: :memory_only_2)

# `persist!`

```elixir
@spec persist!(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Like `persist/2`, raising on failure. Returns the DataFrame, so it pipes.

# `pivot`

```elixir
@spec pivot(Latu.GroupedData.t(), String.t() | atom(), [term()]) ::
  Latu.GroupedData.t()
```

Pivot a grouped frame on a column. See `Latu.GroupedData.pivot/3`.

# `print_schema`

```elixir
@spec print_schema(
  Latu.DataFrame.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

Print the schema tree Spark renders, and return `:ok`.

    Latu.print_schema!(df)
    # root
    #  |-- id: long (nullable = false)

Spark does the rendering, so the output is its own — PySpark renders this one client-side,
which is the difference (`docs/deviations.md`).

## Options

  * `:level` — bound the depth of the tree. A positive integer; anything else raises.
    Defaults to `nil`, meaning the whole schema.

## Examples

    Latu.print_schema!(df)
    Latu.print_schema!(df, level: 1)

# `print_schema!`

```elixir
@spec print_schema!(
  Latu.DataFrame.t(),
  keyword()
) :: :ok
```

Like `print_schema/2`, raising on failure.

# `random_split`

```elixir
@spec random_split(Latu.DataFrame.t(), [number()], keyword()) :: [Latu.DataFrame.t()]
```

Split the frame into slices whose sizes are proportional to `weights`.

    [train, test] = Latu.random_split(df, [0.8, 0.2], seed: 42)

The weights are normalised, so `[8, 2]` and `[0.8, 0.2]` give the same plans. Each slice is
a `Sample` over a window of `[0.0, 1.0]`, and the windows tile it exactly — which is why
every slice carries the **same seed** and `deterministic_order`: they only partition the
frame if each one sees the rows in the same order. Without a `:seed` a random one is drawn
once and shared, as in PySpark, so the plans differ between runs.

Nothing runs here: you get a list of lazy frames.

## Options

  * `:seed` — an integer shared by every slice. Defaults to `nil`, which draws one random
    seed once and gives it to all of them, as in PySpark.

## Examples

    [train, test] = Latu.random_split(df, [0.8, 0.2], seed: 42)
    [a, b, c] = Latu.random_split(df, [1, 1, 1])

# `range`

```elixir
@spec range(Latu.Session.t(), integer()) :: Latu.DataFrame.t()
```

A DataFrame of one `id` column of longs, counting up to but not including `stop`.

    Latu.range(session, 5)         # 0, 1, 2, 3, 4
    Latu.range(session, 2, 5)      # 2, 3, 4
    Latu.range(session, 0, 10, 2)  # 0, 2, 4, 6, 8

Lazy: nothing is sent until an action.

## Options

Every arity takes these last.

  * `:num_partitions` — Spark's own fourth argument to `range`. Defaults to `nil`, which
    leaves the choice to the server: `spark.sql.leafNodeDefaultParallelism` if it is set,
    otherwise its default parallelism. `set_conf/3` sets that for the whole session instead
    of per call.

A keyword list rather than Spark's fifth positional argument, because
`Latu.range(session, 0, 10, 2, 4)` gives a reader no way to tell the step from the partitions
— `docs/deviations.md`.

## Examples

    Latu.range(session, 1_000, num_partitions: 4)
    Latu.range(session, 0, 10, 2, num_partitions: 1)

# `range`

```elixir
@spec range(Latu.Session.t(), integer(), integer()) :: Latu.DataFrame.t()
@spec range(Latu.Session.t(), integer(), keyword()) :: Latu.DataFrame.t()
```

See `range/2`.

# `range`

```elixir
@spec range(Latu.Session.t(), integer(), integer(), integer()) :: Latu.DataFrame.t()
@spec range(Latu.Session.t(), integer(), integer(), keyword()) :: Latu.DataFrame.t()
```

See `range/2`.

# `range`

```elixir
@spec range(Latu.Session.t(), integer(), integer(), integer(), keyword()) ::
  Latu.DataFrame.t()
```

See `range/2`.

# `read`

```elixir
@spec read(
  Latu.Session.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Read from a data source.

    Latu.read(session, format: "csv", path: "/data/people.csv",
              schema: "id INT, name STRING", header: true)

Lazy: nothing is sent until an action. PySpark's builder chain
(`spark.read.format(...).option(...).load(...)`) is one call here; see `docs/deviations.md`.

## Options

Five keys are Latu's. **Every other key is a reader option** — a snake_case atom becomes
Spark's camelCase (`infer_schema:` → `"inferSchema"`), a string key passes verbatim, and a
`nil` value drops its pair.

  * `:format` — the source's short name or class: `"csv"`, `"parquet"`, `"jdbc"`, whatever
    the cluster has. Defaults to `nil`, leaving the server on `spark.sql.sources.default`.
  * `:schema` — a string the server parses: DDL (`"id INT, name STRING"`) or Spark's JSON
    schema form. Defaults to `""`, meaning infer. There is no client-side schema model.
  * `:path` — one path to read, as a string.
  * `:paths` — several, as a list of strings. Defaults to `[]`. Passing both `:path` and
    `:paths` raises; they are two spellings of one thing.
  * `:is_streaming` — read as a stream, PySpark's `readStream`. Defaults to `false`. A file
    source then wants a `:schema` and, to bound each batch, `max_files_per_trigger:`; the
    frame is a plan like any other until `write_stream/2` starts it.

A reader option whose own name is one of those five has to be written as a **string key**,
which passes verbatim: `Latu.read(session, [{"path", "s3://bucket/key"}, format: "custom"])`.

## Examples

    Latu.read(session, format: "csv", path: "/data/people.csv",
              schema: "id INT, name STRING", header: true)

    Latu.read(session, format: "parquet", paths: ["/data/a", "/data/b"])

    Latu.read(session, format: "jdbc", url: url, dbtable: "people", fetchsize: 1_000)

# `release`

```elixir
@spec release(Latu.DataFrame.t()) :: :ok | {:error, Latu.Error.t()}
```

Free a checkpointed frame's server-side storage.

Only a frame `checkpoint/2` handed back can be released — anything else raises, because
releasing what you did not allocate is a mistake worth catching client-side.

**Releasing twice succeeds**, because the server invalidates a cache entry rather than
looking one up (`handleRemoveCachedRemoteRelationCommand`). *Querying* a released frame is
what fails, and the error names the id. So a release is safe to repeat and a frame is not
safe to keep — the opposite pair from what the shape of the API suggests.

# `release!`

```elixir
@spec release!(Latu.DataFrame.t()) :: :ok
```

Like `release/1`, raising on failure.

# `release_session`

```elixir
@spec release_session(
  Latu.Session.t(),
  keyword()
) :: {:ok, Latu.Session.t()} | {:error, Latu.Error.t()}
```

End the session on the server, without closing the channel.

Everything the session held goes with it: temp views, cached frames, artifacts, confs. The
channel stays open, so this is how you end a clone. `disconnect/2` will call it for you with
`release: true`, which is not its default.

## Options

  * `:allow_reconnect` — let a client reconnect to this session id afterwards. Defaults to
    `false`, which is what PySpark sends and what "I am done with this session" means.
  * `:timeout` — deadline for the call in milliseconds, or `:infinity`. Defaults to the
    session's own `:timeout`.

# `release_session!`

```elixir
@spec release_session!(
  Latu.Session.t(),
  keyword()
) :: Latu.Session.t()
```

Like `release_session/2`, raising on failure.

# `rename`

```elixir
@spec rename(Latu.DataFrame.t(), keyword() | map() | [String.t() | atom()]) ::
  Latu.DataFrame.t()
```

Rename columns, by mapping or positionally.

    Latu.rename(df, id: :n)
    Latu.rename(df, [:renamed])

See `Latu.DataFrame.rename/2`.

# `repartition`

```elixir
@spec repartition(Latu.DataFrame.t(), pos_integer() | term()) :: Latu.DataFrame.t()
```

Shuffle into `count` partitions, or partition by these columns, or both.

    Latu.repartition(df, 4)
    Latu.repartition(df, 4, [:suburb])

`coalesce/2` is the same Spark relation without the shuffle, so it can only reduce the
partition count. See `Latu.DataFrame.repartition/2`.

# `repartition`

```elixir
@spec repartition(Latu.DataFrame.t(), pos_integer(), term()) :: Latu.DataFrame.t()
```

See `repartition/2`.

# `repartition_by_range`

```elixir
@spec repartition_by_range(Latu.DataFrame.t(), term(), keyword()) ::
  Latu.DataFrame.t()
```

Range-partition the frame by these columns.

    Latu.repartition_by_range(df, [:suburb, Latu.Column.desc(:price)])
    Latu.repartition_by_range(df, [:suburb], num_partitions: 8)

The same relation `repartition/3` builds, carrying **sort orders** instead of bare
expressions — a range partitioner needs an ordering to cut on. Sort keys, so
`Latu.Column.desc/1` and friends apply.

## Options

  * `:num_partitions` — how many partitions to cut into. Defaults to `nil`, leaving the
    choice to the server, as on `range/2`.

# `replace`

```elixir
@spec replace(Latu.DataFrame.t(), [{term(), term()}], keyword()) :: Latu.DataFrame.t()
```

Replace values with other values, as `{old, new}` pairs.

## Options

  * `:subset` — the columns to replace in: one name, or a list of them. Defaults to `[]`,
    which means every column.

## Examples

    Latu.replace(df, [{"red", "crimson"}])
    Latu.replace(df, [{"red", "crimson"}, {"blue", "navy"}], subset: [:team])

Both sides are literals, and as with `fill_na/3` the types have to line up: a pair whose old
value cannot occur in a column simply never matches there.

# `rollup`

```elixir
@spec rollup(Latu.DataFrame.t(), term()) :: Latu.GroupedData.t()
```

Group by every prefix of these columns, plus the grand total.

# `same_semantics`

```elixir
@spec same_semantics(Latu.DataFrame.t(), Latu.DataFrame.t()) ::
  {:ok, boolean()} | {:error, Latu.Error.t()}
```

Whether two frames compute the same thing, up to the plan Spark analyses.

Spelling a query two ways gives one answer; a different literal gives another. Both frames
must share a session.

# `same_semantics!`

```elixir
@spec same_semantics!(Latu.DataFrame.t(), Latu.DataFrame.t()) :: boolean()
```

Like `same_semantics/2`, raising on failure.

# `sample`

```elixir
@spec sample(Latu.DataFrame.t(), number(), keyword()) :: Latu.DataFrame.t()
```

A random fraction of the rows. Lazy.

## Options

  * `:seed` — an integer. Defaults to `nil`, which draws a random one, as in PySpark — so
    the *plan* differs between runs, not just the result.
  * `:with_replacement` — sample with replacement, so a row can appear more than once.
    Defaults to `false`.
  * `:lower_bound` — where the sampled window starts. Defaults to `0.0`. Only
    `random_split/3` sets it, and it is what makes that verb's slices partition the frame;
    you rarely want it by hand.
  * `:deterministic_order` — make the sample stable by forcing a deterministic row order
    first. Defaults to `false`.

## Examples

    Latu.sample(df, 0.1)
    Latu.sample(df, 0.1, seed: 42)
    Latu.sample(df, 0.1, with_replacement: true, seed: 42)

See `Latu.DataFrame.sample/3`.

# `sample_by`

```elixir
@spec sample_by(Latu.DataFrame.t(), term(), [{term(), number()}] | map(), keyword()) ::
  Latu.DataFrame.t()
```

A stratified sample: a fraction of the rows per stratum.

    Latu.sample_by(df, :team, [{"red", 0.5}, {"blue", 1.0}], seed: 42)

Strata are **values**, not column names, so they are given as strings or numbers — an atom is
a column reference everywhere else in Latu and would be ambiguous here. A stratum the map does
not mention contributes no rows.

## Options

  * `:seed` — an integer. Defaults to `nil`, which draws one at random exactly as `sample/3`
    does, so the plan differs between runs. Pass one to make it reproducible.

## Examples

    Latu.sample_by(df, :team, [{"red", 0.5}, {"blue", 1.0}], seed: 42)
    Latu.sample_by(df, :team, %{"red" => 0.5, "blue" => 1.0})

# `save_as_table`

```elixir
@spec save_as_table(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}
```

Write as a catalog table. An action.

## Options

`write/2`'s six, minus `:path`: `:format`, `:mode`, `:partition_by`, `:sort_by`,
`:cluster_by` and `:bucket_by`. Every other key is a writer option.

**`path:` is therefore a writer option here, not a Latu key**, which is how you give a
managed table an explicit location — Spark's `option("path", ...)` alongside
`saveAsTable`. The table name is the second argument.

## Examples

    Latu.save_as_table(df, "people", mode: :overwrite)
    Latu.save_as_table(df, "people", format: "parquet", partition_by: [:region])
    Latu.save_as_table(df, "people", format: "parquet", path: "/warehouse/people")

# `save_as_table!`

```elixir
@spec save_as_table!(Latu.DataFrame.t(), String.t() | atom(), keyword()) :: :ok
```

Like `save_as_table/3`, raising on failure.

# `save_as_table_with_metrics`

```elixir
@spec save_as_table_with_metrics(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`save_as_table/3`, and the metrics `observe/3` asked for. See `observe/3`.

# `save_as_table_with_metrics!`

```elixir
@spec save_as_table_with_metrics!(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()
```

Like `save_as_table_with_metrics/3`, raising on failure.

# `scalar`

```elixir
@spec scalar(Latu.DataFrame.t()) :: Latu.Plan.expression()
```

This DataFrame as a scalar subquery — a single value, hoisted into the plan that uses it.

    Latu.filter(orders, greater(:amount, Latu.scalar(totals)))

See `Latu.DataFrame.scalar/1`.

# `schema`

```elixir
@spec schema(Latu.DataFrame.t()) ::
  {:ok, [Latu.Result.field()]} | {:error, Latu.Error.t()}
```

The frame's columns, with Spark's own name for each type. An action: the server analyses
the plan.

    Latu.schema(df)
    #=> {:ok, [%{name: "id", type: "bigint", nullable: false},
    #=>        %{name: "tags", type: "array<string>", nullable: true}]}

Nested types render into the type string, so the list stays flat. Latu has no client-side
type model in either direction: a schema you *send* is a string the server parses
(`create_dataframe/3`, `read/2`), and one you *read back* is this.

# `schema!`

```elixir
@spec schema!(Latu.DataFrame.t()) :: [Latu.Result.field()]
```

Like `schema/1`, raising on failure.

# `select`

```elixir
@spec select(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Keep these columns, in this order.

    Latu.select(df, [:id, "name"])
    Latu.select(df, [:id, doubled: Latu.Column.multiply(:id, 2)])

See `Latu.DataFrame.select/2`.

# `select_expr`

```elixir
@spec select_expr(Latu.DataFrame.t(), [String.t()] | String.t()) :: Latu.DataFrame.t()
```

Keep these SQL expressions, written as strings.

    Latu.select_expr(df, ["id", "price * 1.1 as with_tax"])

`select/2` over `Latu.Column.expr/1` for each one, which is exactly PySpark's own
composition — so anything SQL can say fits here without a wrapper.

# `semantic_hash`

```elixir
@spec semantic_hash(Latu.DataFrame.t()) :: {:ok, integer()} | {:error, Latu.Error.t()}
```

A hash of the analysed plan: equal for frames `same_semantics/2` calls equal.

# `semantic_hash!`

```elixir
@spec semantic_hash!(Latu.DataFrame.t()) :: integer()
```

Like `semantic_hash/1`, raising on failure.

# `set_conf`

```elixir
@spec set_conf(
  Latu.Session.t(),
  String.t(),
  String.t() | number() | boolean() | atom()
) ::
  :ok | {:error, Latu.Error.t()}
```

Set one config on the server.

    :ok = Latu.set_conf(session, "spark.sql.shuffle.partitions", 8)
    Latu.set_conf!(session, "spark.sql.ansi.enabled", true)

A number, boolean or atom is written as Spark's own string form; a string passes verbatim.
Spark type-checks the value against the conf, so `"eight"` here is refused, and it refuses a
**static** conf outright (`CANNOT_MODIFY_VALUE_OF_STATIC_CONFIG`) — see `is_modifiable/2`.
An unrecognised key is not an error: Spark stores it, which is how a data source's own
options get set.

The config lives on the server, not in the struct, so there is nothing to hand back: the same
session keeps working, and `Latu.clone_session/2` carries the confs into the clone. A
deprecation warning Spark attaches to a key is logged.

# `set_conf!`

```elixir
@spec set_conf!(
  Latu.Session.t(),
  String.t(),
  String.t() | number() | boolean() | atom()
) :: :ok
```

Like `set_conf/3`, raising on failure.

# `set_confs`

```elixir
@spec set_confs(Latu.Session.t(), Enumerable.t()) :: :ok | {:error, Latu.Error.t()}
```

Set several configs in one round-trip.

    Latu.set_confs(session, %{"spark.sql.ansi.enabled" => true})
    Latu.set_confs(session, [{"spark.sql.shuffle.partitions", 8}, {"a.b", "c"}])

A map or anything enumerating `{key, value}`. Values follow `set_conf/3`. Spark applies a list
in order and stops at the first refusal, so one bad key fails the call with the pairs before
it already applied.

# `set_confs!`

```elixir
@spec set_confs!(Latu.Session.t(), Enumerable.t()) :: :ok
```

Like `set_confs/2`, raising on failure.

# `show`

```elixir
@spec show(
  Latu.DataFrame.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

Print the table Spark renders, and return `:ok`.

Byte for byte what PySpark's `df.show()` prints: Spark formats it server-side and Latu
decodes one string cell.

## Options

All PySpark's, plus `:progress`.

  * `:num_rows` — how many rows to print. Defaults to `20`.
  * `:truncate` — cell width. Defaults to `20`. `true` means 20, `false` means no
    truncation, and an integer is that width.
  * `:vertical` — one field per line instead of a table, which is how a wide frame stays
    readable. Defaults to `false`.
  * `:progress` — as `collect/2` describes. Defaults to `nil`.

## Examples

    Latu.range(session, 5) |> Latu.show()
    Latu.show(df, num_rows: 100)
    Latu.show(df, truncate: false)
    Latu.show(df, vertical: true, num_rows: 3)

`glimpse/2` is usually the better read on a wide frame. See `Latu.DataFrame.show/2`.

# `show!`

```elixir
@spec show!(
  Latu.DataFrame.t(),
  keyword()
) :: :ok
```

Like `show/2`, raising on failure.

# `sort`

```elixir
@spec sort(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Sort rows.

    Latu.sort(df, :id)
    Latu.sort(df, [Latu.Column.desc(:price), :id])

A bare name sorts ascending with nulls first; `Latu.Column.asc/1`, `Latu.Column.desc/1` and
the four explicit `*_nulls_*` spellings are the alternatives.
`sort_within_partitions/2` sorts within each partition rather than across the frame, which
needs no shuffle. See `Latu.DataFrame.sort/2`.

# `sort_within_partitions`

```elixir
@spec sort_within_partitions(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

Sort within each partition, leaving the partitions unordered.

# `spark_version`

```elixir
@spec spark_version(Latu.Session.t()) :: {:ok, String.t()} | {:error, Latu.Error.t()}
```

The Spark version the server reports, e.g. `"4.2.0"`.

Also the cheapest way to confirm a session is alive and usable.

# `spark_version!`

```elixir
@spec spark_version!(Latu.Session.t()) :: String.t()
```

Like `spark_version/1`, raising on failure.

# `sql`

```elixir
@spec sql(Latu.Session.t(), String.t(), [term()] | map() | keyword()) ::
  {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

Run SQL. An action: the query executes when called — so DDL works — and the DataFrame that
comes back queries the *result*, not the query again.

    {:ok, df} = Latu.sql(session, "SELECT * FROM people WHERE age > 30")
    Latu.sql!(session, "DROP TABLE IF EXISTS scratch")

Parameter markers bind from `args` — a list binds `?` positionally, a map binds `:name` —
and values are literals, never spliced text:

    Latu.sql(session, "SELECT * FROM people WHERE age > :min", %{min: 30})
    Latu.sql(session, "SELECT ? + ?", [2, 3])

A DataFrame can be named into the query with `views:`, and the name is yours — write it in
the SQL and pass the frame:

    Latu.sql(session, "SELECT count(*) FROM orders", views: [orders: df])

The frame is hoisted into the plan, so nothing is registered on the server and the name is
gone when the query is. PySpark generates a name and substitutes it into a Python format
string; Latu takes the name (docs/deviations.md). Bindings can ride along as `args:`.

# `sql!`

```elixir
@spec sql!(Latu.Session.t(), String.t(), [term()] | map() | keyword()) ::
  Latu.DataFrame.t()
```

Like `sql/3`, raising on failure.

# `status`

```elixir
@spec status(Latu.Session.t(), [String.t()]) ::
  {:ok, [%{operation_id: String.t(), state: atom()}]} | {:error, Latu.Error.t()}
```

What the server is running for this session: one map per operation, with its state.

    Latu.status!(session)
    #=> [%{operation_id: "b3f...", state: :running}]

States are Spark's own, downcased: `:running`, `:terminating`, `:succeeded`, `:failed`,
`:cancelled`, `:unknown`, `:unspecified`. Pass a list of operation ids to ask about only
those; the default asks about all of them, **including ones that have finished** — Spark keeps
a session's recent operations and reports them here.

Two things that read wrong until you know them. **`:terminating` does not mean "being
cancelled"**: Spark maps finished, failed *and* cancelled executions to it, and it means the
work is over but the server has not cleaned up yet — the outcome shows as `:succeeded`,
`:failed` or `:cancelled` a moment later. And **the server must already know this session**,
which it does from the first thing the session runs; before that this returns an error saying
so, rather than an empty list.

# `status!`

```elixir
@spec status!(Latu.Session.t(), [String.t()]) :: [
  %{operation_id: String.t(), state: atom()}
]
```

Like `status/2`, raising on failure.

# `storage_level`

```elixir
@spec storage_level(Latu.DataFrame.t()) :: {:ok, map()} | {:error, Latu.Error.t()}
```

How the server is storing this frame, if at all.

    Latu.storage_level!(df)
    #=> %{name: :memory_and_disk_deser, use_disk: true, use_memory: true,
    #=>   use_off_heap: false, deserialized: true, replication: 1}

`:name` is Spark's name for that combination of flags, or `nil` where it has none. An
uncached frame is `:none`.

# `storage_level!`

```elixir
@spec storage_level!(Latu.DataFrame.t()) :: map()
```

Like `storage_level/1`, raising on failure.

# `stream`

```elixir
@spec stream(
  Latu.DataFrame.t(),
  keyword()
) :: Enumerable.t()
```

The result as a lazy stream of `Explorer.DataFrame`s, one per Arrow batch.

For results too large to *decode* at once: one `Explorer.DataFrame` is live at a time, and
stopping early releases the execution. It does not bound what is *received* — the transport
keeps taking the rest of the result whether or not you are consuming, so the raw Arrow bytes
of a large result can still accumulate locally; `:window_size` is a throughput setting, not
a memory limit. Raises `Latu.Error` on failure, since an enumeration has no way to return
one.

## Options

  * `:progress` — a 1-arity function called with a `Latu.Progress` as the query runs.
    Defaults to `nil`. See `collect/2`.

## Examples

    df |> Latu.stream() |> Stream.map(&Explorer.DataFrame.n_rows/1) |> Enum.sum()
    df |> Latu.stream() |> Enum.each(&handle/1)

See `Latu.DataFrame.stream/2`.

# `stream_nx`

```elixir
@spec stream_nx(
  Latu.DataFrame.t(),
  keyword()
) :: Enumerable.t()
```

A lazy stream of `to_nx/2`'s tensors, one map per Arrow batch.

What `stream/2` is for Explorer. The tensors are per batch, so stacking them is the caller's
business — `to_nx/2` is the one that concatenates. Raises `Latu.Error` on failure.

## Options

  * `:columns` — as `to_nx/2` describes. Defaults to `nil`.
  * `:progress` — as `collect/2` describes. Defaults to `nil`.

## Examples

    df |> Latu.stream_nx(columns: ["features"]) |> Enum.map(&Nx.sum(&1["features"]))

See `Latu.DataFrame.stream_nx/2`.

# `summary`

```elixir
@spec summary(Latu.DataFrame.t(), [String.t() | atom()] | String.t() | atom()) ::
  Latu.DataFrame.t()
```

Summary statistics: one row per statistic, one column per column Spark can summarise.

    Latu.summary(df)                        # count, mean, stddev, min, 25%, 50%, 75%, max
    Latu.summary(df, ["count", "min", "max"])
    Latu.summary(df, "90%")

A lazy relation, like every other verb here — nothing runs until you collect or show it.
Names are Spark's, percentiles included.

# `table`

```elixir
@spec table(Latu.Session.t(), String.t() | atom()) :: Latu.DataFrame.t()
```

Read a catalog table by name.

    Latu.table(session, "people")
    Latu.table(session, "events", is_streaming: true)

## Options

One key is Latu's. **Every other key is a reader option**, with `read/2`'s key and value
rules; a map of options passes verbatim and carries no flag.

  * `:is_streaming` — read the table as a stream, PySpark's `readStream.table`. Defaults to
    `false`.

# `table`

```elixir
@spec table(Latu.Session.t(), String.t() | atom(), keyword() | map()) ::
  Latu.DataFrame.t()
```

See `table/2`.

# `table_changes`

```elixir
@spec table_changes(Latu.Session.t(), String.t() | atom(), keyword()) ::
  Latu.DataFrame.t()
```

A table's change feed, as a frame.

PySpark spells it `spark.read.changes(table)`.

## Options

  * `:options` — the CDC window, as a keyword list: `starting_version`, `ending_version`,
    `starting_timestamp`, `ending_timestamp` and the rest, camelCased from snake_case as
    reader options are. Defaults to `[]`.
  * `:is_streaming` — mark it a streaming read. Defaults to `false`.

## Examples

    Latu.table_changes(session, "orders", options: [starting_version: 3])
    Latu.table_changes(session, "orders",
      options: [starting_version: 3, ending_version: 9])

# `table_function`

```elixir
@spec table_function(Latu.Session.t(), String.t() | atom(), [term()]) ::
  Latu.DataFrame.t()
```

A table-valued function, as a frame.

    Latu.table_function(session, "explode", [Latu.Column.expr("array(1, 2, 3)")])
    Latu.table_function(session, "sql_keywords")

One builder for all of them, as `Latu.Column.fun/3` is for scalar functions — PySpark wraps a
fixed handful under `spark.tvf` and this covers those plus whatever a Spark release adds.

Arguments are expressions. A **frame** as a table argument is deliberately not offered: the
protocol's `table_arg` subquery is consumed only by a Python UDTF, which needs a Python worker
and is out of Latu's scope — see `docs/deviations.md`. Use `Latu.sql/3` and SQL's `TABLE(...)`.

# `tail`

```elixir
@spec tail(Latu.DataFrame.t(), non_neg_integer(), keyword()) ::
  {:ok, [map()]} | {:error, Latu.Error.t()}
```

The last `count` rows, as maps.

An action, like `take/3`, because Spark's `Tail` relation collects on the driver — which is
also why it is the one place a large `count` costs driver memory rather than yours.

## Options

`collect/2`'s: `:keys` and `:progress`.

## Examples

    Latu.tail(df, 3)

# `tail!`

```elixir
@spec tail!(Latu.DataFrame.t(), non_neg_integer(), keyword()) :: [map()]
```

Like `tail/3`, raising on failure.

# `take`

```elixir
@spec take(Latu.DataFrame.t(), non_neg_integer(), keyword()) ::
  {:ok, [map()]} | {:error, Latu.Error.t()}
```

The first `count` rows, as maps — `limit/2` then `collect/2`, as in PySpark.

## Options

`collect/2`'s: `:keys` and `:progress`.

## Examples

    Latu.take(df, 3)
    Latu.take(df, 3, keys: :strings)

# `take!`

```elixir
@spec take!(Latu.DataFrame.t(), non_neg_integer(), keyword()) :: [map()]
```

Like `take/3`, raising on failure.

# `to`

```elixir
@spec to(Latu.DataFrame.t(), Latu.Plan.data_type()) :: Latu.DataFrame.t()
```

Reconcile a frame to a target schema.

    type = Latu.parse_ddl_type!(session, "name STRING, id BIGINT")
    df = Latu.to(df, type)

Spark matches columns **by name**, case-insensitively unless `spark.sql.caseSensitive` says
otherwise, and then:

  * reorders them into the target's order;
  * projects away source columns the target does not name;
  * casts where the types are compatible — numeric to numeric, erroring on overflow, but not
    string to int;
  * fails when a column the target names is missing **and the target field is not nullable**;
  * **fills it with null when the target field is nullable** — which is the default for
    every field a DDL string declares, so `parse_ddl_type!(session, "a INT")` against a frame
    with no `a` gives a column of nulls rather than an error.

`Dataset.to`'s own scaladoc says missing columns "lead to failures", but `Project.reorderFields`
fills a nullable one with `Literal.create(null, ...)`. Declare `NOT NULL` in the DDL when a
missing column should be an error. `docs/deviations.md`.

Spark's own name for the verb. The schema is a type message rather than a string because the
wire field has no string form — `parse_ddl_type/2` is how you get one, and its docs say why.

# `to_arrow`

```elixir
@spec to_arrow(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, [binary()]} | {:error, Latu.Error.t()}
```

The raw Arrow IPC binaries, one per batch, bypassing Latu's decoder and schema guard.

Never byte-concatenate them; each is a complete IPC stream.

## Options

  * `:progress` — a 1-arity function called with a `Latu.Progress` as the query runs.
    Defaults to `nil`. See `collect/2`.

## Examples

    {:ok, batches} = Latu.to_arrow(df)

See `Latu.DataFrame.to_arrow/2`.

# `to_arrow!`

```elixir
@spec to_arrow!(
  Latu.DataFrame.t(),
  keyword()
) :: [binary()]
```

Like `to_arrow/2`, raising on failure.

# `to_explorer`

```elixir
@spec to_explorer(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Explorer.DataFrame.t()} | {:error, Latu.Error.t()}
```

The result as one `Explorer.DataFrame`.

Unbounded, like `collect/2` and Spark's own `collect`. Bound the plan rather than the
action — `limit/2` is the Spark way to ask for part of a result, and `stream/2` is the
answer for one too large to decode at once.

## Options

  * `:progress` — as `collect/2` describes. Defaults to `nil`.

## Examples

    {:ok, frame} = Latu.to_explorer(df)
    {:ok, frame} = df |> Latu.limit(10_000) |> Latu.to_explorer()

See `Latu.DataFrame.to_explorer/2`.

# `to_explorer!`

```elixir
@spec to_explorer!(
  Latu.DataFrame.t(),
  keyword()
) :: Explorer.DataFrame.t()
```

Like `to_explorer/2`, raising on failure.

# `to_explorer_with_metrics`

```elixir
@spec to_explorer_with_metrics(
  Latu.DataFrame.t(),
  keyword()
) ::
  {:ok, Explorer.DataFrame.t(), Latu.ExecutionInfo.t()}
  | {:error, Latu.Error.t()}
```

`to_explorer/2`, and the metrics `observe/3` asked for.

## Options

`to_explorer/2`'s: `:limit` and `:progress`.

See `observe/3`.

# `to_explorer_with_metrics!`

```elixir
@spec to_explorer_with_metrics!(
  Latu.DataFrame.t(),
  keyword()
) :: {Explorer.DataFrame.t(), Latu.ExecutionInfo.t()}
```

Like `to_explorer_with_metrics/2`, raising on failure and returning `{frame, metrics}`.

# `to_html`

```elixir
@spec to_html(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}
```

The table `show/2` prints, as an HTML string. Spark's own `_repr_html_`.

In Livebook a frame renders itself through this — add `:kino` to your dependencies and
inspect one in a cell; nothing here needs it otherwise.

## Options

`show/2`'s, less `:vertical`, which Spark's `HtmlString` has no field for.

  * `:num_rows` — how many rows. Defaults to `20`.
  * `:truncate` — cell width. Defaults to `20`.
  * `:progress` — as `collect/2` describes. Defaults to `nil`.

## Examples

    Latu.range(session, 5) |> Latu.to_html!()
    Latu.to_html(df, num_rows: 5, truncate: false)

# `to_html!`

```elixir
@spec to_html!(
  Latu.DataFrame.t(),
  keyword()
) :: String.t()
```

Like `to_html/2`, raising on failure.

# `to_nx`

```elixir
@spec to_nx(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, %{required(String.t()) =&gt; term()}} | {:error, Latu.Error.t()}
```

The result as `Nx` tensors, one per column.

Bypasses the Explorer decoder and the schema guard, exactly as `to_arrow/2` does — which is
what makes a `Vector` column readable here when `collect/2` and `to_explorer/2` refuse it.

A numeric column with no nulls becomes a 1-D tensor; a column of equal-length numeric lists,
or of dense `Vector`s, becomes one `{rows, width}` tensor. Nulls, strings, booleans, ragged
lists and sparse vectors are refused by name.

Unbounded, like `collect/2`. Bound the plan, or use `stream_nx/2` for a result too large to
decode at once.

## Options

  * `:columns` — keep only these columns, by name. Pruning copies, because an Arrow buffer is
    a slice of the whole batch and would otherwise hold the rest alive. Defaults to `nil`,
    every column.
  * `:progress` — as `collect/2` describes. Defaults to `nil`.

## Examples

    {:ok, tensors} = Latu.to_nx(df)
    {:ok, %{"features" => t}} = Latu.to_nx(scored, columns: ["features"])

Needs the optional `:nx` dependency. See `Latu.DataFrame.to_nx/2`.

# `to_nx!`

```elixir
@spec to_nx!(
  Latu.DataFrame.t(),
  keyword()
) :: %{required(String.t()) =&gt; term()}
```

Like `to_nx/2`, raising on failure.

# `transpose`

```elixir
@spec transpose(Latu.DataFrame.t(), term() | nil) :: Latu.DataFrame.t()
```

Rows to columns.

    Latu.transpose(df)
    Latu.transpose(df, :metric)

`index_column`'s values become the new column names. Without one Spark uses the first
column — its rule, not a Latu default. Spark 4 only; the whole frame is collected on the
driver, so this is for small results.

# `tree_string`

```elixir
@spec tree_string(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}
```

The schema tree as a string, where `print_schema/2` prints it.

## Options

  * `:level` — bound the depth of the tree. A positive integer; anything else raises.
    Defaults to `nil`, meaning the whole schema.

# `tree_string!`

```elixir
@spec tree_string!(
  Latu.DataFrame.t(),
  keyword()
) :: String.t()
```

Like `tree_string/2`, raising on failure.

# `union`

```elixir
@spec union(Latu.DataFrame.t(), Latu.DataFrame.t(), keyword()) :: Latu.DataFrame.t()
```

All the rows of both, duplicates kept.

Matches by **position** by default, as Spark's `union` does — the column *names* of the
second frame are ignored, so two frames whose columns are in different orders union into
nonsense rather than an error. `by_name: true` is PySpark's `unionByName`.

## Options

  * `:all` — keep duplicates. Defaults to `true` here, which is what `union` means; `false`
    is a distinct union.
  * `:by_name` — match columns by name instead of position. Defaults to `false`.
  * `:allow_missing_columns` — let the two frames have different columns, filling the gaps
    with nulls. Defaults to `false`, and raises unless `by_name: true` — as it does in
    PySpark.

## Examples

    Latu.union(df, other)
    Latu.union(df, other, by_name: true)
    Latu.union(df, other, by_name: true, allow_missing_columns: true)
    Latu.union(df, other, all: false)

See `Latu.DataFrame.union/3`.

# `unpersist`

```elixir
@spec unpersist(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Latu.DataFrame.t()} | {:error, Latu.Error.t()}
```

Drop the server's cache of this frame, and hand it back.

## Options

  * `:blocking` — wait for the blocks to be freed before answering. Defaults to `false`,
    which is PySpark's default too. Sent either way: the field has presence, so an absent
    one would be a different message.

## Examples

    df = Latu.unpersist!(df)
    {:ok, df} = Latu.unpersist(df, blocking: true)

# `unpersist!`

```elixir
@spec unpersist!(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Like `unpersist/2`, raising on failure. Returns the DataFrame, so it pipes.

# `unpivot`

```elixir
@spec unpivot(Latu.DataFrame.t(), term(), keyword()) :: Latu.DataFrame.t()
```

Wide to long: turn a set of columns into two, one holding their names and one their values.

Spark also calls this `melt`; Latu ships the one name.

## Options

  * `:values` — the columns to unpivot, as a list. Defaults to `nil`, which means **every
    column that is not an id**, worked out by the server. `values: []` is not the same
    thing: it is a different message that sends an empty set.
  * `:variable_column_name` — **required.** The name of the column holding the old column
    names.
  * `:value_column_name` — **required.** The name of the column holding their values.

Both names are required, as in PySpark.

## Examples

    Latu.unpivot(df, [:id],
      values: [:jan, :feb, :mar],
      variable_column_name: "month",
      value_column_name: "sales"
    )

    # every non-id column
    Latu.unpivot(df, [:id],
      variable_column_name: "month",
      value_column_name: "sales"
    )

# `unset_conf`

```elixir
@spec unset_conf(Latu.Session.t(), String.t()) :: :ok | {:error, Latu.Error.t()}
```

Put a config back to Spark's default.

    :ok = Latu.unset_conf(session, "spark.sql.shuffle.partitions")

A key that was never set is not an error. A static conf is refused, as it is by `set_conf/3`.

# `unset_conf!`

```elixir
@spec unset_conf!(Latu.Session.t(), String.t()) :: :ok
```

Like `unset_conf/2`, raising on failure.

# `when_matched`

```elixir
@spec when_matched(Latu.MergeInto.t(), atom(), keyword()) :: Latu.MergeInto.t()
```

Add a `WHEN MATCHED` clause: what to do with a source row that has a match in the target.

    Latu.when_matched(merge, :update_all)
    Latu.when_matched(merge, :update, set: [n: col("s.n")], on: expr("s.op = 'U'"))
    Latu.when_matched(merge, :delete, on: expr("s.op = 'D'"))

The action is `:update`, `:update_all` or `:delete`; anything else raises naming the three.

## Options

  * `:set` — the assignments, as a keyword list of `target_column: expression`. **A key is
    a target column name** and travels as an expression string, so a keyword key and a
    string key are the same bytes; anything that is not a name is refused rather than sent.
  * `:on` — an extra condition narrowing the clause to rows that also satisfy it. Defaults
    to `nil`.

`:set` is **required** for `:update` and **refused** for the other two — `:update_all` is
the form that takes the source row whole.

Clauses apply in the order you add them, and **only the first matching clause runs**, which
is why an unconditional one belongs last. Spark enforces that ordering rule for SQL text
only, in its parser, so Latu does not refuse it: the DataFrame path accepts the plan and a
clause after an unconditional one is simply dead.

# `when_not_matched`

```elixir
@spec when_not_matched(Latu.MergeInto.t(), atom(), keyword()) :: Latu.MergeInto.t()
```

Add a `WHEN NOT MATCHED` clause: what to do with a source row that has no match.

    Latu.when_not_matched(merge, :insert_all)
    Latu.when_not_matched(merge, :insert, set: [id: col("s.id"), name: col("s.name")])

The action is `:insert` or `:insert_all` only — there is no row in the target to update or
delete, and Latu refuses those by name rather than letting the server discover it.

## Options

  * `:set` — the assignments, as a keyword list of `target_column: expression`. **A key is
    a target column name** and travels as an expression string, so a keyword key and a
    string key are the same bytes; anything that is not a name is refused rather than sent.
  * `:on` — an extra condition narrowing the clause to rows that also satisfy it. Defaults
    to `nil`.

`:set` is **required** for `:insert` and **refused** for `:insert_all`.

# `when_not_matched_by_source`

```elixir
@spec when_not_matched_by_source(Latu.MergeInto.t(), atom(), keyword()) ::
  Latu.MergeInto.t()
```

Add a `WHEN NOT MATCHED BY SOURCE` clause: what to do with a *target* row that has no match.

    Latu.when_not_matched_by_source(merge, :delete)
    Latu.when_not_matched_by_source(merge, :update, set: [live: false])

The mirror of `when_not_matched/3`, and the clause that makes a merge able to express a full
synchronisation. The action is `:update`, `:update_all` or `:delete`, not the inserts.

## Options

  * `:set` — the assignments, as a keyword list of `target_column: expression`. **A key is
    a target column name** and travels as an expression string, so a keyword key and a
    string key are the same bytes; anything that is not a name is refused rather than sent.
  * `:on` — an extra condition narrowing the clause to rows that also satisfy it. Defaults
    to `nil`.

`:set` is **required** for `:update` and **refused** for the other two.

# `where`

```elixir
@spec where(Latu.DataFrame.t(), term()) :: Latu.DataFrame.t()
```

`filter/2`, spelled Spark's other way.

# `with_checkpoint`

```elixir
@spec with_checkpoint(Latu.DataFrame.t(), keyword(), (Latu.DataFrame.t() -&gt; result)) ::
  {:ok, result} | {:error, Latu.Error.t()}
when result: term()
```

Checkpoint, run your function over the result, and free it on the way out.

    {:ok, counts} =
      Latu.with_checkpoint(expensive, [], fn base ->
        %{all: Latu.count!(base), big: base |> Latu.filter(greater(:n, 100)) |> Latu.count!()}
      end)

The bracket form, as `File.open/3` is to `File.open/2`. The release happens in an `after`, so
it runs even when your function raises — which is the whole point, and the reason to reach
for this over `checkpoint/2` plus `release/1`. A release that itself fails is logged rather
than raised, so it cannot replace your own exception with a duller one.

What you cannot do with it is keep the frame: the checkpoint is gone when the function
returns. Use `checkpoint/2` when you want one to live across REPL prompts.

## Options

`checkpoint/2`'s, in the second argument — `:eager`, `:local`, `:storage_level`. `[]` is the
common case and is not defaulted, so the function is always the last argument and the call
stays pipeable.

# `with_checkpoint!`

```elixir
@spec with_checkpoint!(Latu.DataFrame.t(), keyword(), (Latu.DataFrame.t() -&gt; result)) ::
  result
when result: term()
```

Like `with_checkpoint/3`, raising on failure.

# `with_columns`

```elixir
@spec with_columns(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.DataFrame.t()
```

Add or replace columns, keeping the rest.

    Latu.with_columns(df, doubled: Latu.Column.multiply(:id, 2))

See `Latu.DataFrame.with_columns/2`.

# `with_metadata`

```elixir
@spec with_metadata(Latu.DataFrame.t(), String.t() | atom(), map()) ::
  Latu.DataFrame.t()
```

Attach metadata to an existing column.

    Latu.with_metadata(df, :id, %{"comment" => "the primary key"})

A map, encoded to the JSON string the wire carries. The column keeps its name and its
values; `schema/1` will not show the metadata, because Latu reports a schema as
`simpleString` and Spark keeps metadata outside it — read it back with `explain/2` or from
the source that consumes it.

# `with_stream`

```elixir
@spec with_stream(Latu.DataFrame.t(), keyword(), (Latu.StreamingQuery.t() -&gt; result)) ::
  {:ok, result} | {:error, Latu.Error.t()}
when result: term()
```

Start a streaming write, run your function over the query, and stop it on the way out.

    {:ok, progress} =
      Latu.with_stream(rollups, [format: "parquet", path: out, trigger: :available_now,
                                 checkpoint_location: ckpt], fn query ->
        {:ok, true} = Latu.StreamingQuery.await_termination(query)
        Latu.StreamingQuery.last_progress!(query)
      end)

The bracket form, as `with_checkpoint/3` is to `checkpoint/2`. The stop happens in an `after`,
so a query cannot be left running by a function that raised; a stop that itself fails is
logged rather than raised. What the function does with its time is its own business: wait
for an `:available_now` query to finish, or watch a processing-time one until it has seen
enough.

## Options

`write_stream/2`'s: `:format`, `:output_mode`, `:trigger`, `:query_name`, `:path`, `:table`,
`:partition_by`, `:cluster_by`, `:checkpoint_location`, and writer options.

# `with_stream!`

```elixir
@spec with_stream!(Latu.DataFrame.t(), keyword(), (Latu.StreamingQuery.t() -&gt; result)) ::
  result
when result: term()
```

Like `with_stream/3`, raising on failure.

# `with_watermark`

```elixir
@spec with_watermark(Latu.DataFrame.t(), String.t() | atom(), String.t()) ::
  Latu.DataFrame.t()
```

Mark a streaming frame's event-time column, and how late a row may arrive.

    events
    |> Latu.with_watermark(:ts, "10 seconds")
    |> Latu.group_by([F.window(:ts, "1 minute"), :sku])
    |> Latu.agg(qty: F.sum(:qty))

PySpark's `withWatermark(eventTime, delayThreshold)`, the delay a Spark interval string. What
it buys is bounded state: a windowed aggregate, a stream-stream `join/3` or a
`distinct/3` with `within_watermark: true` can drop what the watermark says is finished.
Nothing on a batch frame; Spark ignores it there.

# `write`

```elixir
@spec write(
  Latu.DataFrame.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}
```

Write to a path. An action: the write runs when called.

    Latu.write(df, format: "parquet", path: "/data/out", mode: :overwrite)
    Latu.write(df, format: "csv", path: "/data/out", header: true,
      partition_by: [:bucket])

Returns `:ok` — a write has no payload, like `show/2`.

## Options

Seven keys are Latu's. **Every other key is a writer option**, with `read/2`'s key and value
rules — a snake_case atom becomes camelCase, a string passes verbatim, a `nil` drops its
pair.

  * `:format` — the sink's short name or class: `"parquet"`, `"csv"`, `"jdbc"`. Defaults to
    `nil`, leaving the server on `spark.sql.sources.default`.
  * `:mode` — what to do when the destination exists. Defaults to `nil`, which sends
    nothing and leaves Spark on its own default of error-if-exists.
    * `:append` — add to what is there.
    * `:overwrite` — replace it.
    * `:error` — fail. Spark's `errorifexists`, and its default.
    * `:ignore` — do nothing and succeed.
  * `:path` — where to write. **Not required** — a JDBC write names its destination in the
    writer options instead. Writing to a *catalog table* is `save_as_table/3` or
    `insert_into/3`, not a key here.
  * `:partition_by` — column names to partition the output by, as a list. Defaults to `[]`.
  * `:sort_by` — column names to sort within each bucket, as a list. Defaults to `[]`.
    Meaningful with `:bucket_by`.
  * `:cluster_by` — column names to cluster by, as a list. Defaults to `[]`.
  * `:bucket_by` — `{buckets, columns}`, e.g. `{8, [:id]}`. Defaults to `nil`; any other
    shape raises.

A writer option whose own name is one of those seven has to be written as a **string key**,
which passes verbatim: `Latu.write(df, [{"path", "s3://bucket/key"}, format: "custom"])`.

## Examples

    Latu.write(df, format: "parquet", path: "/data/out", mode: :overwrite)

    Latu.write(df, format: "csv", path: "/data/out", header: true, partition_by: [:bucket])

    Latu.write(df, format: "parquet", path: "/data/out",
      bucket_by: {8, [:id]}, sort_by: [:ts])

    # a JDBC write names its table in the writer options
    Latu.write(df, format: "jdbc", url: url, dbtable: "people", mode: :append)

# `write!`

```elixir
@spec write!(
  Latu.DataFrame.t(),
  keyword()
) :: :ok
```

Like `write/2`, raising on failure.

# `write_stream`

```elixir
@spec write_stream(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Latu.StreamingQuery.t()} | {:error, Latu.Error.t()}
```

Start a streaming write. An action: the query starts when called, and what comes back is a
`%Latu.StreamingQuery{}` for the query now running on the server.

    {:ok, query} =
      Latu.write_stream(rollups,
        format: "parquet", path: "/data/rollups", output_mode: :append,
        trigger: :available_now, checkpoint_location: "/data/ckpt", query_name: "rollups")

    {:ok, true} = Latu.StreamingQuery.await_termination(query)

PySpark's `df.writeStream...start()` and `.toTable()`, one call. The frame is a streaming
one from `read/2` or `table/3` with `is_streaming: true`. **The query outlives this process**:
stop it with `Latu.StreamingQuery.stop/1`, scope it with `with_stream/3`, or find it again
with `Latu.StreamingQuery.active/1`. Its failure raises nothing here; see
`Latu.StreamingQuery`.

## Options

Nine keys are Latu's. **Every other key is a writer option**, with `read/2`'s key and value
rules.

  * `:format` — the sink's short name or class: `"parquet"`, `"kafka"`, `"console"`.
    Defaults to `nil`, leaving the server on its default.
  * `:output_mode` — what each trigger writes. Defaults to `nil`, which leaves Spark on its
    own default of `:append`.
    * `:append` — new rows only; the only mode without aggregation.
    * `:complete` — the whole result every trigger; aggregations only.
    * `:update` — the rows that changed.
  * `:trigger` — when a batch runs. Defaults to `nil`, Spark's default of a processing-time
    trigger with no interval, one batch straight after another.
    * `:available_now` — process everything the source has now, in batches, then stop.
      Incremental ETL, and the shape a test wants: it terminates by itself.
    * `{:processing_time, interval}` — a batch every `interval`, a Spark duration string
      like `"10 seconds"`.
    * `{:continuous, interval}` — continuous processing, checkpointing every `interval`.
    * `:once` — one batch, then stop. Deprecated by Spark in favour of `:available_now`,
      and offered because the wire still carries it.
  * `:query_name` — a name for the query, shown by `Latu.StreamingQuery.active/1` and the
    Spark UI. Defaults to `nil`.
  * `:path` — the sink's directory, for a file sink. Defaults to `nil`.
  * `:table` — a catalog table to write to, PySpark's `toTable`. Defaults to `nil`. Never
    with `:path`; neither is fine for a sink that names its destination in options
    (`console`, `memory`, `kafka`).
  * `:partition_by` — column names to partition the sink by, as a list. Defaults to `[]`.
  * `:cluster_by` — column names to cluster by, as a list. Defaults to `[]`.
  * `:checkpoint_location` — where Spark keeps the query's offsets and state, so a restart
    resumes. Defaults to `nil`; every query that is not a throwaway sets it. The
    `checkpointLocation` writer option, reserved because of that, and nothing to do with
    `checkpoint/2`.

`foreach` and `foreachBatch` are not offered: both carry a serialised closure, which no
Elixir client can build. `docs/deviations.md` has the alternatives.

# `write_stream!`

```elixir
@spec write_stream!(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.StreamingQuery.t()
```

Like `write_stream/2`, raising on failure.

# `write_v2`

```elixir
@spec write_v2(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}
```

Write to a table through Spark's v2 API (`df.writeTo` in PySpark). An action.

    Latu.write_v2(df, "people", mode: :create, using: "parquet")
    Latu.write_v2(df, "people", mode: :overwrite, condition: equal(:day, "2026-09-01"))

## Options

**Every other key is a writer option**, with `read/2`'s key and value rules.

  * `:mode` — **required**, one per PySpark terminal method.
    * `:create` — create the table; fail if it exists. `.create()`
    * `:replace` — replace it; fail if it does not exist. `.replace()`
    * `:create_or_replace` — either. `.createOrReplace()`
    * `:append` — add rows. `.append()`
    * `:overwrite` — replace the rows matching `:condition`. `.overwrite(cond)`
    * `:overwrite_partitions` — replace the partitions the data touches.
      `.overwritePartitions()`
  * `:condition` — the overwrite predicate, as an expression. Defaults to `nil`, and pairs
    with `mode: :overwrite` only.
  * `:using` — the provider, e.g. `"parquet"`. Defaults to `nil`.
  * `:partition_by` — **expressions**, not just names, so a transform like
    `Latu.Column.fun("years", [:ts])` works. Defaults to `[]`.
  * `:cluster_by` — column names. Defaults to `[]`.
  * `:table_properties` — a keyword list whose keys pass verbatim. Defaults to `[]`.

A writer option whose own name is one of those six has to be written as a **string key**,
which passes verbatim.

## Examples

    Latu.write_v2(df, "people", mode: :create, using: "parquet")
    Latu.write_v2(df, "people", mode: :overwrite, condition: equal(:day, "2026-09-01"))
    Latu.write_v2(df, "events", mode: :create,
      partition_by: [Latu.Column.fun("years", [:ts])],
      table_properties: ["write.format.default": "parquet"])

# `write_v2!`

```elixir
@spec write_v2!(Latu.DataFrame.t(), String.t() | atom(), keyword()) :: :ok
```

Like `write_v2/3`, raising on failure.

# `write_v2_with_metrics`

```elixir
@spec write_v2_with_metrics(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`write_v2/3`, and the metrics `observe/3` asked for. See `observe/3`.

# `write_v2_with_metrics!`

```elixir
@spec write_v2_with_metrics!(Latu.DataFrame.t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()
```

Like `write_v2_with_metrics/3`, raising on failure.

# `write_with_metrics`

```elixir
@spec write_with_metrics(
  Latu.DataFrame.t(),
  keyword()
) :: {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`write/2`, and the metrics `observe/3` asked for. See `observe/3`.

# `write_with_metrics!`

```elixir
@spec write_with_metrics!(
  Latu.DataFrame.t(),
  keyword()
) :: Latu.ExecutionInfo.t()
```

Like `write_with_metrics/2`, raising on failure.

# `zip_with_index`

```elixir
@spec zip_with_index(Latu.DataFrame.t(), String.t() | atom()) :: Latu.DataFrame.t()
```

Add a column of consecutive indices, starting at 0.

    Latu.zip_with_index(df)
    Latu.zip_with_index(df, :row_num)

Spark has no relation for this and PySpark has no function either: it is a projection of
every column plus `distributed_sequence_id`, an internal expression registered for
pandas-on-Spark. Hidden from `DESCRIBE FUNCTION`, so it is not in `Latu.Functions` — this
verb is the way to reach it.

The indices are consecutive across the whole frame, not per partition, which is what makes
it different from `F.monotonically_increasing_id/0`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*