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

A running structured streaming query: a server-side object, addressed by id.

`Latu.write_stream/2` hands one back. The verbs here are one `StreamingQueryCommand` each,
keyed by the id and run id the struct carries, plus the four `spark.streams` verbs, which
take a session as `Latu.Catalog`'s do. Inert data, like everything Latu hands out: no
process watches the query, and nothing stops it for you.

    {:ok, query} =
      Latu.write_stream(rollups,
        format: "parquet", path: "/data/rollups", trigger: :available_now,
        checkpoint_location: "/data/ckpt")

    {:ok, true} = Latu.StreamingQuery.await_termination(query)
    {:ok, progress} = Latu.StreamingQuery.last_progress(query)
    :ok = Latu.StreamingQuery.stop(query)

Three things a server measured that the types cannot say:

  * **A query outlives its client.** Stop it, or it runs until the session ends; a lost handle
    comes back through `get/2` or `active/1`. `Latu.with_stream/3` stops in an `after`.
  * **Failure is asynchronous.** A query that dies at 3am raises nothing here. Whoever is in
    `await_termination/2` gets the failure as `{:error, _}`; anyone else asks `exception/1`,
    or watches `events/1` for a `:terminated`.
  * **`Latu.interrupt/2` with no scope stops streaming queries too.** By tag it reaches the
    ones started under that tag, which is the session-wide kill and a hazard for anyone
    interrupting a batch query in a session that also streams.

Progress and status are snake-cased maps of Spark's own JSON, which over Connect is lossy
against the driver's report: no top-level row counts, and a `start_offset` that is a string.
Sum `:num_input_rows` over `:sources` for the total. `docs/deviations.md` has the shape.

Times are milliseconds, as everywhere in Latu.

# `event`

```elixir
@type event() :: %{:type => event_type(), optional(atom()) => term()}
```

One event off the listener bus: `:type` plus Spark's own JSON, snake-cased.

  * `:progress` — carries `:progress`, exactly what `last_progress/1` returns.
  * `:idle` — `:id`, `:run_id`, `:timestamp`.
  * `:terminated` — `:id`, `:run_id`, `:exception`, `:error_class_on_exception`.
  * `:unknown` — a type this client has not met, with `:event_type` and the raw `:json`
    undecoded. A newer server can add one, and dropping the bus over it would be worse than
    handing it on.

# `event_type`

```elixir
@type event_type() :: :progress | :idle | :terminated | :unknown
```

Which of Spark's three listener events this is, or `:unknown` for a fourth a newer server
invented. See `t:event/0` for what each one carries.

# `progress`

```elixir
@type progress() :: %{optional(atom()) => term()}
```

One progress report, Spark's JSON with snake-cased keys. See `last_progress/1`.

# `status`

```elixir
@type status() :: %{
  message: String.t(),
  is_data_available: boolean(),
  is_trigger_active: boolean(),
  is_active: boolean()
}
```

What `status/1` answers.

# `t`

```elixir
@type t() :: %Latu.StreamingQuery{
  id: String.t(),
  name: String.t() | nil,
  run_id: String.t(),
  session: Latu.Session.t()
}
```

The handle. `name` is `nil` when the query was started without one; `run_id` changes when a
query is restarted from its checkpoint, and the server refuses a command carrying the old one.

# `active`

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

# `active!`

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

Like `active/1`, raising on failure.

# `await_any_termination`

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

# `await_any_termination!`

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

Like `await_any_termination/2`, raising on failure.

# `await_termination`

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

# `await_termination!`

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

Like `await_termination/2`, raising on failure.

# `decode_progress`

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

# `events`

```elixir
@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 `t: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.

# `exception`

```elixir
@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, _}`.

# `exception!`

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

Like `exception/1`, raising on failure.

# `explain`

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

# `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()}
```

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

## Options

`explain/2`'s: `:mode`.

# `explain_string!`

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

Like `explain_string/2`, raising on failure.

# `get`

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

# `get!`

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

Like `get/2`, raising on failure.

# `is_active`

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

Whether the query is still running.

# `is_active!`

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

Like `is_active/1`, raising on failure.

# `last_progress`

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

# `last_progress!`

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

Like `last_progress/1`, raising on failure.

# `process_all_available`

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

# `process_all_available!`

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

Like `process_all_available/1`, raising on failure.

# `recent_progress`

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

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

# `recent_progress!`

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

Like `recent_progress/1`, raising on failure.

# `reset_terminated`

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

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

# `reset_terminated!`

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

Like `reset_terminated/1`, raising on failure.

# `status`

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

# `status!`

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

Like `status/1`, raising on failure.

# `stop`

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

# `stop!`

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

Like `stop/1`, raising on failure.

---

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