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

Every query running on the session, started by this client or not.

The recovery route: a query outlives the process that started it, and this is how the next process finds it.

@spec active!(Latu.Session.t()) :: [t()]

Like active/1, raising on failure.

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

Wait for any query on the session to end. await_termination/2's loop and options, over the session.

Sticky, as Spark's is: once any query has terminated since the session started or since reset_terminated/1, it answers {:ok, true} at once. A query that failed comes back as {:error, _} carrying its exception, again as Spark's does.

Options

await_termination/2's: :timeout, :interval.

@spec await_any_termination!(
  Latu.Session.t(),
  keyword()
) :: boolean()

Like await_any_termination/2, raising on failure.

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

Wait for the query to end.

{:ok, true} = Latu.StreamingQuery.await_termination(query)
{:ok, false} = Latu.StreamingQuery.await_termination(query, timeout: 30_000)

{:ok, true} once it has stopped, by stop/1 or by an :available_now trigger running out of data; {:ok, false} when the timeout passes first. A query that failed comes back as {:error, %Latu.Error{}} carrying its own exception, the server's awaitTermination rethrowing it, so waiting is how a failure finds you.

The wait is a loop of bounded server waits, awaitTermination(interval) until it answers true. A streaming command sends nothing while it waits, and a Connect server ends any silent response stream every senderMaxStreamDuration, so one unbounded wait would hit Latu's empty-reattach ceiling on a healthy query. Spark's semantics are kept; only the RPC is cut into slices. docs/decisions.md has the measurement.

Options

• :timeout — how long to wait in total, or :infinity. Defaults to :infinity.
• :interval — how long each server-side wait lasts. Defaults to 10_000. Shorter means more round trips; longer means a single ExecutePlan idles longer, and only matters when it exceeds the server's senderMaxStreamDuration.

@spec await_termination!(
  t(),
  keyword()
) :: boolean()

Like await_termination/2, raising on failure.

@spec decode_progress(String.t()) :: progress()

Decode one progress report from the JSON the server sends, as last_progress/1 does.

Every key becomes a snake-cased atom, except under start_offset, end_offset, latest_offset and observed_metrics, whose values are left exactly as Spark wrote them: an offset is a string for one source and a JSON object for another, and observed metrics are keyed by the caller's own names. Spark's own key set is small and fixed, so the atoms are bounded.

iex> Latu.StreamingQuery.decode_progress(~s({"batchId": 3, "durationMs": {"addBatch": 12},
...>   "sources": [{"startOffset": "21", "numInputRows": 2}],
...>   "observedMetrics": {"myMetric": {"rowCount": 2}}}))
%{
  batch_id: 3,
  duration_ms: %{add_batch: 12},
  sources: [%{start_offset: "21", num_input_rows: 2}],
  observed_metrics: %{"myMetric" => %{"rowCount" => 2}}
}

@spec events(Latu.Session.t()) :: Enumerable.t()

Every streaming event on the session, as a lazy Stream.

session
|> Latu.StreamingQuery.events()
|> Stream.filter(&(&1.type == :progress))
|> Enum.take(3)

PySpark registers a listener object and calls back on a thread it owns; Latu hands you the events and lets you decide where they run, as Latu.Progress does for a batch query. Nothing is spawned: the bus opens when the stream is first enumerated, and closes when the enumeration ends, however it ends.

Each element is a map with a :type — see event/0. A progress event's :progress is what last_progress/1 answers, so decode_progress/1 is the whole decoder for both.

Four things to know, all of them the server's doing:

• One bus per session. The server answers a second add with a log line and nothing on the wire, so a second events/1 on the same session hangs until the reattach guard gives up and then names this as the likely cause. Enumerate one bus per session at a time.
• Closing is a command, not a release. Halting the stream sends remove_listener_bus_listener; if that fails it is logged and the bus stays registered for the life of the session. A Latu.disconnect/2 clears it either way.
• Events are at-least-once. A reattach replays the responses the server still holds, and neither Latu nor PySpark acknowledges an event, so a duplicate is possible. Match on :run_id and :batch_id if that matters.
• A silent bus is the normal state, so this is the one execution in Latu where an idle stream is not eventually an error. Before the first response it still is.

Raises Latu.Error on failure, as Latu.stream/2 does, since an enumeration has no way to return one.

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

Why the query stopped, if it failed: {:ok, nil} while it runs or after a clean stop, or {:ok, %Latu.Error{kind: :query}} carrying Spark's error class and the server's stack trace.

The poll for a query nobody is waiting on. await_termination/2 gets the same failure as its own {:error, _}.

@spec exception!(t()) :: Latu.Error.t() | nil

Like exception/1, raising on failure.

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

Print the query's physical plan, as Latu.explain/2 does for a frame.

Options

• :mode — :simple (the default) or :extended. The streaming explain has two levels where the frame's has five; the spelling is the same so the two read alike.

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

Like explain/2, raising on failure.

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

The query's physical plan as a string, where explain/2 prints it.

Options

explain/2's: :mode.

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

Like explain_string/2, raising on failure.

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

One active query by id, or nil. The answer carries the current run id, which is what makes a handle recovered here usable where a remembered one may not be.

@spec get!(Latu.Session.t(), String.t()) :: t() | nil

Like get/2, raising on failure.

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

Whether the query is still running.

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

Like is_active/1, raising on failure.

@spec last_progress(t()) :: {:ok, progress() | nil} | {:error, Latu.Error.t()}

The most recent progress report, or nil before the first batch.

{:ok, %{batch_id: 3, sources: [%{num_input_rows: 2}]}} =
  Latu.StreamingQuery.last_progress(query)

Spark's own JSON, keys snake-cased into atoms. The server keeps reports for a while after a query terminates, so this answers on a stopped query too.

@spec last_progress!(t()) :: progress() | nil

Like last_progress/1, raising on failure.

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

Block until every row the source has now has been processed.

A bounded-source verb: over a file directory it returns when the directory is drained, and over a rate or Kafka source it never returns, because there is always more. One server call with no timeout, so a long drain runs into the transport's empty-reattach ceiling (Latu.Client) rather than being looped as await_termination/2 is. The deterministic test shape is trigger: :available_now and await_termination/2 instead.

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

Like process_all_available/1, raising on failure.

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

The recent progress reports, oldest first, as last_progress/1 decodes them.

@spec recent_progress!(t()) :: [progress()]

Like recent_progress/1, raising on failure.

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

Forget the terminated queries await_any_termination/2 would otherwise answer for.

@spec reset_terminated!(Latu.Session.t()) :: :ok

Like reset_terminated/1, raising on failure.

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

The query's status: its message, whether data is waiting, whether a trigger is running, and whether it is active at all.

{:ok, %{message: "Waiting for data to arrive", is_active: true}} =
  Latu.StreamingQuery.status(query)

@spec status!(t()) :: status()

Like status/1, raising on failure.

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

Stop the query. Idempotent on the server: stopping a query that has already terminated is a no-op, which is what lets Latu.with_stream/3 stop in an after whatever happened.

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

Like stop/1, raising on failure.