不重新摄入历史数据，规范化油井和管线遥测宽表

CREATE TABLE sensor_readings (
    recorded_at TIMESTAMPTZ NOT NULL,
    tag_id TEXT NOT NULL, -- tag path, e.g. 'Pad07/Well3/TubingPressure'
    device_id TEXT NOT NULL,
    site TEXT NOT NULL,
    line TEXT NOT NULL,
    firmware_ver TEXT NOT NULL,
    unit TEXT NOT NULL,
    value DOUBLE PRECISION NOT NULL
) WITH (tsdb.hypertable, tsdb.partition_column = 'recorded_at');

CREATE INDEX ON sensor_readings
    (tag_id, device_id, site, line, firmware_ver, recorded_at DESC);

CREATE TABLE tag_metadata (
    tag_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tag_name TEXT NOT NULL UNIQUE, -- 'Pad07/Well3/TubingPressure'
    device_id TEXT NOT NULL,
    site TEXT NOT NULL,
    line TEXT NOT NULL,
    firmware_ver TEXT NOT NULL,
    unit TEXT NOT NULL,
    ts_start TIMESTAMPTZ,
    ts_end TIMESTAMPTZ,
    ts_last_seen TIMESTAMPTZ
);
CREATE TABLE sensor_readings_narrow (
    recorded_at TIMESTAMPTZ NOT NULL,
    tag_id BIGINT NOT NULL,
    value DOUBLE PRECISION NOT NULL
) WITH (
    tsdb.hypertable,
    tsdb.partition_column = 'recorded_at',
    tsdb.create_default_indexes = false,
    tsdb.segmentby = 'tag_id'
);

SELECT alter_job(job_id, scheduled => false) FROM timescaledb_information.jobs WHERE proc_name = 'policy_compression' AND hypertable_name = 'sensor_readings_narrow';

SELECT tag_id, COUNT(*) FROM ( SELECT DISTINCT tag_id, device_id, site, line, firmware_ver, unit FROM sensor_readings ) d GROUP BY tag_id HAVING COUNT(*) > 1;

INSERT INTO tag_metadata (tag_name, device_id, site, line, firmware_ver, unit) SELECT DISTINCT s.tag_id, s.device_id, s.site, s.line, s.firmware_ver, s.unit FROM sensor_readings s WHERE NOT EXISTS ( SELECT 1 FROM tag_metadata m WHERE m.tag_name = s.tag_id );

CREATE TABLE migration_slices ( slice_start TIMESTAMPTZ PRIMARY KEY, slice_end TIMESTAMPTZ NOT NULL, completed_at TIMESTAMPTZ NOT NULL DEFAULT now(), row_count BIGINT NOT NULL );

BEGIN; WITH moved AS ( INSERT INTO sensor_readings_narrow (recorded_at, tag_id, value) SELECT r.recorded_at, m.tag_id, r.value FROM sensor_readings r JOIN tag_metadata m ON m.tag_name = r.tag_id WHERE r.recorded_at >= TIMESTAMPTZ '2024-01-01' AND r.recorded_at < TIMESTAMPTZ '2024-01-08' RETURNING 1 ) INSERT INTO migration_slices (slice_start, slice_end, row_count) SELECT TIMESTAMPTZ '2024-01-01', TIMESTAMPTZ '2024-01-08', count(*) FROM moved; COMMIT;

SELECT g.slice_start, g.slice_start + INTERVAL '7 days' AS slice_end FROM generate_series(TIMESTAMPTZ '2024-01-01', now(), INTERVAL '7 days') AS g(slice_start) WHERE NOT EXISTS ( SELECT 1 FROM migration_slices s WHERE s.slice_start = g.slice_start ) ORDER BY g.slice_start;

CREATE INDEX sensor_readings_narrow_tag_time_idx ON sensor_readings_narrow (tag_id, recorded_at DESC) WITH (timescaledb.transaction_per_chunk);

SELECT indexrelid::regclass FROM pg_index WHERE indisvalid IS FALSE;

BEGIN; SET LOCAL lock_timeout = '5s'; ALTER TABLE sensor_readings RENAME TO sensor_readings_old; CREATE VIEW sensor_readings AS SELECT r.recorded_at, m.tag_name AS tag_id, m.device_id, m.site, m.line, m.firmware_ver, m.unit, r.value FROM sensor_readings_narrow r JOIN tag_metadata m USING (tag_id); COMMIT;

-- Continuous aggregates reading the wide table
SELECT view_name, materialization_hypertable_name
FROM timescaledb_information.continuous_aggregates
WHERE hypertable_name = 'sensor_readings';

-- Retention and columnstore policies bound to it. Refresh policies list
-- under the aggregate's materialization hypertable, found above.
SELECT job_id, application_name, proc_name, config
FROM timescaledb_information.jobs
WHERE hypertable_name = 'sensor_readings';

-- Triggers and grants
SELECT tgname FROM pg_trigger
WHERE tgrelid = 'sensor_readings'::regclass AND NOT tgisinternal;
SELECT grantee, privilege_type FROM information_schema.role_table_grants
WHERE table_name = 'sensor_readings';

SELECT s.slice_start, s.row_count AS moved, (SELECT count(*) FROM sensor_readings_old o WHERE o.recorded_at >= s.slice_start AND o.recorded_at < s.slice_end) AS original FROM migration_slices s WHERE s.row_count <> (SELECT count(*) FROM sensor_readings_old o WHERE o.recorded_at >= s.slice_start AND o.recorded_at < s.slice_end) ORDER BY s.slice_start;

WITH old_side AS ( SELECT recorded_at::date AS day, count(*) AS n, min(recorded_at) AS min_ts, max(recorded_at) AS max_ts, sum(value) AS sum_value FROM sensor_readings_old WHERE recorded_at >= TIMESTAMPTZ '2024-01-01' AND recorded_at < TIMESTAMPTZ '2024-01-08' GROUP BY 1 ), new_side AS ( SELECT recorded_at::date AS day, count(*) AS n, min(recorded_at) AS min_ts, max(recorded_at) AS max_ts, sum(value) AS sum_value FROM sensor_readings_narrow WHERE recorded_at >= TIMESTAMPTZ '2024-01-01' AND recorded_at < TIMESTAMPTZ '2024-01-08' GROUP BY 1 ) SELECT coalesce(o.day, n.day) AS day, o.n AS old_n, n.n AS new_n FROM old_side o FULL OUTER JOIN new_side n USING (day) WHERE o.n IS DISTINCT FROM n.n OR o.min_ts IS DISTINCT FROM n.min_ts OR o.max_ts IS DISTINCT FROM n.max_ts OR abs(o.sum_value - n.sum_value) > 1e-9 * greatest(abs(o.sum_value), abs(n.sum_value), 1);

CREATE TABLE sensor_readings_narrow (
    recorded_at TIMESTAMPTZ NOT NULL,
    tag_id BIGINT NOT NULL,
    value DOUBLE PRECISION, -- measurements
    value_int BIGINT, -- booleans as 0/1, state and alarm codes
    value_text TEXT, -- mode labels, free-text states
    CHECK (num_nonnulls(value, value_int, value_text) = 1)
) WITH (
    tsdb.hypertable,
    tsdb.partition_column = 'recorded_at',
    tsdb.create_default_indexes = false,
    tsdb.segmentby = 'tag_id'
);