Latu.Result (latu v0.1.0)

Copy Markdown View Source

Arrow batches in, Explorer out — and, for Latu.create_dataframe/3, the other direction.

This module is Latu's Explorer boundary in both directions; nothing else touches Arrow.

It also owns the shape a schema comes back as. Latu.schema/1 and Latu.parse_ddl/2 return a list of field/0, each carrying Spark's own name for its type, as a string: there is no client-side type model in either direction — Latu.create_dataframe/3 takes a DDL string going out, and this is what comes back (M8.1 and M10.1; docs/decisions.md). Nested types render into the type string — array<int>, struct<a:int,b:string> — so a schema stays a flat, pattern-matchable list however deep the data is.

A schema is also checked before any bytes are decoded: Spark's interval types panic inside Polars' NIF rather than failing cleanly, so a result carrying one is refused, by column name, using the DataType the server sends ahead of the first batch.

Summary

Types

One top-level column: its name, Spark's name for its type, and its nullability.

Functions

Decode Arrow batches into one Explorer DataFrame.

An Explorer DataFrame from column data — {name, values} pairs, in the given order.

The one cell of a 1x1 result, positionally — PySpark's table[0][0].

The one cell of a 1x1 result, as ShowString and HtmlString produce.

The rows of a decoded frame, as maps.

Row count, so callers outside the Explorer boundary need no Explorer call.

One Arrow IPC stream for the whole frame — exactly the bytes LocalRelation.data carries.

The frame as row slices, each dumped as its own complete IPC stream.

Types

field()

@type field() :: %{name: String.t(), type: String.t(), nullable: boolean()}

One top-level column: its name, Spark's name for its type, and its nullability.

Functions

decode(batches, opts \\ [])

@spec decode(
  [Latu.Client.batch()],
  keyword()
) :: {:ok, Explorer.DataFrame.t()} | {:error, Latu.Error.t()}

Decode Arrow batches into one Explorer DataFrame.

One load_ipc_stream per batch. Each batch is a complete IPC stream with its own end-of- stream marker, so concatenating the binaries first would silently keep the first batch's rows and drop the rest — the decoded row count is checked against the server's for that reason.

Options: :columns, passed to Explorer at load time so pruned columns are never deserialized.

A payload Polars cannot parse raises rather than returning an error; Spark's interval types are the live hazard, and the schema guard described above is what heads them off — callers run it on the execution's schema before handing bytes here.

from_columns(frame)

@spec from_columns([{atom() | String.t(), list()}] | Explorer.DataFrame.t()) ::
  Explorer.DataFrame.t()

An Explorer DataFrame from column data — {name, values} pairs, in the given order.

The encode side of Latu.create_dataframe/3; ragged or untypeable columns raise, as Explorer.DataFrame.new/1 does.

only(frame)

@spec only(Explorer.DataFrame.t()) :: {:ok, term()} | {:error, Latu.Error.t()}

The one cell of a 1x1 result, positionally — PySpark's table[0][0].

only(frame, column)

@spec only(Explorer.DataFrame.t(), String.t()) ::
  {:ok, term()} | {:error, Latu.Error.t()}

The one cell of a 1x1 result, as ShowString and HtmlString produce.

rows(frame, keys)

@spec rows(Explorer.DataFrame.t(), :atoms | :strings) :: [map()]

The rows of a decoded frame, as maps.

Streamed out in chunks rather than through Explorer.DataFrame.to_rows/2, which converts every column to a full list before emitting one row. :atoms interns the column names — bounded by the schemas ever selected, not by data.

size(frame)

Row count, so callers outside the Explorer boundary need no Explorer call.

to_ipc(frame)

@spec to_ipc(Explorer.DataFrame.t()) :: binary()

One Arrow IPC stream for the whole frame — exactly the bytes LocalRelation.data carries.

Raises on a frame Explorer cannot serialize; that is an argument problem, not a transport one.

to_ipc_chunks(frame, rows_per_chunk)

@spec to_ipc_chunks(Explorer.DataFrame.t(), pos_integer()) :: [binary()]

The frame as row slices, each dumped as its own complete IPC stream.

Every chunk must independently decode — the server caches each as a separate artifact and reads them back by hash (ChunkedCachedLocalRelation) — which is why this slices and re-dumps rather than splitting to_ipc/1's bytes.