Arcadic.TimeSeries (Arcadic v0.7.1)

Copy Markdown View Source

ArcadeDB time-series — tenant-blind client for the /api/v1/ts/<db> wire family (InfluxDB line-protocol write, JSON query/latest, PromQL reads) plus the TIMESERIES DDL, downsampling policies, and continuous aggregates. Requires ArcadeDB >= 26.7.2 (older servers answer the /ts routes with a plain 404 %Arcadic.Error{http_status: 404}).

DDL and continuous-aggregate statements ride Arcadic.command/4 (language: "sql" — SQL-only over Bolt, like Arcadic.Schema). The wire family rides optional transport callbacks implemented by the HTTP transport only; on any other transport those functions return {:error, %Arcadic.Error{reason: :not_supported}}.

Identifiers (type/column/tag/field names) are Arcadic.Identifier-validated; grammar tokens (column types, duration units, precisions, aggregation functions) are positive-allowlisted. The two rejection channels differ: an invalid identifier returns {:error, :invalid_identifier}, while an invalid opt value (column-type token, duration, precision, shard count) raises a value-free ArgumentError. Point values are DATA (line protocol / JSON / URL params), never statement text.

Operational contract — write path (each clause live-probed on 26.7.2)

  • Append-only, non-idempotent. No dedup, no upsert, no server-assigned id: the identical point written twice is TWO rows. A lost response followed by a naive retry duplicates every point in the body (the same non-confirmability class as Arcadic.Bulk.ingest/3).
  • Mixed-body partial swallow. When at least one line's type exists, lines naming an UNKNOWN type are silently dropped (HTTP 204). The loud 400 Unknown timeseries type(s) fires only when every line's type is unknown. write/3 guarantees syntactic validity by construction, but cannot know server type existence (tenant-blind, no schema cache) — a typo'd type: in a mixed batch is silent. Verify with query/3 or Arcadic.Schema.types/1 when it matters.
  • Unknown FIELD zero-fill. A line whose field name is not on the type inserts a zero-filled row (204, no error).
  • int64 bound. An integer field value or timestamp outside signed int64 (±9223372036854775807/8) is a 204 + silent line drop server-side (probed both signs) — write/3 raises value-free client-side instead, including on a DateTime whose converted value overflows. This bound is part of the "syntactic validity by construction" guarantee.
  • Unknown tag KEY fails open on query/3/latest/3 (the filter is ignored server-side).
  • write_lines/3 (raw passthrough) additionally inherits the malformed-line silent-skip.

Summary

Types

A duration for :retention / :compaction_interval / downsampling opts.

Functions

Adds a downsampling policy: ALTER TIMESERIES TYPE type ADD DOWNSAMPLING POLICY AFTER n UNIT GRANULARITY n UNIT. Both :after and :granularity are REQUIRED duration/0s.

Adds a downsampling policy, raising on error.

Creates a continuous aggregate: CREATE CONTINUOUS AGGREGATE name AS select_sql. The SELECT rides through VERBATIM — the same injection rationale as Arcadic.MaterializedView: it is raw trailing SQL (not a quoted literal), and ArcadeDB's live-verified single-statement backstop makes a ;-separated second statement a parse error. The result materializes as a document type readable via SELECT FROM name. Refresh with refresh_aggregate/2.

Creates a continuous aggregate, raising on error.

Creates a TIMESERIES type: CREATE TIMESERIES TYPE name TIMESTAMP timestamp_col ….

Creates a TIMESERIES type, raising on error.

Drops a continuous aggregate (no IF EXISTS — a missing aggregate is a server error).

Drops a continuous aggregate, raising on error.

Drops the type's downsampling policy (ALTER TIMESERIES TYPE type DROP DOWNSAMPLING POLICY).

Drops the type's downsampling policy, raising on error.

Drops a TIMESERIES type (DROP TIMESERIES TYPE name — no IF EXISTS; dropping a missing type is a server error, mirroring Arcadic.Trigger/Arcadic.MaterializedView).

Drops a TIMESERIES type, raising on error.

Newest point (GET /api/v1/ts/<db>/latest). opts: :tag — a SINGLE {key, value} pair (the substrate applies only the first tag filter; a multi-entry map is rejected — probed); the value must be non-empty (colon-bearing values are fine: the server splits the wire key:value on the FIRST colon and matches the remainder exactly — probed), :timeout. Returns {:ok, %{columns, latest}}.

