TypeDB.GRPC.Transaction (TypeDB.GRPC v0.1.0)

Copy Markdown View Source

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)
end

A 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, 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

t()

@type t() :: %TypeDB.GRPC.Transaction{
  conn: TypeDB.GRPC.Connection.t(),
  database: String.t(),
  pid: pid(),
  type: type()
}

type()

@type type() :: :read | :write | :schema

Functions

analyze(tx, query, opts \\ [])

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

analyze!(tx, query, opts \\ [])

@spec analyze!(term(), term(), term()) :: map()

Analyses a query, raising on failure.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

close(transaction, opts \\ [])

@spec close(
  t(),
  keyword()
) :: :ok

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.

commit(tx, opts \\ [])

@spec commit(
  t(),
  keyword()
) :: :ok | {:error, TypeDB.Error.t()}

Commits. The transaction is finished either way.

commit!(tx, opts \\ [])

@spec commit!(term(), term()) :: :ok

Commits, raising on failure.

execute_many(tx, queries, opts \\ [])

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

execute_many!(tx, queries, opts \\ [])

@spec execute_many!(term(), term(), term()) :: :ok

Sends queries discarding answers, raising on failure.

open(conn, database, type, opts \\ [])

@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 :schema transaction waits for the exclusive lock
  • :timeout — how long to wait for the open itself

open!(conn, database, type, opts \\ [])

@spec open!(term(), term(), term(), term()) :: t()

Opens a transaction, raising on failure.

open?(transaction)

@spec open?(t()) :: boolean()

Whether the transaction is still usable.

query(tx, query, opts \\ [])

@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's given stage, 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 as query_structure on %TypeDB.Answer.ConceptRows{}, and involved_blocks on 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 way analyze/3 renders 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 at prefetch_size: 20_000 — the server produces the whole batch before sending any of it, so its work stops overlapping with the driver's decoding

query!(tx, query, opts \\ [])

@spec query!(term(), term(), term()) :: TypeDB.Answer.t()

Runs a query, raising on failure.

query_many(tx, queries, opts \\ [])

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

query_many!(tx, queries, opts \\ [])

@spec query_many!(term(), term(), term()) :: [TypeDB.Answer.t()]

Runs several queries, raising on failure.

rollback(tx, opts \\ [])

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

rollback!(tx, opts \\ [])

@spec rollback!(term(), term()) :: :ok

Discards writes, raising on failure.

stream(tx, query, opts \\ [])

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

transaction(conn, database, type, fun, opts \\ [])

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