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

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

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

Like approx_quantile/5, raising on failure.

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

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

See Latu.cache/1.

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

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

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

See Latu.checkpoint/2.

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

Like checkpoint/2, raising on failure.

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

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

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

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

See Latu.col_regex/2.

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

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

Like collect/2, raising on failure.

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

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

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

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

See Latu.columns/1.

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

Like columns/1, raising on failure.

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

See Latu.corr/4.

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

Like corr/4, raising on failure.

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

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

Like count/2, raising on failure.

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

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

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

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

See Latu.cov/4.

@spec cov!(t(), String.t() | atom(), String.t() | atom(), keyword()) :: float()

Like cov/4, raising on failure.

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

See Latu.create_dataframe/3.

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

Like create_dataframe/3, raising on failure.

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

See Latu.create_temp_view/3.

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

Like create_temp_view/3, raising on failure.

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

See Latu.cross_join/2.

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

See Latu.crosstab/3.

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

Group by every combination of these columns.

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

See Latu.describe/2.

@spec distinct(t(), term(), keyword()) :: t()

Drop duplicate rows.

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

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

Remove columns.

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

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

See Latu.drop_na/2.

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

See Latu.dtypes/1.

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

Like dtypes/1, raising on failure.

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

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

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

See Latu.explain/2.

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

Like explain/2, raising on failure.

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

See Latu.explain_string/2.

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

Like explain_string/2, raising on failure.

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

See Latu.fill_na/3.

@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")

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

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

Like first/2, raising on failure.

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

See Latu.freq_items/3.

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

See Latu.glimpse/2.

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

Like glimpse/2, raising on failure.

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

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

See Latu.grouping_sets/3.

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

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

Like head/3, raising on failure.

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

See Latu.hint/3.

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

See Latu.input_files/1.

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

Like input_files/1, raising on failure.

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

See Latu.insert_into/3.

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

Like insert_into/3, raising on failure.

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

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

Like insert_into_with_metrics/3, raising on failure.

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

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

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

See Latu.is_empty/1.

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

Like is_empty/1, raising on failure.

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

See Latu.is_local/1.

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

Like is_local/1, raising on failure.

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

See Latu.is_streaming/1.

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

Like is_streaming/1, raising on failure.

@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)

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

See Latu.join_as_of/3.

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

See Latu.lateral_join/3.

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

Keep at most count rows.

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

See Latu.merge/2.

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

Like merge/2, raising on failure.

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

See Latu.merge_into/4.

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

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

Like merge_with_metrics/2, raising on failure.

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

See Latu.metadata_column/2.

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

See Latu.nearest_by_join/4.

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

See Latu.observe/3.

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

Skip the first count rows.

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

sort/2, spelled Spark's other way.

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

See Latu.parse/2.

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

See Latu.persist/2.

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

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

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

See Latu.print_schema/2.

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

Like print_schema/2, raising on failure.

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

See Latu.random_split/3.

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

See Latu.range/2.

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

See Latu.read/2.

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

See Latu.release/1.

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

Like release/1, raising on failure.

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

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

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

See repartition/2.

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

See Latu.repartition_by_range/3.

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

See Latu.replace/3.

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

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

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

See Latu.same_semantics/2.

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

Like same_semantics/2, raising on failure.

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

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

See Latu.sample_by/4.

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

See Latu.save_as_table/3.

@spec save_as_table!(t(), String.t() | atom(), keyword()) :: :ok

Like save_as_table/3, raising on failure.

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

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

Like save_as_table_with_metrics/3, raising on failure.

@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).

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

See Latu.schema/1.

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

Like schema/1, raising on failure.

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

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

See Latu.select_expr/2.

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

See Latu.semantic_hash/1.

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

Like semantic_hash/1, raising on failure.

@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(...)

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

Like show/2, raising on failure.

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

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

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

See Latu.sql/3.

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

Like sql/3, raising on failure.

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

See Latu.storage_level/1.

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

Like storage_level/1, raising on failure.

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

The result as a lazy stream of Explorer.DataFrames, one per Arrow batch.

Lazy: each batch decodes as it arrives and stopping early releases the execution, so you never hold the whole decoded result. It does not bound what is received: the transport keeps taking the rest of the result as fast as the server sends it, whether or not you are consuming, so a slow consumer can hold the remaining raw Arrow bytes. :window_size tunes HTTP/2 throughput and is not a memory limit (docs/decisions.md, 2026-09-17). 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()

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

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

Like stream/2 but yielding tensors. Each batch decodes on its own, so the tensors are per batch and stacking them is the caller's business — the difference from to_nx/2, which concatenates. The same transport caveat applies: not bounded-memory (docs/decisions.md, 2026-09-17). Raises Latu.Error on failure, since an enumeration has no way to return one. An empty batch (a partition with no rows) yields nothing.

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

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

See Latu.summary/2.

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

See Latu.table/2.

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

See Latu.table_changes/3.

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

See Latu.table_function/3.

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

See Latu.tail/3.

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

Like tail/3, raising on failure.

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

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

Like take/3, raising on failure.

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

See Latu.to/2.

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

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

Like to_arrow/2, raising on failure.

@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 decode at once is what stream/2 is for; it does not shrink what is received (see there).

{: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.

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

Like to_explorer/2, raising on failure.

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

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

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

See Latu.to_html/2.

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

Like to_html/2, raising on failure.

@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 Vectors — 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.

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

Like to_nx/2, raising on failure.

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

See Latu.transpose/2.

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

See Latu.tree_string/2.

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

Like tree_string/2, raising on failure.

@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)

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

See Latu.unpersist/2.

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

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

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

See Latu.unpivot/3.

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

filter/2, spelled Spark's other way.

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

See Latu.with_checkpoint/3.

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

Like with_checkpoint/3, raising on failure.

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

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

See Latu.with_metadata/3.

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

See Latu.with_stream/3.

@spec with_stream!(t(), keyword(), (Latu.StreamingQuery.t() -> result)) :: result
when result: term()

Like with_stream/3, raising on failure.

@spec with_watermark(t(), String.t() | atom(), String.t()) :: t()

See Latu.with_watermark/3.

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

See Latu.write/2.

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

Like write/2, raising on failure.

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

See Latu.write_stream/2.

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

Like write_stream/2, raising on failure.

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

See Latu.write_v2/3.

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

Like write_v2/3, raising on failure.

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

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

Like write_v2_with_metrics/3, raising on failure.

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

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

Like write_with_metrics/2, raising on failure.

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

See Latu.zip_with_index/2.