Newest point, returning the result map or raising.

Values of one label (GET …/prom/api/v1/label/<label>/values). The label must match the Prometheus label grammar ([a-zA-Z_][a-zA-Z0-9_]*__name__ is legal).

Values of one label, returning the list or raising.

Label names (GET …/prom/api/v1/labels).

Label names, returning the list or raising.

PromQL instant query (GET …/prom/api/v1/query). The PromQL text rides as a URL-encoded query value (data, not statement text). opts: :time (epoch-seconds integer or DateTime), :timeout. Returns the decoded Prometheus data object (%{"resultType" => …, "result" => …}). The metric name is the TIMESERIES type name; tags are labels.

PromQL instant query, returning the data or raising.

PromQL range query (GET …/prom/api/v1/query_range). from/to are epoch-seconds integers or DateTimes; step is an integer (seconds) or a Prometheus duration string ("1m").

PromQL range query, returning the data or raising.

Series matching the given selectors (GET …/prom/api/v1/series, repeated match[] params). matches is a list of non-empty selector strings (the LIST may be empty — the server answers an unfiltered call — but an empty-string entry is a meaningless selector and rejects).

Series matching the selectors, returning the list or raising.

Time-series query (POST /api/v1/ts/<db>/query).

Time-series query, returning the result map or raising.

Refreshes a continuous aggregate (REFRESH CONTINUOUS AGGREGATE name).

Refreshes a continuous aggregate, raising on error.

Writes structured points as InfluxDB line protocol (POST /api/v1/ts/<db>/write).

Writes structured points, raising on error.

Writes raw, already-built line protocol (a relay path — e.g. forwarding a Telegraf batch). lines must be a binary or iodata. arcadic performs NO validation or escaping on the content: this path additionally inherits the server's malformed-line silent-skip (see the moduledoc Operational contract). opts: :precision, :timeout.

Writes raw line protocol, raising on error.

Types

duration()

@type duration() :: {pos_integer(), :seconds | :minutes | :hours | :days}

A duration for :retention / :compaction_interval / downsampling opts.

Functions

add_downsampling(conn, type, opts)

@spec add_downsampling(Arcadic.Conn.t(), String.t(), keyword()) ::
  :ok | {:error, atom() | Exception.t()}

Adds a downsampling policy: ALTER TIMESERIES TYPE type ADD DOWNSAMPLING POLICY AFTER n UNIT GRANULARITY n UNIT. Both :after and :granularity are REQUIRED duration/0s.

add_downsampling!(conn, type, opts)

@spec add_downsampling!(Arcadic.Conn.t(), String.t(), keyword()) :: :ok

Adds a downsampling policy, raising on error.

create_aggregate(conn, name, select_sql)

@spec create_aggregate(Arcadic.Conn.t(), String.t(), String.t()) ::
  :ok | {:error, atom() | Exception.t()}

Creates a continuous aggregate: CREATE CONTINUOUS AGGREGATE name AS select_sql. The SELECT rides through VERBATIM — the same injection rationale as Arcadic.MaterializedView: it is raw trailing SQL (not a quoted literal), and ArcadeDB's live-verified single-statement backstop makes a ;-separated second statement a parse error. The result materializes as a document type readable via SELECT FROM name. Refresh with refresh_aggregate/2.

create_aggregate!(conn, name, select_sql)

@spec create_aggregate!(Arcadic.Conn.t(), String.t(), String.t()) :: :ok

Creates a continuous aggregate, raising on error.

create_type(conn, name, timestamp_col, opts \\ [])

@spec create_type(Arcadic.Conn.t(), String.t(), String.t(), keyword()) ::
  :ok | {:error, atom() | Exception.t()}

Creates a TIMESERIES type: CREATE TIMESERIES TYPE name TIMESTAMP timestamp_col ….

Options

  • :fields — REQUIRED non-empty [{name, type}] (keyword or {binary, binary} pairs).
  • :tags[{name, type}], default none (a tagless type is valid — probed).
  • :precision:second | :millisecond | :microsecond | :nanosecond (omit → server default NANOSECOND).

  • :shards — pos_integer.
  • :retention / :compaction_intervalduration/0.

Clause order is the grammar's FIXED order (SHARDS → RETENTION → COMPACTION_INTERVAL); units emit as plural tokens. There is no IF NOT EXISTS (absent from the 26.7.2 grammar) — creating an existing type is a server error. Value-free on any invalid name/token/option.

