Canonical TSAPI series DB schema v1.0.0
Single source of truth for the timeseries PostgreSQL schema shared by the Python (TSAPI_SERIESDB_PG_PY) and C++ (TSAPI_POSTGRES_CPP) libraries. SQL seeded byte-for-byte from TSAPI_SERIESDB_PG_PY/tsapi_pgdb/sql on 2026-09-02. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
commit
eed06559c0
17 changed files with 2080 additions and 0 deletions
426
sql/05_procedures.sql
Normal file
426
sql/05_procedures.sql
Normal file
|
|
@ -0,0 +1,426 @@
|
|||
-- Stored procedures and functions
|
||||
-- Generated: 2025-10-15T12:24:56.582392
|
||||
|
||||
-- Log schema functions
|
||||
|
||||
-- Function: log.notify_event_log
|
||||
-- Writes a breadcrumb to the staging table (UNLOGGED, fast).
|
||||
-- Self-drains every 20,000th insert to prevent unbounded growth.
|
||||
CREATE OR REPLACE FUNCTION log.notify_event_log(
|
||||
p_series_name TEXT,
|
||||
p_db_schema TEXT,
|
||||
p_op_type TEXT,
|
||||
p_details JSONB DEFAULT NULL
|
||||
) RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
BEGIN
|
||||
INSERT INTO log.pending_changes (series_name, db_schema, op_type, details)
|
||||
VALUES (p_series_name, p_db_schema, p_op_type, p_details);
|
||||
|
||||
-- Self-drain every 20,000th insert
|
||||
IF (currval('log.pending_changes_id_seq') % 20000 = 0) THEN
|
||||
PERFORM log.flush_pending_to_event_log();
|
||||
END IF;
|
||||
EXCEPTION WHEN OTHERS THEN
|
||||
RAISE WARNING 'notify_event_log failed: %', SQLERRM;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: log.flush_pending_to_event_log
|
||||
-- Drains the staging table into the permanent event_log.
|
||||
-- Safe for concurrent use: snapshots max(id) before processing.
|
||||
-- After moving events, verifies that meta.series timestamps match actual
|
||||
-- data for any series that had data-modifying events. If a mismatch is
|
||||
-- found, writes a TIMESTAMP_MISMATCH event to log.event_log for visibility.
|
||||
-- (Timestamps are kept accurate by the application on every data operation,
|
||||
-- so this step is a safety net, not a correction mechanism.)
|
||||
CREATE OR REPLACE FUNCTION log.flush_pending_to_event_log()
|
||||
RETURNS INT
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
batch_id BIGINT;
|
||||
v_count INT;
|
||||
_rec RECORD;
|
||||
_actual_first TIMESTAMP;
|
||||
_actual_last TIMESTAMP;
|
||||
_meta_first TIMESTAMP;
|
||||
_meta_last TIMESTAMP;
|
||||
BEGIN
|
||||
-- Serialise flushes. Snapshotting MAX(id) does not make this safe on its
|
||||
-- own: two callers can both read the same batch_id, both copy those rows
|
||||
-- to log.event_log, and only then delete them -- duplicating permanent
|
||||
-- audit events. The lock is transaction-scoped, so it is released when
|
||||
-- this function's transaction ends, however it ends.
|
||||
PERFORM pg_advisory_xact_lock(hashtext('tsapi_pgdb.flush_event_log'));
|
||||
|
||||
SELECT COALESCE(MAX(id), 0) INTO batch_id FROM log.pending_changes;
|
||||
IF batch_id = 0 THEN RETURN 0; END IF;
|
||||
|
||||
-- 1. Move rows from staging to permanent event_log
|
||||
INSERT INTO log.event_log ("timestamp", series_name, db_schema, event_type, details, source, db_user)
|
||||
SELECT "timestamp", series_name, db_schema, op_type, details, 'app', db_user
|
||||
FROM log.pending_changes
|
||||
WHERE id <= batch_id
|
||||
ORDER BY id;
|
||||
|
||||
GET DIAGNOSTICS v_count = ROW_COUNT;
|
||||
|
||||
-- 2. Verify meta.series timestamps for series with data-modifying events
|
||||
FOR _rec IN
|
||||
SELECT DISTINCT series_name, db_schema
|
||||
FROM log.pending_changes
|
||||
WHERE id <= batch_id
|
||||
AND op_type IN (
|
||||
'CSV_IMPORT', 'JSON_APPEND', 'DATAFRAME_IMPORT',
|
||||
'DELETE_FROM', 'DELETE_AFTER', 'DELETE_RANGE', 'DELETE_UNTIL',
|
||||
'DELETE_WEEKENDS', 'DELETE_TIME_RANGE', 'SERIES_TRUNCATED'
|
||||
)
|
||||
LOOP
|
||||
BEGIN
|
||||
EXECUTE format(
|
||||
'SELECT MIN(timestamp), MAX(timestamp) FROM %I.%I',
|
||||
_rec.db_schema, _rec.series_name
|
||||
) INTO _actual_first, _actual_last;
|
||||
|
||||
SELECT first_timestamp, last_timestamp INTO _meta_first, _meta_last
|
||||
FROM meta.series
|
||||
WHERE table_name = _rec.series_name AND db_schema = _rec.db_schema;
|
||||
|
||||
IF _meta_first IS DISTINCT FROM _actual_first
|
||||
OR _meta_last IS DISTINCT FROM _actual_last THEN
|
||||
INSERT INTO log.event_log
|
||||
("timestamp", series_name, db_schema, event_type, details, source, db_user)
|
||||
VALUES (
|
||||
CURRENT_TIMESTAMP,
|
||||
_rec.series_name,
|
||||
_rec.db_schema,
|
||||
'TIMESTAMP_MISMATCH',
|
||||
jsonb_build_object(
|
||||
'meta_first', _meta_first,
|
||||
'meta_last', _meta_last,
|
||||
'actual_first', _actual_first,
|
||||
'actual_last', _actual_last
|
||||
),
|
||||
'flush_verify',
|
||||
CURRENT_USER
|
||||
);
|
||||
END IF;
|
||||
EXCEPTION WHEN OTHERS THEN
|
||||
NULL; -- table may have been dropped in same batch
|
||||
END;
|
||||
END LOOP;
|
||||
|
||||
-- 3. Clear staging
|
||||
DELETE FROM log.pending_changes WHERE id <= batch_id;
|
||||
|
||||
RETURN v_count;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Meta schema functions
|
||||
|
||||
-- Function: meta.update_series_last_timestamp
|
||||
-- Lightweight timestamp update for append operations.
|
||||
-- Uses GREATEST to conditionally update last_timestamp (no table scan).
|
||||
-- Sets first_timestamp on first insert via COALESCE.
|
||||
-- p_first_timestamp is optional: when NULL, uses p_last_timestamp for first_timestamp.
|
||||
-- For single inserts, only p_last_timestamp is needed.
|
||||
-- For batch inserts, pass min timestamp as p_first_timestamp.
|
||||
-- Both Python and C++ clients call this after appending data.
|
||||
CREATE OR REPLACE FUNCTION meta.update_series_last_timestamp(
|
||||
p_table_name TEXT,
|
||||
p_db_schema TEXT DEFAULT 'series',
|
||||
p_last_timestamp TIMESTAMP DEFAULT NULL,
|
||||
p_first_timestamp TIMESTAMP DEFAULT NULL
|
||||
) RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
BEGIN
|
||||
UPDATE meta.series
|
||||
SET first_timestamp = COALESCE(first_timestamp, p_first_timestamp, p_last_timestamp),
|
||||
last_timestamp = GREATEST(last_timestamp, p_last_timestamp)
|
||||
WHERE table_name = p_table_name AND db_schema = p_db_schema;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.refresh_series_timestamps
|
||||
-- Full timestamp refresh using MIN/MAX scan.
|
||||
-- Used after deletes, truncates, or any operation that may change first_timestamp.
|
||||
-- BRIN indexes make MIN/MAX fast even on large tables.
|
||||
-- Both Python and C++ clients call this after data deletion.
|
||||
CREATE OR REPLACE FUNCTION meta.refresh_series_timestamps(
|
||||
p_table_name TEXT,
|
||||
p_db_schema TEXT DEFAULT 'series'
|
||||
) RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
_first_ts TIMESTAMP;
|
||||
_last_ts TIMESTAMP;
|
||||
BEGIN
|
||||
EXECUTE format(
|
||||
'SELECT MIN(timestamp), MAX(timestamp) FROM %I.%I',
|
||||
p_db_schema, p_table_name
|
||||
) INTO _first_ts, _last_ts;
|
||||
|
||||
UPDATE meta.series
|
||||
SET first_timestamp = _first_ts, last_timestamp = _last_ts
|
||||
WHERE table_name = p_table_name AND db_schema = p_db_schema;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.drop_all_series
|
||||
CREATE OR REPLACE FUNCTION meta.drop_all_series()
|
||||
RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
_rec record;
|
||||
BEGIN
|
||||
-- Set the context to allow drops
|
||||
PERFORM set_config('application.dropping_series', 'true', true);
|
||||
|
||||
-- Loop through all series in series table
|
||||
FOR _rec IN SELECT "table_name", "db_schema" FROM meta.series
|
||||
LOOP
|
||||
-- Drop each table
|
||||
EXECUTE format('DROP TABLE IF EXISTS %I.%I', _rec."db_schema", _rec."table_name");
|
||||
END LOOP;
|
||||
|
||||
-- Clear the series table
|
||||
DELETE FROM meta.series;
|
||||
|
||||
-- Reset the context
|
||||
PERFORM set_config('application.dropping_series', 'false', true);
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.drop_series
|
||||
CREATE OR REPLACE FUNCTION meta.drop_series(_table_name text, _schema_name text DEFAULT 'series'::text, _raise_if_missing boolean DEFAULT true)
|
||||
RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
BEGIN
|
||||
-- Check if the series exists in series table
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM meta.series
|
||||
WHERE "table_name" = _table_name
|
||||
AND "db_schema" = _schema_name
|
||||
) THEN
|
||||
IF _raise_if_missing THEN
|
||||
RAISE EXCEPTION 'Series %.% does not exist', _schema_name, _table_name;
|
||||
END IF;
|
||||
RETURN;
|
||||
END IF;
|
||||
|
||||
-- Set the context to allow drops
|
||||
PERFORM set_config('application.dropping_series', 'true', true);
|
||||
|
||||
-- Drop the table
|
||||
EXECUTE format('DROP TABLE IF EXISTS %I.%I', _schema_name, _table_name);
|
||||
|
||||
-- Remove from series table
|
||||
DELETE FROM meta.series
|
||||
WHERE "table_name" = _table_name
|
||||
AND "db_schema" = _schema_name;
|
||||
|
||||
-- Reset the context
|
||||
PERFORM set_config('application.dropping_series', 'false', true);
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.drop_series_if_exists
|
||||
CREATE OR REPLACE FUNCTION meta.drop_series_if_exists(_table_name text, _schema_name text DEFAULT 'series'::text)
|
||||
RETURNS void
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
BEGIN
|
||||
-- Check if the series exists in series table
|
||||
IF EXISTS (
|
||||
SELECT 1 FROM meta.series
|
||||
WHERE "table_name" = _table_name
|
||||
AND "db_schema" = _schema_name
|
||||
) THEN
|
||||
-- Signal the prevent_direct_drop event trigger that this DROP is
|
||||
-- coming from a sanctioned procedure. Without it the trigger rejects
|
||||
-- the statement below, because the table is registered in meta.series.
|
||||
-- Local to the transaction, so it cannot leak to later statements.
|
||||
PERFORM set_config('application.dropping_series', 'true', true);
|
||||
|
||||
-- Drop the table
|
||||
EXECUTE format('DROP TABLE IF EXISTS %I.%I', _schema_name, _table_name);
|
||||
|
||||
-- Remove from series table
|
||||
DELETE FROM meta.series
|
||||
WHERE "table_name" = _table_name
|
||||
AND "db_schema" = _schema_name;
|
||||
|
||||
PERFORM set_config('application.dropping_series', 'false', true);
|
||||
END IF;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.prevent_direct_drop
|
||||
CREATE OR REPLACE FUNCTION meta.prevent_direct_drop()
|
||||
RETURNS event_trigger
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
obj record;
|
||||
BEGIN
|
||||
-- Get the current context
|
||||
IF current_setting('application.dropping_series', true) = 'true' THEN
|
||||
-- Allow the drop if it's coming from our procedures
|
||||
RETURN;
|
||||
END IF;
|
||||
|
||||
FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects()
|
||||
WHERE object_type = 'table'
|
||||
LOOP
|
||||
IF EXISTS (
|
||||
SELECT 1 FROM meta.series
|
||||
WHERE table_name = obj.object_name
|
||||
AND db_schema = obj.schema_name
|
||||
) THEN
|
||||
RAISE EXCEPTION 'Table %.% is registered in series and can only be dropped using TimeseriesDB.drop_series()',
|
||||
obj.schema_name, obj.object_name;
|
||||
END IF;
|
||||
END LOOP;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.read_series_batch
|
||||
CREATE OR REPLACE FUNCTION meta.read_series_batch(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000)
|
||||
RETURNS TABLE(result_json jsonb)
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
v_query TEXT;
|
||||
v_qualified_name TEXT;
|
||||
BEGIN
|
||||
-- Build qualified table name (with proper quoting)
|
||||
v_qualified_name := format('%I.%I', p_schema_name, p_table_name);
|
||||
|
||||
-- Build query with proper WHERE clause
|
||||
IF p_start_ts IS NULL THEN
|
||||
-- Read from beginning
|
||||
v_query := format('
|
||||
SELECT to_jsonb(t.*) AS result_json
|
||||
FROM (
|
||||
SELECT *
|
||||
FROM %s
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
) t
|
||||
', v_qualified_name, p_batch_size);
|
||||
ELSE
|
||||
-- Read from specified timestamp (inclusive)
|
||||
v_query := format('
|
||||
SELECT to_jsonb(t.*) AS result_json
|
||||
FROM (
|
||||
SELECT *
|
||||
FROM %s
|
||||
WHERE timestamp >= %L
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
) t
|
||||
', v_qualified_name, p_start_ts, p_batch_size);
|
||||
END IF;
|
||||
|
||||
-- Execute and return
|
||||
RETURN QUERY EXECUTE v_query;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.read_series_batch_cursor
|
||||
CREATE OR REPLACE FUNCTION meta.read_series_batch_cursor(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000, p_cursor_name text DEFAULT 'series_cursor'::text)
|
||||
RETURNS refcursor
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
v_query TEXT;
|
||||
v_qualified_name TEXT;
|
||||
v_cursor refcursor;
|
||||
BEGIN
|
||||
-- Build qualified table name (with proper quoting)
|
||||
v_qualified_name := format('%I.%I', p_schema_name, p_table_name);
|
||||
|
||||
-- Open cursor with name
|
||||
v_cursor := p_cursor_name;
|
||||
|
||||
-- Build query with proper WHERE clause
|
||||
IF p_start_ts IS NULL THEN
|
||||
-- Read from beginning
|
||||
v_query := format('
|
||||
SELECT *
|
||||
FROM %s
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
', v_qualified_name, p_batch_size);
|
||||
ELSE
|
||||
-- Read from specified timestamp (inclusive)
|
||||
v_query := format('
|
||||
SELECT *
|
||||
FROM %s
|
||||
WHERE timestamp >= %L
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
', v_qualified_name, p_start_ts, p_batch_size);
|
||||
END IF;
|
||||
|
||||
-- Open cursor
|
||||
OPEN v_cursor FOR EXECUTE v_query;
|
||||
|
||||
RETURN v_cursor;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
|
||||
-- Function: meta.read_series_batch_ohlc
|
||||
CREATE OR REPLACE FUNCTION meta.read_series_batch_ohlc(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000)
|
||||
RETURNS TABLE(ts timestamp without time zone, open double precision, high double precision, low double precision, close double precision, volume bigint)
|
||||
LANGUAGE plpgsql
|
||||
AS $function$
|
||||
DECLARE
|
||||
v_query TEXT;
|
||||
v_qualified_name TEXT;
|
||||
BEGIN
|
||||
-- Build qualified table name (with proper quoting)
|
||||
v_qualified_name := format('%I.%I', p_schema_name, p_table_name);
|
||||
|
||||
-- Build query with proper WHERE clause
|
||||
IF p_start_ts IS NULL THEN
|
||||
-- Read from beginning
|
||||
v_query := format('
|
||||
SELECT timestamp AS ts, open, high, low, close, volume
|
||||
FROM %s
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
', v_qualified_name, p_batch_size);
|
||||
ELSE
|
||||
-- Read from specified timestamp (inclusive)
|
||||
v_query := format('
|
||||
SELECT timestamp AS ts, open, high, low, close, volume
|
||||
FROM %s
|
||||
WHERE timestamp >= %L
|
||||
ORDER BY timestamp
|
||||
LIMIT %s
|
||||
', v_qualified_name, p_start_ts, p_batch_size);
|
||||
END IF;
|
||||
|
||||
-- Execute and return
|
||||
RETURN QUERY EXECUTE v_query;
|
||||
END;
|
||||
$function$;
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue