Where the Spark Connect server is, and the identity it keys state on.
A plain struct, not a process. Latu.connect/2 builds one; every function that talks to the
server takes it and lives on Latu (Latu.conf/2, Latu.interrupt/2, Latu.sql/3). What
lives here is pure: building a session from a URL, and its tags. That is the whole rule.
Latu starts and supervises nothing — which makes the session the only place Latu has to keep a default, so the knobs live here and travel with it:
:timeout,:connect_timeout— per-RPC and establishment deadlines.:window_size,:keepalive,:keepalive_tolerance— HTTP/2 flow control and liveness.:retry— aLatu.Retry, the policy every execution retries under.
Every one is a Latu.connect/2 option, so none of them needs a struct poke.
Timeouts are in milliseconds, or :infinity.
Summary
Functions
Add an operation tag, for Latu.interrupt/2 to match on.
Check that a response belongs to this session, and latch the server's id.
Parse a Spark Connect URL.
Like from_url/2, raising on a malformed URL.
Drop a tag. A tag that is not set is not an error.
Types
@type t() :: %Latu.Session{ channel: struct() | nil, client_type: String.t(), connect_timeout: timeout(), headers: [{String.t(), String.t()}], host: String.t(), keepalive: timeout(), keepalive_tolerance: non_neg_integer(), port: :inet.port_number(), retry: Latu.Retry.t(), server_session_id: String.t() | nil, session_id: String.t(), tags: [String.t()], timeout: timeout(), token: String.t() | nil, use_ssl: boolean(), user_id: String.t(), user_name: String.t(), window_size: non_neg_integer() }
Functions
Add an operation tag, for Latu.interrupt/2 to match on.
Every execution this session runs carries its tags, so Latu.interrupt(session, tag: "etl")
cancels them from anywhere — which is the point, since the process running the query is
blocked in it.
session = Latu.Session.add_tag(session, "exploration")A tag is a string or an atom, cannot be empty, and cannot contain a comma, because Spark
joins tags with one. Adding a tag twice does nothing. The tags are session.tags, and
%{session | tags: []} clears them — a struct, so no getters.
@spec confirm(t(), String.t() | nil, String.t() | nil) :: {:ok, t()} | {:error, Latu.Error.t()}
Check that a response belongs to this session, and latch the server's id.
Every response carries both ids. A mismatched session_id means we are reading someone
else's answer; a changed server_session_id on a pinned session means the server restarted
and our session no longer exists on it. Both are worth catching here rather than as
inexplicable behaviour later.
@spec from_url( String.t(), keyword() ) :: {:ok, t()} | {:error, Latu.Error.t()}
Parse a Spark Connect URL.
sc://host[:port][/[;key=value...]]Params are semicolon-separated path parameters, not a query string. Recognised: use_ssl,
token, user_id, user_agent, session_id. Anything else becomes a gRPC metadata
header, which is how Databricks-style URLs pass x-databricks-cluster-id and friends.
Values are percent-decoded, so a = in a token must be written %3D.
Options override the URL: [:session_id, :user_id, :user_name, :client_type, :timeout, :connect_timeout, :tags, :window_size, :keepalive, :keepalive_tolerance, :retry].
iex> {:ok, s} = Latu.Session.from_url("sc://localhost:15002")
iex> {s.host, s.port, s.use_ssl}
{"localhost", 15002, false}Like from_url/2, raising on a malformed URL.
Drop a tag. A tag that is not set is not an error.