create_type!(conn, name, timestamp_col, opts \\ [])

@spec create_type!(Arcadic.Conn.t(), String.t(), String.t(), keyword()) :: :ok

Creates a TIMESERIES type, raising on error.

drop_aggregate(conn, name)

@spec drop_aggregate(Arcadic.Conn.t(), String.t()) ::
  :ok | {:error, atom() | Exception.t()}

Drops a continuous aggregate (no IF EXISTS — a missing aggregate is a server error).

drop_aggregate!(conn, name)

@spec drop_aggregate!(Arcadic.Conn.t(), String.t()) :: :ok

Drops a continuous aggregate, raising on error.

drop_type(conn, name)

@spec drop_type(Arcadic.Conn.t(), String.t()) ::
  :ok | {:error, atom() | Exception.t()}

Drops a TIMESERIES type (DROP TIMESERIES TYPE name — no IF EXISTS; dropping a missing type is a server error, mirroring Arcadic.Trigger/Arcadic.MaterializedView).

drop_type!(conn, name)

@spec drop_type!(Arcadic.Conn.t(), String.t()) :: :ok

Drops a TIMESERIES type, raising on error.

latest(conn, type, opts \\ [])

@spec latest(Arcadic.Conn.t(), String.t(), keyword()) ::
  {:ok, map()} | {:error, atom() | Exception.t()}

Newest point (GET /api/v1/ts/<db>/latest). opts: :tag — a SINGLE {key, value} pair (the substrate applies only the first tag filter; a multi-entry map is rejected — probed); the value must be non-empty (colon-bearing values are fine: the server splits the wire key:value on the FIRST colon and matches the remainder exactly — probed), :timeout. Returns {:ok, %{columns, latest}}.

latest!(conn, type, opts \\ [])

@spec latest!(Arcadic.Conn.t(), String.t(), keyword()) :: map()

Newest point, returning the result map or raising.

prom_label_values(conn, label, opts \\ [])

@spec prom_label_values(Arcadic.Conn.t(), String.t(), keyword()) ::
  {:ok, list()} | {:error, atom() | Exception.t()}

Values of one label (GET …/prom/api/v1/label/<label>/values). The label must match the Prometheus label grammar ([a-zA-Z_][a-zA-Z0-9_]*__name__ is legal).

prom_label_values!(conn, label, opts \\ [])

@spec prom_label_values!(Arcadic.Conn.t(), String.t(), keyword()) :: list()

Values of one label, returning the list or raising.

prom_labels(conn, opts \\ [])

@spec prom_labels(
  Arcadic.Conn.t(),
  keyword()
) :: {:ok, list()} | {:error, atom() | Exception.t()}

Label names (GET …/prom/api/v1/labels).

prom_labels!(conn, opts \\ [])

@spec prom_labels!(
  Arcadic.Conn.t(),
  keyword()
) :: list()

Label names, returning the list or raising.

prom_query(conn, promql, opts \\ [])

@spec prom_query(Arcadic.Conn.t(), String.t(), keyword()) ::
  {:ok, map() | list()} | {:error, atom() | Exception.t()}

PromQL instant query (GET …/prom/api/v1/query). The PromQL text rides as a URL-encoded query value (data, not statement text). opts: :time (epoch-seconds integer or DateTime), :timeout. Returns the decoded Prometheus data object (%{"resultType" => …, "result" => …}). The metric name is the TIMESERIES type name; tags are labels.

prom_query!(conn, promql, opts \\ [])

@spec prom_query!(Arcadic.Conn.t(), String.t(), keyword()) :: map() | list()

PromQL instant query, returning the data or raising.

prom_query_range(conn, promql, from, to, step, opts \\ [])

@spec prom_query_range(
  Arcadic.Conn.t(),
  String.t(),
  integer() | DateTime.t(),
  integer() | DateTime.t(),
  integer() | String.t(),
  keyword()
) :: {:ok, map() | list()} | {:error, atom() | Exception.t()}

PromQL range query (GET …/prom/api/v1/query_range). from/to are epoch-seconds integers or DateTimes; step is an integer (seconds) or a Prometheus duration string ("1m").

prom_query_range!(conn, promql, from, to, step, opts \\ [])

