Postgres 15

By using db<>fiddle, you agree to license everything you submit by Creative Commons CC0.
-- table for storing data processing status

CREATE TABLE status_table (
  base_name  text      NOT NULL
, version    smallint  NOT NULL
, ref_time   int       NOT NULL
, processed  bool      NOT NULL
, processing bool      NOT NULL
, updated    int       NOT NULL
, PRIMARY KEY (base_name, version)
);
 
-- explanation:
-- a batch of data is referenced by 'ref_time'
-- each batch of data comes as a number of files, each referenced by 'base_name'
-- data files don't appear all at once, but in dribs and drabs
-- each data file is processed as it appears, extracting data into data_table (see below)
-- the 'updated' timestamp (in mins) is updated each time the 'processed' status changes
-- we only want to use data from a batch that has been completely processed
-- i.e. where all 'processed' flags are set to true

-- (extract(epoch from now()) / 60) is current epoch time in mins
INSERT INTO status_table VALUES 
-- --------- new ref_time batch 27793000 ----------
-- all processed
  ('abc', 1, 27793000, 't', 'f', (extract(epoch from now()) / 60) - 500)
, ('jkl', 1, 27793000, 't', 'f', (extract(epoch from now()) / 60) - 450)
, ('pqr', 1, 27793000, 't', 'f', (extract(epoch from now()) / 60) - 400)
-- --------- new ref_time batch 27793300 ----------
-- all processed
, ('def', 1, 27793300, 't', 'f', (extract(epoch from now()) / 60) - 350)
, ('hft', 1, 27793300, 't', 'f', (extract(epoch from now()) / 60) - 300)
-- --------- new ref_time batch 27793600 ----------
-- only some processed
, ('ghi', 1, 27793600, 'f', 't', (extract(epoch from now()) / 60) - 275)
, ('wqt', 1, 27793600, 't', 'f', (extract(epoch from now()) / 60) - 250)
-- --------- new ref_time batch 27793900 ----------
, ('rfe', 1, 27793900, 't', 'f', (extract(epoch from now()) / 60) - 225)
-- any 'updated' time above here (threshold is -200) counts as 'settled' (if all in ref_time batch are settled)
, ('mno', 1, 27793900, 'f', 't', (extract(epoch from now()) / 60) - 175)
, ('dpw', 1, 27793900, 't', 'f', (extract(epoch from now()) / 60) - 150)
-- --------- new ref_time batch 27794200 ----------
-- all processed but not settled (more to process could be added)
, ('eqz', 1, 27794200, 't', 'f', (extract(epoch from now()) / 60) - 125)
, ('fwp', 1, 27794200, 't', 'f', (extract(epoch from now()) / 60) - 100)

CREATE TABLE



INSERT 0 12


-- the actual data table

CREATE TABLE data_table (
  location  text      NOT NULL
, param_id  text      NOT NULL
, ref_time  int       NOT NULL
, fcst_time smallint  NOT NULL
, timestamp int       NOT NULL
, value     text      NOT NULL
, PRIMARY KEY (location, param_id, ref_time, fcst_time)
);

-- note that: 'ref_time' + 'fcst_time' = 'timestamp'
-- I know it's not ideal to store 'timestamp' when it can be derived from 'ref_time' and 'fcst_time' (could just store those two)
-- but all three bits of info are in the raw data so I store all three rather than deriving 'timestamp' after extraction
-- maybe I could/should reduce table storage by storing only 'ref_time' and 'fcst_time' and deriving 'timestamp'?

INSERT INTO data_table VALUES
-- --------- new ref_time batch 27793000 ----------
  ('aaa', 'temp', 27793000,   0, 27793000, '21.0')
