The ETS store behind InfluxElixir.Client.Local: the one module that
knows the key layout.
One :ordered_set per instance, :public so async: true tests can
write from any process, with write_concurrency so writers do not queue
on one lock. Every mutation is an insert or delete of its own keys (a
payload's points go in as one batched insert), so concurrent writers —
tests sharing a database, BatchWriter flushes racing direct writes —
never read-modify-write a shared value and no write is lost:
{:database, name}=>true{:bucket, name}=>%{retention: seconds}{:token, name}=> the token map, and:token_id=> the last id given (a token is created by one insert of its full map, under a lock that also gives the id, so a delete can never be overwritten){:point, database, measurement, seq}=> the point —seqis a monotonic integer, so points scan in insertion order{:series_time, database, measurement, timestamp, tags}— one per point written; a second write of the same key adds{:duplicates, database, measurement}— reads merge that measurement's duplicate points (same tags and time) only when this marker exists{:column, database, measurement, column}=> the column's kind (iox::column_type::tagoriox::column_type::field::<type>), fixed by the first write that names the column{:lock, resource}=> the process holding that resource's lock, for as long as it runs the one operation that needs it (database limit, token ids)
The store is policy-free: what a write may contain, which errors the
engine returns and how a query reads rows live in Client.Local and the
modules it delegates to.
Summary
Functions
A bucket's metadata, or nil when it is not registered.
Whether a bucket is registered.
The buckets as {name, meta}, sorted by name.
A column's registered kind, or nil — a read that registers nothing.
Every column in a database as {measurement, column, kind}, in that order.
Registers a database after check approves it, atomically: check is
called with the registered databases while no other process creates one,
so a limit such as Core's five cannot be passed by concurrent first
writes (each would have seen four). A database that exists is :ok
without taking the lock. check returns :ok or {:error, reason}.
Creates the token named name: build is called with the next id and
returns the token map, which is stored by one insert_new. {:ok, token},
or :exists when the name is taken, which spends no id (as on the
engine). The operator token _admin (id 0) always exists.
Whether a database is registered.
The registered databases.
Removes a bucket; :error when it was not registered.
Deletes the points of a measurement that match? accepts, judged on the
merged points as the engine sees them; every stored object behind a
matching point is deleted by its own key, so a concurrent write is never
lost. Returns the number of (merged) points deleted.
Removes the token named name: :ok, or :error when there is none.
Deletes the store. Safe to call again: the table dies with its owner, and
an on_exit callback can run after that, so a missing table is :ok.
Drops a database with everything in it — points, schema, the
duplicate index — so a re-created database starts empty. :error when
it was not registered.
The database's measurements, sorted by name: the ones with a table, which a write creates by registering its columns. They are read from the column keys, which are few, not from the points, which are many. The order is the keys', alphabetical, which is how InfluxDB lists them.
Creates a store with databases registered.
The double's "now", in nanoseconds: used to stamp untimed points and for
SQL and Flux now(). System.os_time/1 and System.system_time/1 can
differ by microseconds (time warp), and a test stamps points with either;
taking the later of the two means a point written a moment ago is never
after now(), as it never is on a real server.
A measurement's points in insertion order, duplicates merged. InfluxDB — both versions, verified — treats points with the same tags and time as one point whose fields merge, the later write winning per field.
The points of a database, duplicates merged: every measurement's with
:all, or only the listed measurements' (a query that names them need
not read the rest).
Registers a bucket with its metadata (replacing any earlier one).
Registers a column's kind if it is new. The first writer fixes it
atomically (insert_new), so a concurrent writer never loses a column.
Returns :ok when the kind matches (or was just registered) and
{:conflict, existing} when the column already has another kind.
Stores a payload's points as written, each its own object, in one batched
insert. A point without a timestamp gets the server's time here as a
fallback; Client.Local.write/3 stamps a write's untimed lines itself,
all with one time, as the engines do.
Whether the measurement has any column registered (the table exists).
A measurement's tag columns.
Types
@type point() :: InfluxElixir.Client.Local.LineProtocolParser.point()
A stored point.
@type t() :: :ets.table()
A store instance.
Functions
A bucket's metadata, or nil when it is not registered.
Whether a bucket is registered.
The buckets as {name, meta}, sorted by name.
A column's registered kind, or nil — a read that registers nothing.
Every column in a database as {measurement, column, kind}, in that order.
@spec create_database(t(), binary(), (Enumerable.t(binary()) -> :ok | {:error, term()})) :: :ok | {:error, term()}
Registers a database after check approves it, atomically: check is
called with the registered databases while no other process creates one,
so a limit such as Core's five cannot be passed by concurrent first
writes (each would have seen four). A database that exists is :ok
without taking the lock. check returns :ok or {:error, reason}.
@spec create_token(t(), binary(), (pos_integer() -> map())) :: {:ok, map()} | :exists
Creates the token named name: build is called with the next id and
returns the token map, which is stored by one insert_new. {:ok, token},
or :exists when the name is taken, which spends no id (as on the
engine). The operator token _admin (id 0) always exists.
The id is chosen, and the name checked, under a lock so that no id is
spent on a duplicate; the token goes in whole, never as a placeholder to
be filled in later, so a delete_token/2 can never be overwritten and
bring the token back.
Whether a database is registered.
The registered databases.
Removes a bucket; :error when it was not registered.
Deletes the points of a measurement that match? accepts, judged on the
merged points as the engine sees them; every stored object behind a
matching point is deleted by its own key, so a concurrent write is never
lost. Returns the number of (merged) points deleted.
The series_time key of a doomed point is deleted with it, which a
writer racing the delete may have just claimed. The invariant that makes
this safe: a writer that finds the key taken has already put the
measurement's duplicates marker, which is never removed, so reads merge
it with any point left at that series and time; one that finds the key
gone writes a point that nothing deleted here refers to. Either way the
point the writer stored is kept, as a write after the delete would be, and
no point is merged with one that was deleted.
Removes the token named name: :ok, or :error when there is none.
@spec drop(t()) :: :ok
Deletes the store. Safe to call again: the table dies with its owner, and
an on_exit callback can run after that, so a missing table is :ok.
Drops a database with everything in it — points, schema, the
duplicate index — so a re-created database starts empty. :error when
it was not registered.
The database's measurements, sorted by name: the ones with a table, which a write creates by registering its columns. They are read from the column keys, which are few, not from the points, which are many. The order is the keys', alphabetical, which is how InfluxDB lists them.
@spec new(Enumerable.t()) :: t()
Creates a store with databases registered.
@spec now_ns() :: integer()
The double's "now", in nanoseconds: used to stamp untimed points and for
SQL and Flux now(). System.os_time/1 and System.system_time/1 can
differ by microseconds (time warp), and a test stamps points with either;
taking the later of the two means a point written a moment ago is never
after now(), as it never is on a real server.
A measurement's points in insertion order, duplicates merged. InfluxDB — both versions, verified — treats points with the same tags and time as one point whose fields merge, the later write winning per field.
The points of a database, duplicates merged: every measurement's with
:all, or only the listed measurements' (a query that names them need
not read the rest).
Registers a bucket with its metadata (replacing any earlier one).
Registers a column's kind if it is new. The first writer fixes it
atomically (insert_new), so a concurrent writer never loses a column.
Returns :ok when the kind matches (or was just registered) and
{:conflict, existing} when the column already has another kind.
Stores a payload's points as written, each its own object, in one batched
insert. A point without a timestamp gets the server's time here as a
fallback; Client.Local.write/3 stamps a write's untimed lines itself,
all with one time, as the engines do.
Concurrent writers never read-modify-write a shared value, so no write is
lost. A series and time already written, by this payload or another
writer, marks the measurement duplicates so reads merge it. The payload's
own repeats are found with a map and the keys it adds are claimed by one
insert_new, which is all-or-nothing: only when another writer holds one
of them are the keys claimed singly to find which.
Whether the measurement has any column registered (the table exists).
A measurement's tag columns.