@spec prom_query_range!(
  Arcadic.Conn.t(),
  String.t(),
  integer() | DateTime.t(),
  integer() | DateTime.t(),
  integer() | String.t(),
  keyword()
) :: map() | list()

PromQL range query, returning the data or raising.

prom_series(conn, matches, opts \\ [])

@spec prom_series(Arcadic.Conn.t(), [String.t()], keyword()) ::
  {:ok, list()} | {:error, atom() | Exception.t()}

Series matching the given selectors (GET …/prom/api/v1/series, repeated match[] params). matches is a list of non-empty selector strings (the LIST may be empty — the server answers an unfiltered call — but an empty-string entry is a meaningless selector and rejects).

prom_series!(conn, matches, opts \\ [])

@spec prom_series!(Arcadic.Conn.t(), [String.t()], keyword()) :: list()

Series matching the selectors, returning the list or raising.

query(conn, type, opts \\ [])

@spec query(Arcadic.Conn.t(), String.t(), keyword()) ::
  {:ok, map()} | {:error, atom() | Exception.t()}

Time-series query (POST /api/v1/ts/<db>/query).

opts: :from/:to (integer epoch-milliseconds or DateTime — note: milliseconds regardless of the type's declared PRECISION, probed), :fields (non-empty projection — NOTE the server has a live projection defect returning misaligned values under a wrong-width header; see the usage-rules time-series section), :tags (map — the server ANDs all entries; an UNKNOWN tag key fails open, ignored server-side), :limit (positive integer; server default 20000), :aggregation ([%{field: name, type: :sum | :avg | :min | :max | :count, alias: String?}]) with REQUIRED :bucket_interval (duration/0 or a positive ms integer), :timeout.

Returns the server's columnar shape with atomized keys: raw {:ok, %{columns, rows, count}}; aggregated {:ok, %{aggregations, buckets, count}} (each bucket %{timestamp, values}). Deliberately NOT zipped into row-maps.

query!(conn, type, opts \\ [])

@spec query!(Arcadic.Conn.t(), String.t(), keyword()) :: map()

Time-series query, returning the result map or raising.

refresh_aggregate(conn, name)

@spec refresh_aggregate(Arcadic.Conn.t(), String.t()) ::
  :ok | {:error, atom() | Exception.t()}

Refreshes a continuous aggregate (REFRESH CONTINUOUS AGGREGATE name).

refresh_aggregate!(conn, name)

@spec refresh_aggregate!(Arcadic.Conn.t(), String.t()) :: :ok

Refreshes a continuous aggregate, raising on error.

write(conn, points, opts \\ [])

@spec write(Arcadic.Conn.t(), [map()], keyword()) ::
  :ok | {:error, atom() | Exception.t()}

Writes structured points as InfluxDB line protocol (POST /api/v1/ts/<db>/write).

Each point: %{type: String, fields: map (REQUIRED, non-empty), tags: map \\ %{}, timestamp: integer | DateTime | nil}. Field values: integer (emits 42i), float, boolean, or String; tag values: String. Type, tag and field NAMES are Arcadic.Identifier-validated (they are schema columns); atom names are allowed and converted. An integer timestamp passes through in the :precision unit; a DateTime is converted; nil omits it (server-assigned now). Tags and fields emit in lexicographic key order (deterministic, Influx canonical form).

opts: :precision (:ns default | :us | :ms | :s), :timeout (ms).

Read the moduledoc Operational contract before relying on retries or mixed-type batches. Value-free on every invalid input (Rule 3): shape violations raise with static messages; a bad name returns {:error, :invalid_identifier}; caller values never appear in an error.

write!(conn, points, opts \\ [])

@spec write!(Arcadic.Conn.t(), [map()], keyword()) :: :ok

Writes structured points, raising on error.

write_lines(conn, lines, opts \\ [])

@spec write_lines(Arcadic.Conn.t(), iodata(), keyword()) ::
  :ok | {:error, atom() | Exception.t()}

Writes raw, already-built line protocol (a relay path — e.g. forwarding a Telegraf batch). lines must be a binary or iodata. arcadic performs NO validation or escaping on the content: this path additionally inherits the server's malformed-line silent-skip (see the moduledoc Operational contract). opts: :precision, :timeout.

write_lines!(conn, lines, opts \\ [])

@spec write_lines!(Arcadic.Conn.t(), iodata(), keyword()) :: :ok

Writes raw line protocol, raising on error.