, ('aaa', 'temp', 27793000, 100, 27793100, '20.1')
, ('aaa', 'temp', 27793000, 200, 27793200, '19.6')
, ('aaa', 'temp', 27793000, 300, 27793300, '18.3')
, ('aaa', 'temp', 27793000, 400, 27793400, '17.1')
, ('aaa', 'temp', 27793000, 500, 27793500, '16.4')
, ('aaa', 'temp', 27793000, 600, 27793600, '16.4')
, ('aaa', 'temp', 27793000, 700, 27793700, '16.4')
, ('aaa', 'temp', 27793000, 800, 27793800, '16.4')
, ('aaa', 'temp', 27793000, 900, 27793900, '16.4')
-- new param in same ref_time batch
-- note that the timestamps overlap with previous batch
, ('aaa', 'humi', 27793000,   0, 27793000, '55.0')
, ('aaa', 'humi', 27793000, 100, 27793100, '57.1')
, ('aaa', 'humi', 27793000, 200, 27793200, '62.6')
, ('aaa', 'humi', 27793000, 300, 27793300, '76.3')
, ('aaa', 'humi', 27793000, 400, 27793400, '72.1')
, ('aaa', 'humi', 27793000, 500, 27793500, '65.4')
, ('aaa', 'humi', 27793000, 600, 27793600, '63.4')
, ('aaa', 'humi', 27793000, 700, 27793700, '62.2')
, ('aaa', 'humi', 27793000, 800, 27793800, '59.1')
, ('aaa', 'humi', 27793000, 900, 27793900, '57.8')
-- --------- new ref_time batch 27793300 ----------
-- note that the timestamps overlap with previous batch
, ('aaa', 'temp', 27793300,   0, 27793300, '18.0')
, ('aaa', 'temp', 27793300, 100, 27793400, '17.1')
, ('aaa', 'temp', 27793300, 200, 27793500, '16.6')
, ('aaa', 'temp', 27793300, 300, 27793600, '15.3')
, ('aaa', 'temp', 27793300, 400, 27793700, '13.1')
, ('aaa', 'temp', 27793300, 500, 27793800, '11.4')
-- new param in same ref_time batch
, ('aaa', 'humi', 27793300,   0, 27793300, '76.0')
, ('aaa', 'humi', 27793300, 100, 27793400, '72.1')
, ('aaa', 'humi', 27793300, 200, 27793500, '65.6')
, ('aaa', 'humi', 27793300, 300, 27793600, '63.3')
, ('aaa', 'humi', 27793300, 400, 27793700, '61.1')
, ('aaa', 'humi', 27793300, 500, 27793800, '59.4')
-- --------- new ref_time batch 27793600 ----------
-- note that the timestamps overlap with previous batch
, ('aaa', 'temp', 27793600,   0, 27793600, '18.0')
, ('aaa', 'temp', 27793600, 100, 27793700, '17.1')
, ('aaa', 'temp', 27793600, 200, 27793800, '16.6')
, ('aaa', 'temp', 27793600, 300, 27793900, '15.3')
, ('aaa', 'temp', 27793600, 400, 27794000, '13.1')
, ('aaa', 'temp', 27793600, 500, 27794100, '11.4')
-- --------- new ref_time batch 27793900 ----------
, ('aaa', 'temp', 27793900,   0, 27793900, '18.0')
, ('aaa', 'temp', 27793900, 100, 27794000, '17.1')
, ('aaa', 'temp', 27793900, 200, 27794100, '16.6')
, ('aaa', 'temp', 27793900, 300, 27794200, '15.3')
, ('aaa', 'temp', 27793900, 400, 27794300, '13.1')
, ('aaa', 'temp', 27793900, 500, 27794400, '11.4')
-- --------- new ref_time batch 27794200 ----------
, ('aaa', 'temp', 27794200,   0, 27794200, '15.2')
, ('aaa', 'temp', 27794200, 100, 27794300, '13.5')
, ('aaa', 'temp', 27794200, 200, 27794400, '12.9')
, ('aaa', 'temp', 27794200, 300, 27794500, '11.3')
, ('aaa', 'temp', 27794200, 400, 27794600, '10.1')
, ('aaa', 'temp', 27794200, 500, 27794700,  '9.4')

CREATE TABLE



INSERT 0 50


WITH
  stats AS (
    SELECT ref_time
      , max(updated) < (round(extract(epoch from now()) / 60) - 200) AS settled
      , (count(*) FILTER (WHERE processed) = count(*)) AND (max(updated) < (round(extract(epoch from now()) / 60) - 200)) AS ready
    FROM status_table
    GROUP BY ref_time
  ),
  min_ts AS (
    SELECT ref_time FROM stats WHERE ready ORDER BY ref_time DESC LIMIT 1
  ),
  sel1 AS (
    -- records that would be selected by an actual data lookup (use same logic)... we need to keep these (don't delete)
    SELECT DISTINCT ON (d.location, d.timestamp, d.param_id)
      d.location, d.param_id, d.ref_time, d.fcst_time
    FROM data_table AS d
    INNER JOIN stats s USING (ref_time)
    WHERE s.ready AND d.timestamp >= (SELECT ref_time FROM min_ts)
    ORDER BY d.location, d.timestamp, d.param_id, d.ref_time DESC
  ),
  sel2 AS (
    -- also keep all records that are in-progress (not 'settled')
    SELECT
      d.location, d.param_id, d.ref_time, d.fcst_time
    FROM data_table AS d
    INNER JOIN stats AS s USING (ref_time)
    WHERE NOT s.settled
  )
DELETE FROM data_table 
  WHERE
    (location, param_id, ref_time, fcst_time) NOT IN (SELECT location, param_id, ref_time, fcst_time FROM sel1)
  AND
    (location, param_id, ref_time, fcst_time) NOT IN (SELECT location, param_id, ref_time, fcst_time FROM sel2);

DELETE 24