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

A lazy DataFrame: a session and an inert plan.

Building one does no IO and touches no process. Nothing reaches the server until an action,
such as `Latu.show/2` or `Latu.collect/2`.

# `t`

```elixir
@type t() :: %Latu.DataFrame{plan: Latu.Plan.relation(), session: Latu.Session.t()}
```

# `agg`

```elixir
@spec agg(t(), term()) :: t()
```

Aggregate the whole frame, with no grouping.

    Latu.agg(df, total: F.sum(:price))

The same `Aggregate` relation `group_by/2` builds, with no grouping expressions — which is
what PySpark's `df.agg(...)` sends too.

# `approx_quantile`

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

See `Latu.approx_quantile/5`.

# `approx_quantile!`

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

Like `approx_quantile/5`, raising on failure.

# `as`

```elixir
@spec as(t(), String.t() | atom()) :: 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. See `Latu.Plan.as/2`.

# `cache`

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

See `Latu.cache/1`.

# `cache!`

```elixir
@spec cache!(t()) :: t()
```

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

# `checkpoint`

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

See `Latu.checkpoint/2`.

# `checkpoint!`

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

Like `checkpoint/2`, raising on failure.

# `coalesce`

```elixir
@spec coalesce(t(), pos_integer()) :: t()
```

Fewer partitions without a shuffle. `repartition/2` with `shuffle: false`.

# `col`

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

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

    Latu.col(orders, :id)

Needed only when two DataFrames in one pipeline share a column name, and it has to be a
DataFrame the plan already contains: a self-join is fine — both branches *are* the relation —
but **selecting one frame's column from another is refused by Spark**
(`CANNOT_RESOLVE_DATAFRAME_COLUMN`), hoisted or not. Measured; `docs/decisions.md` (M9.1).
For a value from another frame, use a subquery — `scalar/1`, `exists/1`, or
`Latu.Column.isin/2` over a DataFrame.

`col(df, "*")` is every column of this frame and no other's — PySpark's `df["*"]`, and the
way to keep one side of a join whole.

# `col_regex`

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

See `Latu.col_regex/2`.

# `collect`

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

All the rows, as maps.

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

Atom keys by default: they pattern-match, and the atom table only grows by the set of
column names ever selected. `keys: :strings` when names come out of dynamic SQL.

The whole result is held at once; `stream/2` is the lazy escape for results that do not
fit.

# `collect!`

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

Like `collect/2`, raising on failure.

# `collect_with_metrics`

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

`collect/2`, and what the run reported besides the rows.

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

    info.observed  #=> %{checks: %{total: 4}}
    info.metrics   #=> Spark's own per-node SQL metrics

See `Latu.ExecutionInfo`. Options are `collect/2`'s.

# `collect_with_metrics!`

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

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

# `columns`

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

See `Latu.columns/1`.

# `columns!`

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

Like `columns/1`, raising on failure.

# `corr`

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

See `Latu.corr/4`.

# `corr!`

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

Like `corr/4`, raising on failure.

# `count`

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

How many rows, counted by the server.

    Latu.count(df)  #=> {:ok, 10}

PySpark's own composition — `agg(count(lit(1)))`, unaliased and all — pinned by the
`count_action` fixture.

# `count!`

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

Like `count/2`, raising on failure.

# `count_with_metrics`

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

`count/2`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

# `count_with_metrics!`

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

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

# `cov`

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

See `Latu.cov/4`.

# `cov!`

```elixir
@spec cov!(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, t()} | {:error, Latu.Error.t()}
```

See `Latu.create_dataframe/3`.

# `create_dataframe!`

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

Like `create_dataframe/3`, raising on failure.

# `create_temp_view`

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

See `Latu.create_temp_view/3`.

# `create_temp_view!`

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

Like `create_temp_view/3`, raising on failure.

# `cross_join`

```elixir
@spec cross_join(t(), t()) :: t()
```

See `Latu.cross_join/2`.

# `crosstab`

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

See `Latu.crosstab/3`.

# `cube`

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

Group by every combination of these columns.

# `describe`

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

See `Latu.describe/2`.

# `distinct`

```elixir
@spec distinct(t(), term()) :: t()
```

Drop duplicate rows.

    Latu.distinct(df)              # every column is a key
    Latu.distinct(df, [:suburb])   # these columns are

# `drop`

```elixir
@spec drop(t(), term()) :: t()
```

Remove columns.

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

# `drop_na`

```elixir
@spec drop_na(
  t(),
  keyword()
) :: t()
```

See `Latu.drop_na/2`.

# `dtypes`

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

See `Latu.dtypes/1`.

# `dtypes!`

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

Like `dtypes/1`, raising on failure.

# `except`

```elixir
@spec except(t(), t(), keyword()) :: t()
```

Rows in the first and not the second.

Distinct unless `all: true`. PySpark spells these `subtract` and `exceptAll`; `except` is
Spark's own Scala name and SQL's.

# `exists`

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

A predicate that holds when this DataFrame has any rows at all.

    Latu.filter(orders, Latu.exists(Latu.filter(alerts, :open)))

Hoisted like `scalar/1`, including its note about sessions. Not to be confused with
`Latu.Functions.exists/2`, which is Spark's higher-order function over an array — Spark named
both.

# `explain`

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

See `Latu.explain/2`.

# `explain!`

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

Like `explain/2`, raising on failure.

# `explain_string`

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

See `Latu.explain_string/2`.

# `explain_string!`

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

Like `explain_string/2`, raising on failure.

# `fill_na`

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

See `Latu.fill_na/3`.

# `filter`

```elixir
@spec filter(t(), term()) :: t()
```

Keep the rows the condition holds for.

A string is SQL, parsed by the server — `filter(df, "id > 3")` is `filter(df, expr("id >
3"))`. This is the only position where a string means SQL: in `select/2` it is a column name,
and inside an expression it is a literal. PySpark reads all three the same way.

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

# `first`

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

The first row, or nil when there are none. Options are `collect/2`'s.

# `first!`

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

Like `first/2`, raising on failure.

# `freq_items`

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

See `Latu.freq_items/3`.

# `glimpse`

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

See `Latu.glimpse/2`.

# `glimpse!`

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

Like `glimpse/2`, raising on failure.

# `group_by`

```elixir
@spec group_by(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: F.sum(:price))

Spark has no `group_by` relation, so nothing is built until `agg/2`.

# `grouping_sets`

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

See `Latu.grouping_sets/3`.

# `head`

```elixir
@spec head(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 are `collect/2`'s.

# `head!`

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

Like `head/3`, raising on failure.

# `hint`

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

See `Latu.hint/3`.

# `input_files`

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

See `Latu.input_files/1`.

# `input_files!`

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

Like `input_files/1`, raising on failure.

# `insert_into`

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

See `Latu.insert_into/3`.

# `insert_into!`

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

Like `insert_into/3`, raising on failure.

# `insert_into_with_metrics`

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

`insert_into/3`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

# `insert_into_with_metrics!`

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

Like `insert_into_with_metrics/3`, raising on failure.

# `intersect`

```elixir
@spec intersect(t(), t(), keyword()) :: t()
```

Rows in both. Distinct unless `all: true`, which is PySpark's `intersectAll`.

# `is_empty`

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

See `Latu.is_empty/1`.

# `is_empty!`

```elixir
@spec is_empty!(t()) :: boolean()
```

Like `is_empty/1`, raising on failure.

# `is_local`

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

See `Latu.is_local/1`.

# `is_local!`

```elixir
@spec is_local!(t()) :: boolean()
```

Like `is_local/1`, raising on failure.

# `is_streaming`

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

See `Latu.is_streaming/1`.

# `is_streaming!`

```elixir
@spec is_streaming!(t()) :: boolean()
```

Like `is_streaming/1`, raising on failure.

# `join`

```elixir
@spec join(t(), t(), keyword()) :: t()
```

Join two DataFrames.

    Latu.join(orders, customers, on: :customer_id)
    Latu.join(orders, customers, on: expr("o.id = c.id"), how: :left)

# `join_as_of`

```elixir
@spec join_as_of(t(), t(), keyword()) :: t()
```

See `Latu.join_as_of/3`.

# `lateral_join`

```elixir
@spec lateral_join(t(), t(), keyword()) :: t()
```

See `Latu.lateral_join/3`.

# `limit`

```elixir
@spec limit(t(), non_neg_integer()) :: t()
```

Keep at most `count` rows.

# `merge`

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

See `Latu.merge/2`.

# `merge!`

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

Like `merge/2`, raising on failure.

# `merge_into`

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

See `Latu.merge_into/4`.

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

# `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(t(), String.t() | atom()) :: Latu.Plan.expression()
```

See `Latu.metadata_column/2`.

# `nearest_by_join`

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

See `Latu.nearest_by_join/4`.

# `observe`

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

See `Latu.observe/3`.

# `offset`

```elixir
@spec offset(t(), non_neg_integer()) :: t()
```

Skip the first `count` rows.

# `order_by`

```elixir
@spec order_by(t(), term()) :: t()
```

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

# `parse`

```elixir
@spec parse(
  t(),
  keyword()
) :: t()
```

See `Latu.parse/2`.

# `persist`

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

See `Latu.persist/2`.

# `persist!`

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

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

# `print_schema`

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

See `Latu.print_schema/2`.

# `print_schema!`

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

Like `print_schema/2`, raising on failure.

# `random_split`

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

See `Latu.random_split/3`.

# `range`

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

See `Latu.range/2`.

# `read`

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

See `Latu.read/2`.

# `release`

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

See `Latu.release/1`.

# `release!`

```elixir
@spec release!(t()) :: :ok
```

Like `release/1`, raising on failure.

# `rename`

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

Rename columns.

    Latu.rename(df, id: :n)         # by mapping, leaving the rest
    Latu.rename(df, [:renamed])     # positionally, one name per column

Two Spark relations behind one verb: `WithColumnsRenamed` for pairs, `ToDF` for a plain list.
`Explorer.DataFrame.rename/2` reads both shapes the same way.

# `repartition`

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

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

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

Two Spark relations: `Repartition` for a count alone, `RepartitionByExpression` once columns
are named.

# `repartition`

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

See `repartition/2`.

# `repartition_by_range`

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

See `Latu.repartition_by_range/3`.

# `replace`

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

See `Latu.replace/3`.

# `rollup`

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

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

# `same_semantics`

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

See `Latu.same_semantics/2`.

# `same_semantics!`

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

Like `same_semantics/2`, raising on failure.

# `sample`

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

A random fraction of the rows.

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

Without a `:seed` a random one is drawn, as in PySpark, so the plan differs between runs.

# `sample_by`

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

See `Latu.sample_by/4`.

# `save_as_table`

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

See `Latu.save_as_table/3`.

# `save_as_table!`

```elixir
@spec save_as_table!(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(t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`save_as_table/3`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

# `save_as_table_with_metrics!`

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

Like `save_as_table_with_metrics/3`, raising on failure.

# `scalar`

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

This DataFrame as a scalar subquery: a single value, usable wherever a value belongs.

    totals = Latu.agg(orders, total: F.sum(:amount))
    Latu.filter(orders, greater(:amount, Latu.scalar(totals)))

The frame is hoisted into the plan that uses it, so the whole thing is one query and the two
frames need no relationship beyond sharing a session. Spark refuses it at analysis if the
subquery yields more than one row or column.

This is the reference Spark resolves; a bare `col/2` pointing outside its own tree is not.

Unlike `join/3` and the set operations, this does **not** check that both frames come from
one session: the referenced plan travels inline, so a cross-session subquery still executes.
What does not travel is session-scoped state — a temp view the other frame reads, an artifact
behind its local data, a conf set on that session — and Spark names whatever is missing.
`docs/decisions.md` (M9.3).

# `schema`

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

See `Latu.schema/1`.

# `schema!`

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

Like `schema/1`, raising on failure.

# `select`

```elixir
@spec select(t(), term()) :: t()
```

Keep these columns, in this order.

A string or an atom is a column name; anything else is an expression. Trailing keywords name
what they hold:

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

A single column needs no list.

# `select_expr`

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

See `Latu.select_expr/2`.

# `semantic_hash`

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

See `Latu.semantic_hash/1`.

# `semantic_hash!`

```elixir
@spec semantic_hash!(t()) :: integer()
```

Like `semantic_hash/1`, raising on failure.

# `show`

```elixir
@spec show(
  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, Latu decodes
one string cell. Options, all PySpark's:

  * `:num_rows` — how many rows, default 20
  * `:truncate` — cell width, default 20; `true` means 20 and `false` means no truncation
  * `:vertical` — one row per block, default false

Pipelines want `Kernel.tap/2`:

    df |> tap(&Latu.show!/1) |> Latu.filter(...)

# `show!`

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

Like `show/2`, raising on failure.

# `sort`

```elixir
@spec sort(t(), term()) :: t()
```

Sort rows.

    Latu.sort(df, :id)                    # ascending, nulls first
    Latu.sort(df, [desc(:price), :id])
    Latu.order_by(df, "id")

A bare name sorts ascending with nulls first, as PySpark's `orderBy("id")` does.
`Latu.Column.asc/1`, `Latu.Column.desc/1` and the four explicit `*_nulls_*` spellings are
the alternatives.

# `sort_within_partitions`

```elixir
@spec sort_within_partitions(t(), term()) :: t()
```

Sort within each partition, leaving the partitions unordered.

The same relation as `sort/2` with `is_global: false`, and cheaper: no shuffle.

# `sql`

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

See `Latu.sql/3`.

# `sql!`

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

Like `sql/3`, raising on failure.

# `storage_level`

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

See `Latu.storage_level/1`.

# `storage_level!`

```elixir
@spec storage_level!(t()) :: map()
```

Like `storage_level/1`, raising on failure.

# `stream`

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

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

Backpressure for results too large to hold: each batch decodes as it arrives, and stopping
early releases the execution. Raises `Latu.Error` on failure, since an enumeration has no
way to return one. The schema guard runs on the `DataType` the server sends ahead of the
first batch.

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

# `stream_nx`

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

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

Backpressure for results too large to hold, as `stream/2` is for Explorer. Each batch decodes
on its own, so the tensors are per batch and stacking them is the caller's business — that is
the difference from `to_nx/2`, which concatenates. Raises `Latu.Error` on failure, since an
enumeration has no way to return one.

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

# `summary`

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

See `Latu.summary/2`.

# `table`

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

See `Latu.table/2`.

# `table_changes`

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

See `Latu.table_changes/3`.

# `table_function`

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

See `Latu.table_function/3`.

# `tail`

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

See `Latu.tail/3`.

# `tail!`

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

Like `tail/3`, raising on failure.

# `take`

```elixir
@spec take(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 are
`collect/2`'s.

# `take!`

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

Like `take/3`, raising on failure.

# `to`

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

See `Latu.to/2`.

# `to_arrow`

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

The raw Arrow IPC streaming-format binaries, one per batch, for doing your own thing.

Bypasses the decoder AND the schema guard on purpose: these bytes are headed for some other
Arrow reader, whose capabilities are its own business. Each binary is a complete IPC stream
— schema, record batch, end marker — and they must never be byte-concatenated.

# `to_arrow!`

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

Like `to_arrow/2`, raising on failure.

# `to_explorer`

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

The result as one `Explorer.DataFrame`.

**Unbounded**, like `collect/2`, `to_arrow/2` and Spark's own `collect`: the whole result
comes back. To take part of it, bound the *plan* — which is how Spark does it, and what
`limit/2` is for. A result too large to hold at all is what `stream/2` is for.

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

An empty result is a 0-row frame with the right columns and dtypes — the server sends the
Arrow schema even when there are no rows.

# `to_explorer!`

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

Like `to_explorer/2`, raising on failure.

# `to_explorer_with_metrics`

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

`to_explorer/2`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

# `to_explorer_with_metrics!`

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

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

# `to_html`

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

See `Latu.to_html/2`.

# `to_html!`

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

Like `to_html/2`, raising on failure.

# `to_nx`

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

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

Bypasses the Explorer decoder and the schema guard, as `to_arrow/2` does, and for the same
reason: these bytes are read by `Latu.Result.Arrow` rather than by Polars, and what Polars
cannot take is not this path's concern. That is what makes a `Vector` column readable here
when `collect/2` and `to_explorer/2` both refuse it.

Two shapes decode. A numeric column with no nulls becomes a 1-D tensor, and a column of
equal-length numeric lists — or of dense `Vector`s — becomes one `{rows, width}` tensor.
Anything else is refused by name: nulls, strings, booleans, ragged lists, sparse vectors.

**Unbounded**, like `collect/2` and `to_arrow/2`: bound the plan, or use `stream_nx/2`.

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

Needs the optional `:nx` dependency; without it this says so rather than failing obscurely.

# `to_nx!`

```elixir
@spec to_nx!(
  t(),
  keyword()
) :: %{required(String.t()) => term()}
```

Like `to_nx/2`, raising on failure.

# `transpose`

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

See `Latu.transpose/2`.

# `tree_string`

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

See `Latu.tree_string/2`.

# `tree_string!`

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

Like `tree_string/2`, raising on failure.

# `union`

```elixir
@spec union(t(), t(), keyword()) :: t()
```

All the rows of both, duplicates kept.

  * `:all` — default `true`, as Spark's `union` is `UNION ALL` rather than SQL's `UNION`
  * `:by_name` — match columns by name rather than position
  * `:allow_missing_columns` — fill a missing column with null; needs `:by_name`

    Latu.union(df, other)
    Latu.union(df, other, by_name: true)

# `unpersist`

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

See `Latu.unpersist/2`.

# `unpersist!`

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

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

# `unpivot`

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

See `Latu.unpivot/3`.

# `where`

```elixir
@spec where(t(), term()) :: t()
```

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

# `with_checkpoint`

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

See `Latu.with_checkpoint/3`.

# `with_checkpoint!`

```elixir
@spec with_checkpoint!(t(), keyword(), (t() -> result)) :: result when result: term()
```

Like `with_checkpoint/3`, raising on failure.

# `with_columns`

```elixir
@spec with_columns(
  t(),
  keyword()
) :: t()
```

Add or replace columns, keeping the rest.

    Latu.with_columns(df, doubled: multiply(:id, 2))
    Latu.with_columns(df, a: add(:id, 1), b: subtract(:id, 1))

A keyword list, because it is ordered and a map is not. There is no `with_column`: Spark
has no singular relation, and the keyword form is already short.

# `with_metadata`

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

See `Latu.with_metadata/3`.

# `write`

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

See `Latu.write/2`.

# `write!`

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

Like `write/2`, raising on failure.

# `write_v2`

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

See `Latu.write_v2/3`.

# `write_v2!`

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

Like `write_v2/3`, raising on failure.

# `write_v2_with_metrics`

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

`write_v2/3`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

# `write_v2_with_metrics!`

```elixir
@spec write_v2_with_metrics!(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(
  t(),
  keyword()
) :: {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}
```

`write/2`, and what the run reported besides the result. See `Latu.ExecutionInfo`.

This is the shape `observe` was built for: a data-quality aggregate attached on the way in,
the write done, the counts read back. There are no rows to return, so the metrics are the
whole result.

    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}}

# `write_with_metrics!`

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

Like `write_with_metrics/2`, raising on failure.

# `zip_with_index`

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

See `Latu.zip_with_index/2`.

---

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