A transaction, which on this transport is a bidirectional stream.
rpc transaction (stream Transaction.Client) returns (stream Transaction.Server)
— one stream per transaction, and closing the stream closes the transaction.
That is a different object from TypeDB.Transaction, which is an id in a
struct and nothing else, and the difference has consequences worth stating
before the API rather than after it.
It is a process, and the handle is still a struct
A stream has to be owned by something, so a transaction here is a GenServer.
The handle you hold is still a plain struct — it carries the pid — so it can
still be passed between processes freely, which is the property the sibling
driver documents and which would otherwise have been lost. Requests from
different processes are tracked independently, so concurrent callers on one
handle each get their own answer. What changes is lifetime: the transaction
dies with its process, so a transaction is no longer something that outlives
the VM that opened it.
A call that times out abandons its own request and leaves the transaction
open, because with several callers the alternative is one slow read taking
everybody else's work down with it. Ending the transaction is the caller's
decision, and transaction/5 makes it.
Watching a transaction end
TypeDB's other drivers offer an on_close callback. There is no such function
here, and that is the API rather than a gap: a transaction is a process, so
the language already has a better answer.
{:ok, tx} = Transaction.open(conn, "social", :write)
ref = Process.monitor(tx.pid)
receive do
{:DOWN, ^ref, :process, _pid, reason} -> handle_the_end(reason)
endA monitor does everything a callback does and several things it cannot. It
delivers to your mailbox rather than running your code inside the
transaction's process, so a slow or crashing handler cannot take the
transaction down with it. It works from any process holding the handle, and
more than one at a time. It can be cancelled with Process.demonitor/2. And
it fires whatever ended the transaction — a commit, a close/2, a stream that
died, the VM shutting down — where a callback list has to be maintained.
reason is :normal for a transaction that finished on purpose, including
after commit/2 and close/2.
Reads pipeline. Writes do not.
Transaction.Client carries a repeated field and requests are correlated by
req_id rather than by order, so several can be in flight at once — and for
reads that is a large win: 200 reads sent together answer in 47 ms against
the same 200 waiting for each other.
Writes are a different matter, and the reason query_many/3 documents itself
the way it does. TypeDB aborts a write query's answer stream when the next
write in the same transaction starts executing, with
[TSV13] Execution interrupted by to a concurrent write query. Measured
against 3.12.1: of 500 pipelined inserts, 4 answered and 496 came back TSV13.
What makes that dangerous rather than merely disappointing is what happens
next. The writes themselves do land — committing anyway put all 500 rows in
the database — so a caller that ignored the errors would commit work it was
told had failed. query_many/3 refuses instead: it returns the first failure
and the bracket closes without committing, which is why the same 500 land
nowhere when the driver is driving.
So: pipeline reads. Send writes one at a time with query/3, which is what
the sequential number in the README measures.
Answers have no ceiling
The HTTP API returns one answer, capped at 10 000 by default, with a warning
when it truncated. There is no cap here: answers arrive in parts and this
module drives the flow control that asks for the next one, so an answer is as
large as the query makes it. TypeDB.Answer.truncated?/1 on an answer from
this driver is always false, and honestly so.
query/3 collects the parts before returning, so a very large answer is a very
large list. stream/3 does not: it hands the parts to the caller as they
arrive and asks for the next only when the consumer wants it, so memory
follows the batch rather than the answer. That was true from the streaming
step onwards, and this paragraph went on denying it — Audit VI, VI-1.
Summary
Functions
Analyses a query without running it.
Analyses a query, raising on failure.
Returns a specification to start this module under a supervisor.
Closes the transaction, discarding uncommitted writes.
Commits. The transaction is finished either way.
Commits, raising on failure.
Sends every query pipelined and discards the answers.
Sends queries discarding answers, raising on failure.
Opens a transaction.
Opens a transaction, raising on failure.
Whether the transaction is still usable.
Runs query and waits for its whole answer.
Runs a query, raising on failure.
Sends every query before waiting for any answer.
Runs several queries, raising on failure.
Discards everything written so far, leaving the transaction open.
Discards writes, raising on failure.
Reads a query as a Stream, pulling from the server as it is consumed.
Runs fun inside a transaction, committing on success.
Types
@type t() :: %TypeDB.GRPC.Transaction{ conn: TypeDB.GRPC.Connection.t(), database: String.t(), pid: pid(), type: type() }
@type type() :: :read | :write | :schema
Functions
@spec analyze(t(), String.t(), keyword()) :: {:ok, map()} | {:error, TypeDB.Error.t()}
Analyses a query without running it.
The sibling's TypeDB.Transaction.analyze/3, on this transport. TypeDB parses
and type-checks the query and describes what it would do — the conjunctions,
the constraints, the variable annotations — which is what makes it useful for
validating a generated query before it touches data.
Options
:include_plan— ask for the execution plan as well as the structure:timeout— how long to wait
The two transports do not return the same tree
Both return {:ok, map} and both describe the same analysis, but the server
renders it twice: the HTTP API emits hand-written JSON, and gRPC emits a
protobuf message this driver walks into maps. The conventions were matched
where matching them costs nothing — json_name keys, so "textSpan" rather
than "text_span", and a "tag" on every oneof — and they still differ in
detail, starting with a variable id that is a string over there and an integer
here.
So this is a debugging and validation tool, not a portability point. Code that reads the analysis is code that knows which transport it is on.
Analyses a query, raising on failure.
Returns a specification to start this module under a supervisor.
See Supervisor.
Closes the transaction, discarding uncommitted writes.
Idempotent, and never fails on a transaction that is already finished — so it
is safe in an after block, which is where it belongs.
Takes :timeout, defaulting to 5 s: closing waits for the server to finish
whatever was in flight, and a caller under a deadline of its own may want to
bound that.
@spec commit( t(), keyword() ) :: :ok | {:error, TypeDB.Error.t()}
Commits. The transaction is finished either way.
Commits, raising on failure.
@spec execute_many(t(), [String.t() | {String.t(), keyword()}], keyword()) :: :ok | {:error, TypeDB.Error.t()}
Sends every query pipelined and discards the answers.
This is how writes are sent fast on this transport, and it exists because of
a behaviour that took measuring to pin down. TypeDB aborts a write's answer
stream when the next write in the same transaction starts, reporting TSV13
— but the write itself runs, in order, and lands. So the answers are the only
thing lost, and a caller that does not want them can have the pipeline.
Measured against 3.12.1: a thousand inserts in one transaction take about 150 ms this way against 622 ms one at a time.
What it does not discard is failure. TSV13 is ignored, because on this
path it means "the answer was dropped" rather than "the write failed". Every
other per-query error is returned — which matters most for a query that does
not parse, since TypeDB reports TQL0, keeps the transaction usable, and
simply does not run that one. A mode that swallowed every error would let a
malformed query vanish and commit the rest. Measured: 100 good inserts plus
one unparseable query commits the 100 and reports TQL0.
Errors that are about the data rather than the text — an unknown type, a
value of the wrong type, a @key violation — abort the stream, so the
transaction cannot be committed at all and nothing lands. Those cannot be
missed whatever this function does.
Transaction.transaction(conn, "social", :write, fn tx ->
Transaction.execute_many(tx, Enum.map(people, &insert_query/1))
end)
Sends queries discarding answers, raising on failure.
@spec open(TypeDB.GRPC.Connection.t(), String.t(), type(), keyword()) :: {:ok, t()} | {:error, TypeDB.Error.t()}
Opens a transaction.
Options
:transaction_timeout_millis— how long the server keeps it without traffic:schema_lock_acquire_timeout_millis— how long a:schematransaction waits for the exclusive lock:timeout— how long to wait for the open itself
Opens a transaction, raising on failure.
Whether the transaction is still usable.
@spec query(t(), String.t(), keyword()) :: {:ok, TypeDB.Answer.t()} | {:error, TypeDB.Error.t()}
Runs query and waits for its whole answer.
Options
:given_rows— rows for TypeQL'sgivenstage, in the same shape the sibling driver takes:timeout— how long to wait for the answer:include_instance_types— ask the server to attach the type of every instance in the answer:include_query_structure— ask for the analysed pipeline alongside the rows. It arrives asquery_structureon%TypeDB.Answer.ConceptRows{}, andinvolved_blockson each row then says which of its conjunctions produced that row. The same option, the same two fields and the same purpose as the sibling's; the tree is rendered the wayanalyze/3renders one, so the caveat there applies here — the transports describe the same analysis and do not spell it identically:prefetch_size— how many answers the server sends before waiting to be asked for more. Worth knowing before reaching for it: raising it makes a streamed read slower, not faster. Measured on 20 000 rows, 432 ms at the default against 985 ms atprefetch_size: 20_000— the server produces the whole batch before sending any of it, so its work stops overlapping with the driver's decoding
@spec query!(term(), term(), term()) :: TypeDB.Answer.t()
Runs a query, raising on failure.
@spec query_many(t(), [String.t() | {String.t(), keyword()}], keyword()) :: {:ok, [TypeDB.Answer.t()]} | {:error, TypeDB.Error.t()}
Sends every query before waiting for any answer.
For reads. queries is a list of query strings or {query, opts}
pairs, and answers come back in the same order.
Pipelining writes through this does not work, and the failure is the loud
kind rather than the quiet kind: TypeDB aborts each write's answer stream when
the next one starts, with TSV13, and this returns that failure. The bracket
then closes without committing — which is the outcome you want, because the
writes had begun to land and committing them would commit work the server
reported as failed. Send writes with query/3.
@spec query_many!(term(), term(), term()) :: [TypeDB.Answer.t()]
Runs several queries, raising on failure.
@spec rollback( t(), keyword() ) :: :ok | {:error, TypeDB.Error.t()}
Discards everything written so far, leaving the transaction open.
The same semantics as the sibling's rollback/2, and the same warning: this
does not finish a transaction. close/2 does.
Discards writes, raising on failure.
@spec stream(t(), String.t(), keyword()) :: Enumerable.t()
Reads a query as a Stream, pulling from the server as it is consumed.
The reason this transport is worth having for large reads. A batch is held at a time and the next one is asked for only when the consumer wants it, so memory follows the batch rather than the answer — and stopping early stops the server:
Transaction.stream(tx, "match $p isa person, has name $n; select $n;")
|> Stream.map(&TypeDB.ConceptRow.typed_value(&1, "n"))
|> Enum.take(10)That takes one batch out of however many the query would have produced, because the continuation signal TypeDB waits on is sent by the consumer's demand and never sent again.
The stream is tied to this transaction and does not close it — use
TypeDB.GRPC.stream/4 when the transaction exists only for the read.
@spec transaction( TypeDB.GRPC.Connection.t(), String.t(), type(), (t() -> result), keyword() ) :: result | {:error, TypeDB.Error.t()} when result: term()
Runs fun inside a transaction, committing on success.
Commits when fun returns anything but {:error, _}; closes without
committing on an error, a raise, a throw or an exit. A :read transaction is
closed rather than committed.