# `Latu`
[🔗](https://github.com/zero-one-group/latu/blob/v0.4.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 linked to whoever
consumes 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.

That is the line structured streaming falls the wrong side of, and why it is a separate
package: a checkpoint is data at rest and can be bracketed, a streaming query is a running
computation that outlives any bracket — a lifecycle, which wants an owner.
`docs/decisions.md` has the argument, and why MLlib is separate on different grounds.

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).
  * `: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.

# `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 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()) :: Latu.DataFrame.t()
```

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

    Latu.distinct(df)
    Latu.distinct(df, [:suburb])

See `Latu.DataFrame.distinct/2`.

# `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 — and the **message whole**, where the gRPC status carries the server's
2048-character abbreviation of it.

So it is an explicit call rather than something every failure pays for. PySpark fetches it
eagerly on every error; a Latu action returns `{:error, _}` for expected refusals too, and
spending 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. Latu does not build one yet; the answer is no.

# `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.

A reader option whose own name is one of those four 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 hold at once; stopping early releases the execution. 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")

Options (`table/3`) follow `read/2`'s key and value rules.

# `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 hold.

## 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
hold.

## 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.

# `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_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*