telemetrydatabase
/
schema
/
telemetry_schema.sql
1 contributor
BEGIN;
-- Extensions.
CREATE SCHEMA telemetry;
CREATE EXTENSION btree_gist;
-- Enums.
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states such as thermostat_mode, pump_state, hvac_mode, or
-- door_state. Values will likely be stored as smallint and use the
-- same segment engine as boolean metrics, but with exact comparison
-- over enumerated values.
CREATE TYPE telemetry.metric_type_enum AS ENUM ('numeric', 'boolean', 'state');
CREATE TYPE telemetry.comparison_mode_enum AS ENUM ('epsilon', 'exact');
-- Registry utility functions.
CREATE OR REPLACE FUNCTION telemetry.normalize_metric_table_name(p_table_name text)
RETURNS text
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
v_parts text[];
BEGIN
IF p_table_name IS NULL OR btrim(p_table_name) = '' THEN
RAISE EXCEPTION 'table_name is required';
END IF;
v_parts := pg_catalog.parse_ident(p_table_name, false);
CASE COALESCE(array_length(v_parts, 1), 0)
WHEN 1 THEN
RETURN v_parts[1];
WHEN 2 THEN
IF v_parts[1] <> 'telemetry' THEN
RAISE EXCEPTION 'metric tables must live in schema telemetry: %', p_table_name;
END IF;
RETURN v_parts[2];
ELSE
RAISE EXCEPTION 'invalid metric table reference: %', p_table_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.metric_table_sql(p_table_name text)
RETURNS text
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_table_name text := telemetry.normalize_metric_table_name(p_table_name);
v_table_sql text := format('telemetry.%I', v_table_name);
BEGIN
IF to_regclass(v_table_sql) IS NULL THEN
RAISE EXCEPTION 'could not resolve metric table %', v_table_sql;
END IF;
RETURN v_table_sql;
END;
$$;
-- Core registry tables.
CREATE TABLE telemetry.metrics (
metric_pk bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
metric_name text NOT NULL UNIQUE,
table_name text NOT NULL UNIQUE,
domain_name text NOT NULL DEFAULT 'generic',
metric_type telemetry.metric_type_enum NOT NULL,
comparison_mode telemetry.comparison_mode_enum NOT NULL,
epsilon double precision,
min_value double precision,
max_value double precision,
rounding_precision double precision,
allow_null boolean NOT NULL DEFAULT true,
max_sampling_interval interval NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
CHECK (min_value IS NULL OR max_value IS NULL OR min_value <= max_value),
CONSTRAINT metrics_policy_shape_check
CHECK (
(metric_type = 'numeric'::telemetry.metric_type_enum
AND comparison_mode = 'epsilon'::telemetry.comparison_mode_enum
AND epsilon IS NOT NULL
AND rounding_precision IS NOT NULL)
OR
(metric_type IN (
'boolean'::telemetry.metric_type_enum,
'state'::telemetry.metric_type_enum
)
AND comparison_mode = 'exact'::telemetry.comparison_mode_enum
AND epsilon IS NULL
AND min_value IS NULL
AND max_value IS NULL
AND rounding_precision IS NULL)
),
CONSTRAINT metrics_table_name_check
CHECK (table_name = telemetry.normalize_metric_table_name(table_name))
);
-- Policy tables.
CREATE TABLE telemetry.metric_policies (
policy_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
metric_name text NOT NULL
REFERENCES telemetry.metrics(metric_name)
ON UPDATE CASCADE
ON DELETE CASCADE,
valid_from timestamptz NOT NULL,
comparison_mode telemetry.comparison_mode_enum NOT NULL,
epsilon double precision,
min_value double precision,
max_value double precision,
rounding_precision double precision,
allow_null boolean NOT NULL,
max_sampling_interval interval NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
UNIQUE (metric_name, valid_from)
);
CREATE TABLE telemetry.metric_retention_policies (
metric_name text PRIMARY KEY
REFERENCES telemetry.metrics(metric_name)
ON UPDATE CASCADE
ON DELETE CASCADE,
raw_retention interval NULL,
compact_after interval NULL
);
-- Device registry tables.
CREATE TABLE telemetry.devices (
device_pk bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
device_id text NOT NULL UNIQUE,
device_type text NULL,
location text NULL,
metadata jsonb NULL,
last_seen timestamptz NULL,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE telemetry.metric_device_watermarks (
metric_name text NOT NULL
REFERENCES telemetry.metrics(metric_name)
ON UPDATE CASCADE
ON DELETE CASCADE,
device_id text NOT NULL
REFERENCES telemetry.devices(device_id)
ON UPDATE CASCADE
ON DELETE CASCADE,
last_observed_at timestamptz NOT NULL,
PRIMARY KEY (metric_name, device_id)
);
-- Runtime helpers.
CREATE OR REPLACE FUNCTION telemetry.round_to_increment(
p_value double precision,
p_increment double precision
)
RETURNS double precision
LANGUAGE sql
IMMUTABLE
STRICT
AS $$
SELECT CASE
WHEN p_increment <= 0 THEN p_value
ELSE round(p_value / p_increment) * p_increment
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.ensure_device(p_device_id text)
RETURNS void
LANGUAGE sql
AS $$
INSERT INTO telemetry.devices (device_id)
VALUES ($1)
ON CONFLICT (device_id) DO NOTHING;
$$;
CREATE OR REPLACE FUNCTION telemetry.touch_device_last_seen(
p_device_id text,
p_observed_at timestamptz
)
RETURNS void
LANGUAGE sql
AS $$
UPDATE telemetry.devices
SET last_seen = CASE
WHEN last_seen IS NULL OR last_seen < $2 THEN $2
ELSE last_seen
END
WHERE device_id = $1;
$$;
CREATE OR REPLACE FUNCTION telemetry.require_device(p_device_id text)
RETURNS telemetry.devices
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_device telemetry.devices%ROWTYPE;
BEGIN
SELECT *
INTO v_device
FROM telemetry.devices AS d
WHERE d.device_id = p_device_id;
IF NOT FOUND THEN
RAISE EXCEPTION 'unknown device: %', p_device_id;
END IF;
RETURN v_device;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.lock_key_from_text(p_value text)
RETURNS integer
LANGUAGE sql
IMMUTABLE
STRICT
AS $$
SELECT (hashtextextended($1, 0) % 2147483647)::integer;
$$;
-- Segment template and table creation.
CREATE OR REPLACE FUNCTION telemetry.create_segment_table(
p_table_name text,
p_value_type text
)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
v_base_table_name text := telemetry.normalize_metric_table_name(p_table_name);
v_table_name text := format('telemetry.%I', v_base_table_name);
v_metric_name text;
v_initial_policy_id bigint;
v_open_idx text := format('%s_open_segment_uidx', v_base_table_name);
v_start_idx text := format('%s_device_start_cover_idx', v_base_table_name);
v_device_period_gist_idx text := format('%s_device_period_gist', v_base_table_name);
v_period_excl text := format('%s_device_period_excl', v_base_table_name);
v_device_pk_fk text := format('%s_device_pk_fkey', v_base_table_name);
v_device_fk text := format('%s_device_id_fkey', v_base_table_name);
v_policy_fk text := format('%s_policy_id_fkey', v_base_table_name);
v_samples_check text := format('%s_samples_count_check', v_base_table_name);
v_end_time_check text := format('%s_check', v_base_table_name);
BEGIN
IF p_value_type NOT IN ('double precision', 'boolean', 'smallint') THEN
RAISE EXCEPTION 'unsupported segment value type: %', p_value_type;
END IF;
SELECT m.metric_name
INTO v_metric_name
FROM telemetry.metrics AS m
WHERE m.table_name = v_base_table_name;
IF v_metric_name IS NULL THEN
RAISE EXCEPTION
'metric metadata must exist before creating segment table %',
v_base_table_name;
END IF;
SELECT mp.policy_id
INTO v_initial_policy_id
FROM telemetry.metric_policies AS mp
WHERE mp.metric_name = v_metric_name
AND mp.valid_from = '-infinity'::timestamptz;
IF v_initial_policy_id IS NULL THEN
RAISE EXCEPTION
'initial policy row is missing for metric %',
v_metric_name;
END IF;
EXECUTE format(
'CREATE TABLE IF NOT EXISTS %s (
segment_id bigserial PRIMARY KEY,
device_pk bigint NOT NULL,
device_id text NOT NULL,
start_time timestamptz NOT NULL,
end_time timestamptz NULL,
value %s NULL,
-- samples_count = 0 is intentional for synthetic gap segments inserted
-- during lazy gap detection; those intervals represent missing data, not samples.
samples_count integer NOT NULL DEFAULT 1
CONSTRAINT %I CHECK (samples_count >= 0),
policy_id bigint NOT NULL,
segment_period tstzrange GENERATED ALWAYS AS (
tstzrange(start_time, COALESCE(end_time, ''infinity''::timestamptz), ''[)'')
) STORED,
CONSTRAINT %I CHECK (end_time IS NULL OR end_time > start_time),
CONSTRAINT %I
FOREIGN KEY (device_pk)
REFERENCES telemetry.devices(device_pk)
ON UPDATE CASCADE
ON DELETE CASCADE,
CONSTRAINT %I
FOREIGN KEY (device_id)
REFERENCES telemetry.devices(device_id)
ON UPDATE CASCADE
ON DELETE CASCADE,
CONSTRAINT %I
FOREIGN KEY (policy_id)
REFERENCES telemetry.metric_policies(policy_id)
) WITH (
fillfactor = 90,
autovacuum_vacuum_scale_factor = 0.02,
autovacuum_analyze_scale_factor = 0.01
)',
v_table_name,
p_value_type,
v_samples_check,
v_end_time_check,
v_device_pk_fk,
v_device_fk,
v_policy_fk
);
EXECUTE format(
'CREATE UNIQUE INDEX IF NOT EXISTS %I
ON %s (device_id)
WHERE end_time IS NULL',
v_open_idx,
v_table_name
);
EXECUTE format(
'CREATE INDEX IF NOT EXISTS %I
ON %s (device_id, start_time DESC)
INCLUDE (end_time, value, samples_count)',
v_start_idx,
v_table_name
);
EXECUTE format(
'CREATE INDEX IF NOT EXISTS %I
ON %s
USING gist (device_id, segment_period)',
v_device_period_gist_idx,
v_table_name
);
BEGIN
EXECUTE format(
'ALTER TABLE %s
ADD CONSTRAINT %I
EXCLUDE USING gist (
device_id WITH =,
segment_period WITH &&
)',
v_table_name,
v_period_excl
);
EXCEPTION
WHEN duplicate_object THEN
NULL;
WHEN duplicate_table THEN
NULL;
END;
END;
$$;
-- Metric registration.
CREATE OR REPLACE FUNCTION telemetry.register_numeric_metric(
p_metric_name text,
p_table_name text,
p_domain_name text,
p_epsilon double precision,
p_min_value double precision,
p_max_value double precision,
p_rounding_precision double precision,
p_max_sampling_interval interval,
p_allow_null boolean DEFAULT true
)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
v_table_name text := telemetry.normalize_metric_table_name(p_table_name);
BEGIN
INSERT INTO telemetry.metrics (
metric_name,
table_name,
domain_name,
metric_type,
comparison_mode,
epsilon,
min_value,
max_value,
rounding_precision,
max_sampling_interval,
allow_null
)
VALUES (
p_metric_name,
v_table_name,
p_domain_name,
'numeric',
'epsilon',
p_epsilon,
p_min_value,
p_max_value,
p_rounding_precision,
p_max_sampling_interval,
p_allow_null
)
ON CONFLICT (metric_name) DO UPDATE
SET table_name = EXCLUDED.table_name,
domain_name = EXCLUDED.domain_name,
metric_type = EXCLUDED.metric_type,
comparison_mode = EXCLUDED.comparison_mode,
epsilon = EXCLUDED.epsilon,
min_value = EXCLUDED.min_value,
max_value = EXCLUDED.max_value,
rounding_precision = EXCLUDED.rounding_precision,
max_sampling_interval = EXCLUDED.max_sampling_interval,
allow_null = EXCLUDED.allow_null,
updated_at = now();
INSERT INTO telemetry.metric_policies (
metric_name,
valid_from,
comparison_mode,
epsilon,
min_value,
max_value,
rounding_precision,
allow_null,
max_sampling_interval
)
VALUES (
p_metric_name,
'-infinity'::timestamptz,
'epsilon',
p_epsilon,
p_min_value,
p_max_value,
p_rounding_precision,
p_allow_null,
p_max_sampling_interval
)
ON CONFLICT (metric_name, valid_from) DO NOTHING;
PERFORM telemetry.create_segment_table(v_table_name, 'double precision');
PERFORM telemetry.metric_table_sql(v_table_name);
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.register_boolean_metric(
p_metric_name text,
p_table_name text,
p_domain_name text,
p_max_sampling_interval interval,
p_allow_null boolean DEFAULT true
)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
v_table_name text := telemetry.normalize_metric_table_name(p_table_name);
BEGIN
INSERT INTO telemetry.metrics (
metric_name,
table_name,
domain_name,
metric_type,
comparison_mode,
epsilon,
min_value,
max_value,
rounding_precision,
max_sampling_interval,
allow_null
)
VALUES (
p_metric_name,
v_table_name,
p_domain_name,
'boolean',
'exact',
NULL,
NULL,
NULL,
NULL,
p_max_sampling_interval,
p_allow_null
)
ON CONFLICT (metric_name) DO UPDATE
SET table_name = EXCLUDED.table_name,
domain_name = EXCLUDED.domain_name,
metric_type = EXCLUDED.metric_type,
comparison_mode = EXCLUDED.comparison_mode,
epsilon = EXCLUDED.epsilon,
min_value = EXCLUDED.min_value,
max_value = EXCLUDED.max_value,
rounding_precision = EXCLUDED.rounding_precision,
max_sampling_interval = EXCLUDED.max_sampling_interval,
allow_null = EXCLUDED.allow_null,
updated_at = now();
INSERT INTO telemetry.metric_policies (
metric_name,
valid_from,
comparison_mode,
epsilon,
min_value,
max_value,
rounding_precision,
allow_null,
max_sampling_interval
)
VALUES (
p_metric_name,
'-infinity'::timestamptz,
'exact',
NULL,
NULL,
NULL,
NULL,
p_allow_null,
p_max_sampling_interval
)
ON CONFLICT (metric_name, valid_from) DO NOTHING;
PERFORM telemetry.create_segment_table(v_table_name, 'boolean');
PERFORM telemetry.metric_table_sql(v_table_name);
END;
$$;
-- Metric and policy lookup.
CREATE OR REPLACE FUNCTION telemetry.require_metric(p_metric_name text)
RETURNS telemetry.metrics
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
BEGIN
SELECT *
INTO v_metric
FROM telemetry.metrics AS m
WHERE m.metric_name = p_metric_name;
IF NOT FOUND THEN
RAISE EXCEPTION 'unknown metric: %', p_metric_name;
END IF;
RETURN v_metric;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.require_metric_policy(
p_metric_name text,
p_observed_at timestamptz
)
RETURNS telemetry.metric_policies
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_policy telemetry.metric_policies%ROWTYPE;
BEGIN
SELECT *
INTO v_policy
FROM telemetry.metric_policies AS mp
WHERE mp.metric_name = p_metric_name
AND mp.valid_from <= p_observed_at
ORDER BY mp.valid_from DESC
LIMIT 1;
IF NOT FOUND THEN
RAISE EXCEPTION
'no active policy for metric % at %',
p_metric_name,
p_observed_at;
END IF;
RETURN v_policy;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.add_metric_policy(
p_metric_name text,
p_valid_from timestamptz,
p_comparison_mode telemetry.comparison_mode_enum,
p_epsilon double precision,
p_min_value double precision,
p_max_value double precision,
p_rounding_precision double precision,
p_max_sampling_interval interval,
p_allow_null boolean
)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_policy_id bigint;
BEGIN
IF p_valid_from IS NULL THEN
RAISE EXCEPTION 'valid_from is required';
END IF;
IF p_max_sampling_interval IS NULL THEN
RAISE EXCEPTION 'max_sampling_interval is required';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
IF p_min_value IS NOT NULL
AND p_max_value IS NOT NULL
AND p_min_value > p_max_value THEN
RAISE EXCEPTION 'min_value % exceeds max_value % for metric %',
p_min_value,
p_max_value,
p_metric_name;
END IF;
CASE v_metric.metric_type
WHEN 'numeric' THEN
IF p_comparison_mode <> 'epsilon'::telemetry.comparison_mode_enum THEN
RAISE EXCEPTION
'numeric metric % requires comparison_mode epsilon',
p_metric_name;
END IF;
IF p_epsilon IS NULL OR p_rounding_precision IS NULL THEN
RAISE EXCEPTION
'numeric metric % requires epsilon and rounding_precision',
p_metric_name;
END IF;
WHEN 'boolean', 'state' THEN
IF p_comparison_mode <> 'exact'::telemetry.comparison_mode_enum THEN
RAISE EXCEPTION
'metric % with type % requires comparison_mode exact',
p_metric_name,
v_metric.metric_type;
END IF;
IF p_epsilon IS NOT NULL
OR p_min_value IS NOT NULL
OR p_max_value IS NOT NULL
OR p_rounding_precision IS NOT NULL THEN
RAISE EXCEPTION
'metric % with type % does not accept epsilon/min/max/rounding_precision in policies',
p_metric_name,
v_metric.metric_type;
END IF;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
IF EXISTS (
SELECT 1
FROM telemetry.metric_policies AS mp
WHERE mp.metric_name = p_metric_name
AND mp.valid_from = p_valid_from
) THEN
RAISE EXCEPTION
'policy already exists for metric % at %',
p_metric_name,
p_valid_from;
END IF;
INSERT INTO telemetry.metric_policies (
metric_name,
valid_from,
comparison_mode,
epsilon,
min_value,
max_value,
rounding_precision,
allow_null,
max_sampling_interval
)
VALUES (
p_metric_name,
p_valid_from,
p_comparison_mode,
p_epsilon,
p_min_value,
p_max_value,
p_rounding_precision,
p_allow_null,
p_max_sampling_interval
)
RETURNING policy_id
INTO v_policy_id;
UPDATE telemetry.metrics AS m
SET comparison_mode = p_comparison_mode,
epsilon = p_epsilon,
min_value = p_min_value,
max_value = p_max_value,
rounding_precision = p_rounding_precision,
allow_null = p_allow_null,
max_sampling_interval = p_max_sampling_interval,
updated_at = now()
WHERE m.metric_name = p_metric_name
AND NOT EXISTS (
SELECT 1
FROM telemetry.metric_policies AS later
WHERE later.metric_name = p_metric_name
AND later.valid_from > p_valid_from
);
RETURN v_policy_id;
END;
$$;
-- Ingestion functions.
CREATE OR REPLACE FUNCTION telemetry.assert_append_only(
p_metric_name text,
p_device_id text,
p_observed_at timestamptz
)
RETURNS timestamptz
LANGUAGE plpgsql
AS $$
DECLARE
v_last_observed_at timestamptz;
BEGIN
SELECT w.last_observed_at
INTO v_last_observed_at
FROM telemetry.metric_device_watermarks AS w
WHERE w.metric_name = p_metric_name
AND w.device_id = p_device_id;
IF FOUND AND p_observed_at <= v_last_observed_at THEN
RAISE EXCEPTION
'out-of-order measurement for metric % device %: incoming % <= last observed %',
p_metric_name,
p_device_id,
p_observed_at,
v_last_observed_at;
END IF;
RETURN v_last_observed_at;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.bump_watermark(
p_metric_name text,
p_device_id text,
p_observed_at timestamptz
)
RETURNS void
LANGUAGE sql
AS $$
INSERT INTO telemetry.metric_device_watermarks (
metric_name,
device_id,
last_observed_at
)
VALUES ($1, $2, $3)
ON CONFLICT ON CONSTRAINT metric_device_watermarks_pkey DO UPDATE
SET last_observed_at = EXCLUDED.last_observed_at;
$$;
CREATE OR REPLACE FUNCTION telemetry.ingest_segment(
p_table_name text,
p_metric_type telemetry.metric_type_enum,
p_comparison_mode telemetry.comparison_mode_enum,
p_policy_id bigint,
p_policy_valid_from timestamptz,
p_device_pk bigint,
p_device_id text,
p_value anyelement,
p_observed_at timestamptz,
p_last_observed_at timestamptz,
p_epsilon double precision,
p_max_sampling_interval interval,
p_allow_null boolean
)
RETURNS text
LANGUAGE plpgsql
AS $$
DECLARE
v_current record;
v_has_current boolean;
v_gap_start timestamptz;
v_rowcount bigint;
v_equal boolean;
v_policy_changed boolean;
v_effective_last_observed_at timestamptz := p_last_observed_at;
v_table_name text := telemetry.normalize_metric_table_name(p_table_name);
v_table_sql text := telemetry.metric_table_sql(p_table_name);
BEGIN
EXECUTE format(
'SELECT segment_id, device_pk, device_id, start_time, end_time, value, samples_count, policy_id
FROM %s
WHERE device_id = $1
AND end_time IS NULL
FOR UPDATE',
v_table_sql
)
INTO v_current
USING p_device_id;
GET DIAGNOSTICS v_rowcount = ROW_COUNT;
v_has_current := v_rowcount > 0;
IF p_value IS NULL AND NOT p_allow_null THEN
RAISE EXCEPTION 'metric table % does not allow explicit NULL measurements', v_table_name;
END IF;
IF NOT v_has_current THEN
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, $4, 1, $5)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_value, p_policy_id;
RETURN CASE WHEN p_value IS NULL THEN 'opened_null' ELSE 'opened' END;
END IF;
v_policy_changed := v_current.policy_id IS DISTINCT FROM p_policy_id;
IF v_policy_changed
AND p_policy_valid_from > v_current.start_time THEN
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING p_policy_valid_from, v_current.segment_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
-- samples_count = 0 marks a synthetic continuation across a policy boundary.
VALUES ($1, $2, $3, NULL, $4, 0, $5)
RETURNING segment_id, device_pk, device_id, start_time, end_time, value, samples_count, policy_id',
v_table_sql
)
INTO v_current
USING v_current.device_pk, v_current.device_id, p_policy_valid_from, v_current.value, p_policy_id;
v_policy_changed := false;
v_effective_last_observed_at := GREATEST(
COALESCE(p_last_observed_at, p_policy_valid_from),
p_policy_valid_from
);
END IF;
IF v_effective_last_observed_at IS NOT NULL
AND p_observed_at - v_effective_last_observed_at > p_max_sampling_interval
AND v_current.value IS NOT NULL THEN
v_gap_start := v_effective_last_observed_at + p_max_sampling_interval;
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING v_gap_start, v_current.segment_id;
IF p_value IS NULL THEN
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, NULL, 1, $4)',
v_table_sql
)
USING p_device_pk, p_device_id, v_gap_start, p_policy_id;
RETURN 'gap_to_null';
END IF;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
-- samples_count = 0 marks a historian-generated gap interval.
VALUES ($1, $2, $3, $4, NULL, 0, $5)',
v_table_sql
)
USING p_device_pk, p_device_id, v_gap_start, p_observed_at, p_policy_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, $4, 1, $5)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_value, p_policy_id;
RETURN 'gap_split';
END IF;
IF v_current.samples_count = 0
AND p_observed_at = v_current.start_time THEN
IF p_value IS NULL THEN
EXECUTE format(
'UPDATE %s
SET value = NULL,
samples_count = 1
WHERE segment_id = $1',
v_table_sql
)
USING v_current.segment_id;
RETURN CASE
WHEN v_current.value IS NULL THEN 'extended_null'
ELSE 'value_to_null'
END;
END IF;
IF v_current.value IS NULL THEN
EXECUTE format(
'UPDATE %s
SET value = $1,
samples_count = 1
WHERE segment_id = $2',
v_table_sql
)
USING p_value, v_current.segment_id;
RETURN 'null_to_value';
END IF;
CASE p_comparison_mode
WHEN 'epsilon' THEN
EXECUTE
'SELECT abs($1::double precision - $2::double precision) <= $3'
INTO v_equal
USING v_current.value, p_value, p_epsilon;
WHEN 'exact' THEN
EXECUTE
'SELECT $1 IS NOT DISTINCT FROM $2'
INTO v_equal
USING v_current.value, p_value;
ELSE
RAISE EXCEPTION 'unsupported comparison_mode % for table %', p_comparison_mode, v_table_name;
END CASE;
EXECUTE format(
'UPDATE %s
SET value = $1,
samples_count = 1
WHERE segment_id = $2',
v_table_sql
)
USING p_value, v_current.segment_id;
RETURN CASE
WHEN v_equal THEN 'extended'
ELSE 'split'
END;
END IF;
IF v_current.value IS NULL THEN
IF p_value IS NULL THEN
IF v_policy_changed THEN
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING p_observed_at, v_current.segment_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, NULL, 1, $4)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_policy_id;
RETURN 'split';
END IF;
EXECUTE format(
'UPDATE %s
SET samples_count = samples_count + 1
WHERE segment_id = $1',
v_table_sql
)
USING v_current.segment_id;
RETURN 'extended_null';
END IF;
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING p_observed_at, v_current.segment_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, $4, 1, $5)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_value, p_policy_id;
RETURN 'null_to_value';
END IF;
IF p_value IS NULL THEN
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING p_observed_at, v_current.segment_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, NULL, 1, $4)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_policy_id;
RETURN 'value_to_null';
END IF;
CASE p_comparison_mode
WHEN 'epsilon' THEN
EXECUTE
'SELECT abs($1::double precision - $2::double precision) <= $3'
INTO v_equal
USING v_current.value, p_value, p_epsilon;
WHEN 'exact' THEN
EXECUTE
'SELECT $1 IS NOT DISTINCT FROM $2'
INTO v_equal
USING v_current.value, p_value;
ELSE
RAISE EXCEPTION 'unsupported comparison_mode % for table %', p_comparison_mode, v_table_name;
END CASE;
IF v_equal AND NOT v_policy_changed THEN
EXECUTE format(
'UPDATE %s
SET samples_count = samples_count + 1
WHERE segment_id = $1',
v_table_sql
)
USING v_current.segment_id;
RETURN 'extended';
END IF;
EXECUTE format(
'UPDATE %s
SET end_time = $1
WHERE segment_id = $2',
v_table_sql
)
USING p_observed_at, v_current.segment_id;
EXECUTE format(
'INSERT INTO %s (
device_pk,
device_id,
start_time,
end_time,
value,
samples_count,
policy_id
)
VALUES ($1, $2, $3, NULL, $4, 1, $5)',
v_table_sql
)
USING p_device_pk, p_device_id, p_observed_at, p_value, p_policy_id;
RETURN 'split';
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.ingest_measurement(
p_metric_name text,
p_device_id text,
p_value double precision,
p_observed_at timestamptz
)
RETURNS TABLE (
metric_name text,
device_id text,
table_name text,
normalized_value double precision,
action text
)
LANGUAGE plpgsql
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_policy telemetry.metric_policies%ROWTYPE;
v_device telemetry.devices%ROWTYPE;
v_last_observed_at timestamptz;
v_normalized double precision;
v_action text;
v_metric_lock_key integer;
v_device_lock_key integer;
BEGIN
IF p_metric_name IS NULL OR btrim(p_metric_name) = '' THEN
RAISE EXCEPTION 'metric_name is required';
END IF;
IF p_device_id IS NULL OR btrim(p_device_id) = '' THEN
RAISE EXCEPTION 'device_id is required';
END IF;
IF p_observed_at IS NULL THEN
RAISE EXCEPTION 'observed_at is required';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
PERFORM telemetry.ensure_device(p_device_id);
v_device := telemetry.require_device(p_device_id);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'numeric' THEN
NULL;
WHEN 'boolean' THEN
RAISE EXCEPTION 'metric % is boolean; use the boolean overload of ingest_measurement(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state ingestion is not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
v_metric_lock_key := telemetry.lock_key_from_text(p_metric_name);
v_device_lock_key := telemetry.lock_key_from_text(p_device_id);
PERFORM pg_advisory_xact_lock(v_metric_lock_key, v_device_lock_key);
v_last_observed_at := telemetry.assert_append_only(p_metric_name, p_device_id, p_observed_at);
v_policy := telemetry.require_metric_policy(p_metric_name, p_observed_at);
IF p_value IS NULL THEN
v_normalized := NULL;
ELSE
v_normalized := telemetry.round_to_increment(p_value, v_policy.rounding_precision);
IF v_policy.min_value IS NOT NULL AND v_normalized < v_policy.min_value THEN
RAISE EXCEPTION 'value % is below min_value % for metric %',
v_normalized, v_policy.min_value, p_metric_name;
END IF;
IF v_policy.max_value IS NOT NULL AND v_normalized > v_policy.max_value THEN
RAISE EXCEPTION 'value % is above max_value % for metric %',
v_normalized, v_policy.max_value, p_metric_name;
END IF;
END IF;
v_action := telemetry.ingest_segment(
v_metric.table_name,
v_metric.metric_type,
v_policy.comparison_mode,
v_policy.policy_id,
v_policy.valid_from,
v_device.device_pk,
p_device_id,
v_normalized,
p_observed_at,
v_last_observed_at,
v_policy.epsilon,
v_policy.max_sampling_interval,
v_policy.allow_null
);
PERFORM telemetry.bump_watermark(p_metric_name, p_device_id, p_observed_at);
PERFORM telemetry.touch_device_last_seen(p_device_id, p_observed_at);
RETURN QUERY
SELECT
p_metric_name,
p_device_id,
v_metric.table_name,
v_normalized,
v_action;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.ingest_measurement(
p_metric_name text,
p_device_id text,
p_value boolean,
p_observed_at timestamptz
)
RETURNS TABLE (
metric_name text,
device_id text,
table_name text,
normalized_value boolean,
action text
)
LANGUAGE plpgsql
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_policy telemetry.metric_policies%ROWTYPE;
v_device telemetry.devices%ROWTYPE;
v_last_observed_at timestamptz;
v_action text;
v_metric_lock_key integer;
v_device_lock_key integer;
BEGIN
IF p_metric_name IS NULL OR btrim(p_metric_name) = '' THEN
RAISE EXCEPTION 'metric_name is required';
END IF;
IF p_device_id IS NULL OR btrim(p_device_id) = '' THEN
RAISE EXCEPTION 'device_id is required';
END IF;
IF p_observed_at IS NULL THEN
RAISE EXCEPTION 'observed_at is required';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
PERFORM telemetry.ensure_device(p_device_id);
v_device := telemetry.require_device(p_device_id);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'boolean' THEN
NULL;
WHEN 'numeric' THEN
RAISE EXCEPTION 'metric % is numeric; use the numeric overload of ingest_measurement(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state ingestion is not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
v_metric_lock_key := telemetry.lock_key_from_text(p_metric_name);
v_device_lock_key := telemetry.lock_key_from_text(p_device_id);
PERFORM pg_advisory_xact_lock(v_metric_lock_key, v_device_lock_key);
v_last_observed_at := telemetry.assert_append_only(p_metric_name, p_device_id, p_observed_at);
v_policy := telemetry.require_metric_policy(p_metric_name, p_observed_at);
v_action := telemetry.ingest_segment(
v_metric.table_name,
v_metric.metric_type,
v_policy.comparison_mode,
v_policy.policy_id,
v_policy.valid_from,
v_device.device_pk,
p_device_id,
p_value,
p_observed_at,
v_last_observed_at,
v_policy.epsilon,
v_policy.max_sampling_interval,
v_policy.allow_null
);
PERFORM telemetry.bump_watermark(p_metric_name, p_device_id, p_observed_at);
PERFORM telemetry.touch_device_last_seen(p_device_id, p_observed_at);
RETURN QUERY
SELECT
p_metric_name,
p_device_id,
v_metric.table_name,
p_value,
v_action;
END;
$$;
-- Query functions.
CREATE OR REPLACE FUNCTION telemetry.value_at_from_table(
p_table_name text,
p_metric_name text,
p_device_id text,
p_at timestamptz
)
RETURNS jsonb
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_value jsonb;
v_table_sql text := telemetry.metric_table_sql(p_table_name);
BEGIN
EXECUTE format(
'WITH watermark AS (
SELECT w.last_observed_at
FROM telemetry.metric_device_watermarks AS w
WHERE w.metric_name = $1
AND w.device_id = $2
)
SELECT to_jsonb(s.value)
FROM %s AS s
JOIN telemetry.metric_policies AS mp
ON mp.policy_id = s.policy_id
LEFT JOIN watermark AS w ON true
WHERE s.device_id = $2
AND tstzrange(
s.start_time,
CASE
WHEN s.end_time IS NOT NULL THEN s.end_time
WHEN s.value IS NULL THEN ''infinity''::timestamptz
ELSE COALESCE(
GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval,
''infinity''::timestamptz
)
END,
''[)''
) @> $3
LIMIT 1',
v_table_sql
)
INTO v_value
USING p_metric_name, p_device_id, p_at;
RETURN v_value;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.value_at(
p_metric_name text,
p_device_id text,
p_at timestamptz
)
RETURNS double precision
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_value jsonb;
BEGIN
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'numeric' THEN
v_value := telemetry.value_at_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_at
);
RETURN CASE
WHEN v_value IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first; direct
-- casts such as (v_value)::double precision are not supported.
ELSE v_value::text::double precision
END;
WHEN 'boolean' THEN
RAISE EXCEPTION 'metric % is boolean; use telemetry.boolean_value_at(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.boolean_value_at(
p_metric_name text,
p_device_id text,
p_at timestamptz
)
RETURNS boolean
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_value jsonb;
BEGIN
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'boolean' THEN
v_value := telemetry.value_at_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_at
);
RETURN CASE
WHEN v_value IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first; direct
-- casts such as (v_value)::boolean are not supported.
ELSE v_value::text::boolean
END;
WHEN 'numeric' THEN
RAISE EXCEPTION 'metric % is numeric; use telemetry.value_at(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.segments_between_from_table(
p_table_name text,
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz
)
RETURNS TABLE (
device_id text,
start_time timestamptz,
end_time timestamptz,
value_json jsonb,
samples_count integer
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_table_sql text := telemetry.metric_table_sql(p_table_name);
BEGIN
RETURN QUERY EXECUTE format(
'WITH watermark AS (
SELECT w.last_observed_at
FROM telemetry.metric_device_watermarks AS w
WHERE w.metric_name = $1
AND w.device_id = $2
),
open_segment AS (
SELECT s.*
FROM %s AS s
WHERE s.device_id = $2
AND s.end_time IS NULL
LIMIT 1
),
stored_segments AS (
SELECT
s.device_id,
GREATEST(s.start_time, $3) AS start_time,
LEAST(
CASE
WHEN s.end_time IS NOT NULL THEN s.end_time
WHEN s.value IS NULL THEN $4
ELSE LEAST(
COALESCE(
GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval,
$4
),
$4
)
END,
$4
) AS end_time,
to_jsonb(s.value) AS value_json,
s.samples_count
FROM %s AS s
JOIN telemetry.metric_policies AS mp
ON mp.policy_id = s.policy_id
LEFT JOIN watermark AS w ON true
WHERE s.device_id = $2
AND tstzrange(
s.start_time,
CASE
WHEN s.end_time IS NOT NULL THEN s.end_time
WHEN s.value IS NULL THEN ''infinity''::timestamptz
ELSE COALESCE(
GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval,
''infinity''::timestamptz
)
END,
''[)''
) && tstzrange($3, $4, ''[)'')
),
synthetic_tail AS (
SELECT
$2::text AS device_id,
GREATEST(
GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval,
$3
) AS start_time,
$4 AS end_time,
NULL::jsonb AS value_json,
0 AS samples_count
FROM open_segment AS s
JOIN telemetry.metric_policies AS mp
ON mp.policy_id = s.policy_id
JOIN watermark AS w ON true
WHERE s.value IS NOT NULL
AND GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval < $4
)
SELECT *
FROM (
SELECT *
FROM stored_segments
WHERE end_time > start_time
UNION ALL
SELECT *
FROM synthetic_tail
WHERE end_time > start_time
) AS combined_segments
ORDER BY start_time',
v_table_sql,
v_table_sql
)
USING p_metric_name, p_device_id, p_from, p_to;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.metric_segments(
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz
)
RETURNS TABLE (
device_id text,
start_time timestamptz,
end_time timestamptz,
value double precision,
samples_count integer
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
BEGIN
IF p_from IS NULL OR p_to IS NULL OR p_to <= p_from THEN
RAISE EXCEPTION 'p_from and p_to must define a non-empty interval';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'numeric' THEN
RETURN QUERY
SELECT
s.device_id,
s.start_time,
s.end_time,
CASE
WHEN s.value_json IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first;
-- direct casts such as (s.value_json)::double precision
-- are not supported.
ELSE s.value_json::text::double precision
END AS value,
s.samples_count
FROM telemetry.segments_between_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_from,
p_to
) AS s;
WHEN 'boolean' THEN
RAISE EXCEPTION 'metric % is boolean; use telemetry.boolean_segments_between(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.boolean_segments_between(
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz
)
RETURNS TABLE (
device_id text,
start_time timestamptz,
end_time timestamptz,
value boolean,
samples_count integer
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
BEGIN
IF p_from IS NULL OR p_to IS NULL OR p_to <= p_from THEN
RAISE EXCEPTION 'p_from and p_to must define a non-empty interval';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'boolean' THEN
RETURN QUERY
SELECT
s.device_id,
s.start_time,
s.end_time,
CASE
WHEN s.value_json IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first;
-- direct casts such as (s.value_json)::boolean are not
-- supported.
ELSE s.value_json::text::boolean
END AS value,
s.samples_count
FROM telemetry.segments_between_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_from,
p_to
) AS s;
WHEN 'numeric' THEN
RAISE EXCEPTION 'metric % is numeric; use telemetry.metric_segments(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.samples_from_table(
p_table_name text,
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz,
p_points integer
)
RETURNS TABLE (
sample_at timestamptz,
value_json jsonb
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_table_sql text := telemetry.metric_table_sql(p_table_name);
BEGIN
RETURN QUERY EXECUTE format(
'WITH watermark AS (
SELECT w.last_observed_at
FROM telemetry.metric_device_watermarks AS w
WHERE w.metric_name = $1
AND w.device_id = $2
),
sample_points AS (
SELECT $3 + (($4 - $3) * gs.i::double precision / ($5 - 1)::double precision) AS sample_at
FROM generate_series(0, $5 - 1) AS gs(i)
)
SELECT
sp.sample_at,
to_jsonb(seg.value) AS value_json
FROM sample_points AS sp
LEFT JOIN watermark AS w ON true
LEFT JOIN LATERAL (
SELECT s.value
FROM %s AS s
JOIN telemetry.metric_policies AS mp
ON mp.policy_id = s.policy_id
WHERE s.device_id = $2
AND tstzrange(
s.start_time,
CASE
WHEN s.end_time IS NOT NULL THEN s.end_time
WHEN s.value IS NULL THEN ''infinity''::timestamptz
ELSE COALESCE(
GREATEST(COALESCE(w.last_observed_at, s.start_time), s.start_time)
+ mp.max_sampling_interval,
''infinity''::timestamptz
)
END,
''[)''
) @> sp.sample_at
LIMIT 1
) AS seg ON true
ORDER BY sp.sample_at',
v_table_sql
)
USING p_metric_name, p_device_id, p_from, p_to, p_points;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.sample_metric(
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz,
p_points integer
)
RETURNS TABLE (
sample_at timestamptz,
value double precision
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
BEGIN
IF p_points IS NULL OR p_points < 2 THEN
RAISE EXCEPTION 'p_points must be at least 2';
END IF;
IF p_from IS NULL OR p_to IS NULL OR p_to <= p_from THEN
RAISE EXCEPTION 'p_from and p_to must define a non-empty interval';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'numeric' THEN
RETURN QUERY
SELECT
s.sample_at,
CASE
WHEN s.value_json IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first;
-- direct casts such as (s.value_json)::double precision
-- are not supported.
ELSE s.value_json::text::double precision
END AS value
FROM telemetry.samples_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_from,
p_to,
p_points
) AS s;
WHEN 'boolean' THEN
RAISE EXCEPTION 'metric % is boolean; use telemetry.boolean_samples(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.boolean_samples(
p_metric_name text,
p_device_id text,
p_from timestamptz,
p_to timestamptz,
p_points integer
)
RETURNS TABLE (
sample_at timestamptz,
value boolean
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
BEGIN
IF p_points IS NULL OR p_points < 2 THEN
RAISE EXCEPTION 'p_points must be at least 2';
END IF;
IF p_from IS NULL OR p_to IS NULL OR p_to <= p_from THEN
RAISE EXCEPTION 'p_from and p_to must define a non-empty interval';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
-- 'state' is reserved for future implementation of discrete enumerated
-- machine states, likely stored as smallint. It will use the same segment
-- logic as boolean metrics, but with exact comparison instead of epsilon.
CASE v_metric.metric_type
WHEN 'boolean' THEN
RETURN QUERY
SELECT
s.sample_at,
CASE
WHEN s.value_json IS NULL THEN NULL
-- PostgreSQL must cast jsonb scalars via text first;
-- direct casts such as (s.value_json)::boolean are not
-- supported.
ELSE s.value_json::text::boolean
END AS value
FROM telemetry.samples_from_table(
v_metric.table_name,
p_metric_name,
p_device_id,
p_from,
p_to,
p_points
) AS s;
WHEN 'numeric' THEN
RAISE EXCEPTION 'metric % is numeric; use telemetry.sample_metric(...)', p_metric_name;
WHEN 'state' THEN
RAISE EXCEPTION 'metric % is state; state query wrappers are not implemented yet', p_metric_name;
ELSE
RAISE EXCEPTION 'unsupported metric_type % for metric %', v_metric.metric_type, p_metric_name;
END CASE;
END;
$$;
-- Maintenance utilities.
CREATE OR REPLACE FUNCTION telemetry.verify_segments(p_metric_name text)
RETURNS TABLE (
metric_name text,
device_id text,
issue text,
segment_id bigint,
related_segment_id bigint,
start_time timestamptz,
end_time timestamptz,
details text
)
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_table_sql text;
BEGIN
v_metric := telemetry.require_metric(p_metric_name);
v_table_sql := telemetry.metric_table_sql(v_metric.table_name);
RETURN QUERY EXECUTE format(
'WITH ordered AS (
SELECT
s.segment_id,
s.device_id,
s.policy_id,
s.start_time,
s.end_time,
lag(s.segment_id) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_segment_id,
lag(s.end_time) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_end_time,
lag(s.policy_id) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_policy_id
FROM %s AS s
),
open_counts AS (
SELECT
s.device_id,
count(*)::integer AS open_count
FROM %s AS s
WHERE s.end_time IS NULL
GROUP BY s.device_id
HAVING count(*) > 1
)
SELECT
$1::text AS metric_name,
o.device_id,
''invalid_interval''::text AS issue,
o.segment_id,
NULL::bigint AS related_segment_id,
o.start_time,
o.end_time,
''end_time must be greater than start_time''::text AS details
FROM ordered AS o
WHERE o.end_time IS NOT NULL
AND o.end_time <= o.start_time
UNION ALL
SELECT
$1::text,
oc.device_id,
''multiple_open_segments''::text,
NULL::bigint,
NULL::bigint,
NULL::timestamptz,
NULL::timestamptz,
format(''%%s open segments found'', oc.open_count)
FROM open_counts AS oc
UNION ALL
SELECT
$1::text,
o.device_id,
''overlap''::text,
o.segment_id,
o.prev_segment_id,
o.start_time,
o.end_time,
format(''previous segment ends at %%s'', o.prev_end_time)
FROM ordered AS o
WHERE o.prev_end_time IS NOT NULL
AND o.prev_end_time > o.start_time
UNION ALL
SELECT
$1::text,
o.device_id,
''unexpected_gap''::text,
o.segment_id,
o.prev_segment_id,
o.start_time,
o.end_time,
format(
''stored gap of %%s violates expected continuity (max_sampling_interval=%%s)'',
o.start_time - o.prev_end_time,
mp.max_sampling_interval
)
FROM ordered AS o
JOIN telemetry.metric_policies AS mp
ON mp.policy_id = o.prev_policy_id
WHERE o.prev_end_time IS NOT NULL
AND o.start_time > o.prev_end_time
ORDER BY device_id, start_time NULLS FIRST, segment_id NULLS FIRST',
v_table_sql,
v_table_sql
)
USING p_metric_name;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.compact_segments(
p_metric_name text,
p_before timestamptz
)
RETURNS TABLE (
device_id text,
kept_segment_id bigint,
merged_segments integer,
start_time timestamptz,
end_time timestamptz,
total_samples integer
)
LANGUAGE plpgsql
AS $$
DECLARE
v_metric telemetry.metrics%ROWTYPE;
v_table_sql text;
v_merge record;
BEGIN
IF p_before IS NULL THEN
RAISE EXCEPTION 'p_before is required';
END IF;
v_metric := telemetry.require_metric(p_metric_name);
v_table_sql := telemetry.metric_table_sql(v_metric.table_name);
FOR v_merge IN EXECUTE format(
'WITH candidates AS (
SELECT
s.segment_id,
s.device_id,
s.start_time,
s.end_time,
s.samples_count,
s.policy_id,
s.value,
lag(s.end_time) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_end_time,
lag(s.policy_id) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_policy_id,
lag(s.value) OVER (
PARTITION BY s.device_id
ORDER BY s.start_time, s.segment_id
) AS prev_value
FROM %s AS s
WHERE s.end_time IS NOT NULL
AND s.end_time <= $1
),
grouped AS (
SELECT
c.*,
sum(
CASE
WHEN c.prev_end_time IS NULL
OR c.start_time <> c.prev_end_time
OR c.policy_id IS DISTINCT FROM c.prev_policy_id
OR c.value IS DISTINCT FROM c.prev_value
THEN 1
ELSE 0
END
) OVER (
PARTITION BY c.device_id
ORDER BY c.start_time, c.segment_id
) AS grp
FROM candidates AS c
),
aggregates AS (
SELECT
g.device_id,
(array_agg(g.segment_id ORDER BY g.start_time, g.segment_id))[1] AS keep_segment_id,
array_remove(
array_agg(g.segment_id ORDER BY g.start_time, g.segment_id),
(array_agg(g.segment_id ORDER BY g.start_time, g.segment_id))[1]
) AS delete_segment_ids,
min(g.start_time) AS merged_start_time,
max(g.end_time) AS merged_end_time,
sum(g.samples_count)::integer AS total_samples,
count(*)::integer AS merged_segments
FROM grouped AS g
GROUP BY g.device_id, g.grp
HAVING count(*) > 1
)
SELECT
a.device_id,
a.keep_segment_id,
a.delete_segment_ids,
a.merged_segments,
a.merged_start_time,
a.merged_end_time,
a.total_samples
FROM aggregates AS a
ORDER BY a.device_id, a.merged_start_time',
v_table_sql
)
USING p_before
LOOP
EXECUTE format(
'DELETE FROM %s
WHERE segment_id = ANY($1)',
v_table_sql
)
USING v_merge.delete_segment_ids;
EXECUTE format(
'UPDATE %s
SET end_time = $1,
samples_count = $2
WHERE segment_id = $3',
v_table_sql
)
USING v_merge.merged_end_time, v_merge.total_samples, v_merge.keep_segment_id;
device_id := v_merge.device_id;
kept_segment_id := v_merge.keep_segment_id;
merged_segments := v_merge.merged_segments;
start_time := v_merge.merged_start_time;
end_time := v_merge.merged_end_time;
total_samples := v_merge.total_samples;
RETURN NEXT;
END LOOP;
RETURN;
END;
$$;
CREATE OR REPLACE FUNCTION telemetry.inactive_devices(p_threshold interval)
RETURNS TABLE (
device_pk bigint,
device_id text,
device_type text,
location text,
last_seen timestamptz,
inactive_for interval
)
LANGUAGE sql
STABLE
AS $$
SELECT
d.device_pk,
d.device_id,
d.device_type,
d.location,
d.last_seen,
now() - d.last_seen AS inactive_for
FROM telemetry.devices AS d
WHERE d.last_seen IS NOT NULL
AND now() - d.last_seen > $1
ORDER BY inactive_for DESC, d.device_id;
$$;
-- Example metric registrations.
SELECT telemetry.register_numeric_metric(
p_metric_name => 'ambient_temperature',
p_table_name => 'ambient_temperature_segments',
p_domain_name => 'environmental',
p_epsilon => 0.05,
p_min_value => -40,
p_max_value => 60,
p_rounding_precision => 0.1,
p_max_sampling_interval => '5 minutes',
p_allow_null => true
);
SELECT telemetry.register_numeric_metric(
p_metric_name => 'cpu_temperature',
p_table_name => 'cpu_temperature_segments',
p_domain_name => 'system',
p_epsilon => 0.5,
p_min_value => 0,
p_max_value => 120,
p_rounding_precision => 1,
p_max_sampling_interval => '10 seconds',
p_allow_null => true
);
SELECT telemetry.register_numeric_metric(
p_metric_name => 'humidity',
p_table_name => 'humidity_segments',
p_domain_name => 'environmental',
p_epsilon => 0.5,
p_min_value => 0,
p_max_value => 100,
p_rounding_precision => 0.1,
p_max_sampling_interval => '5 minutes',
p_allow_null => true
);
COMMIT;