Postgresql – Optimize aggregate query on massive table to refresh materialized view

Parallel plan

If you first collect the store_record statistics and then join that with the store_database, you’ll get a better, parallelisable plan.

EXPLAIN (analyze, buffers, verbose)
SELECT
    store_database.id          as database_id,
    s.latest_record_updated_at as latest_record_updated_at,
    coalesce(s.record_count,0) as record_count
FROM store_database
    LEFT JOIN 
    (   SELECT
            store_record.database_id     as database_id,
            MAX(store_record.updated_at) as latest_record_updated_at,
            COUNT(store_record.id)       as record_count
        FROM store_record
        GROUP BY store_record.database_id)
    AS s ON (store_database.id = s.database_id);

Here’s a demo – at the end you can see both queries return the exact same results, but the one I suggest runs faster and has a more flexible plan. The number of workers dispatched depends on your max_worker_processes, max_parallel_workers, max_parallel_workers_per_gather settings as well as some additional logic inside the planner.

With more rows in store_record the difference will be more pronounced. On my system with 40 million test rows it went down from 14 seconds to 3 seconds with one worker, 1.4 seconds when it caps out dispatching six workers out of 16 available.

Caching

I’m thinking of caching these values on every write to the Task table. Either in Redis or in PostgreSQL itself.

If it’s an option, it’s worth a try – you can maintain proper accuracy and instantly available statistics at the cost of some (deferrable) table throughput overhead. You can replace your materialized view with a regular table or add the statistics columns to store_database

create table store_record_statistics(
  database_id smallint unique references store_database(id) 
    on update cascade,
  latest_record_updated_at timestamptz,
  record_count integer default 0);

insert into store_record_statistics --initializes table with view definition
SELECT g.id, MAX(s.updated_at), COUNT(*)
FROM store_database g LEFT JOIN store_record s ON g.id = s.database_id
GROUP BY g.id;

create index store_record_statistics_idx 
  on store_record_statistics (database_id) 
  include (latest_record_updated_at,record_count);

cluster verbose store_record_statistics using store_record_statistics_idx;

And leave keeping the table up to date to a trigger that fires each time store_record changes.

CREATE FUNCTION maintain_store_record_statistics_trigger() 
  RETURNS TRIGGER LANGUAGE plpgsql AS
$$ BEGIN
  IF TG_OP IN ('UPDATE', 'DELETE') THEN --decrement and find second most recent updated_at
    UPDATE store_record_statistics srs
    SET (record_count,
         latest_record_updated_at) 
    =   (record_count - 1,
         (SELECT s.updated_at
          FROM   store_record s
          WHERE  s.database_id = srs.database_id
          ORDER  BY s.updated_at DESC NULLS LAST
          LIMIT  1))
    WHERE database_id = old.database_id;
  END IF;
  IF TG_OP in ('INSERT','UPDATE') THEN --increment and pick most recent updated_at
    UPDATE store_record_statistics 
    SET (record_count,
         latest_record_updated_at) 
    =   (record_count + 1,
         greatest(
            latest_record_updated_at,
            new.updated_at))
    WHERE database_id=new.database_id;
  END IF;
  RETURN NULL;
END $$;

Making the trigger deferrable decouples its execution time from the main operation but it’ll still infer its costs at the end of the transaction.

CREATE CONSTRAINT TRIGGER maintain_store_record_statistics
AFTER INSERT OR UPDATE OF database_id OR DELETE ON store_record 
INITIALLY DEFERRED FOR EACH ROW
EXECUTE PROCEDURE maintain_store_record_statistics_trigger();

TRUNCATE trigger cannot be declared FOR EACH ROW with the rest of events, so it has to be defined separately

CREATE FUNCTION maintain_store_record_statistics_truncate_trigger() 
RETURNS TRIGGER LANGUAGE plpgsql AS
$$ BEGIN
  update store_record_statistics 
  set (record_count, latest_record_updated_at)
    = (0           , null);--wipes/resets all stats
  RETURN NULL;
END $$;

CREATE TRIGGER maintain_store_record_statistics_truncate
AFTER TRUNCATE ON store_record
EXECUTE PROCEDURE maintain_store_record_statistics_truncate_trigger();

In my test, an update or delete of 10000 random rows in a 100-million-row table run in seconds. A single insert of 1000 new, randomly generated rows took 25ms without and 200ms with the trigger. A million was 30s and 3 minutes correspondingly.

A demo.

Partitioning-backed parallel plan

store_record might be a good fit for partitioning:

create table store_record(
  id serial not null,
  updated_at timestamptz default now(),
  database_id smallint references store_database(id)
  ) partition by range (database_id);

DO $$
declare 
  vi_database_max_id           smallint:=0;
  vi_database_id               smallint:=0;
  vi_database_id_per_partition smallint:=40;--tweak for lower/higher granularity
begin
  select max(id) from store_database into vi_database_max_id;
  for vi_database_id in 1 .. vi_database_max_id by vi_database_id_per_partition loop
    execute format ('
      drop table if exists store_record_p_%1$s;
      create table store_record_p_%1$s
      partition of store_record
      for VALUES from (%1$s) to (%1$s + %2$s)
      with (parallel_workers=16);
      ', vi_database_id, vi_database_id_per_partition);
  end loop;
end $$ ;

Splitting objects in this manner lets the planner split their scans accordingly, which works best with parallel workers, but doesn’t require them. Even your initial, unaltered query behind the view is able to take advantage of this structure:

HashAggregate  (cost=60014.27..60041.47 rows=2720 width=18) (actual time=910.657..910.698 rows=169 loops=1)
  Output: store_database.id, max(store_record_p_1.updated_at), count(store_record_p_1.id)
  Group Key: store_database.id
  Buffers: shared hit=827 read=9367 dirtied=5099 written=4145
  ->  Hash Right Join  (cost=71.20..45168.91 rows=1979382 width=14) (actual time=0.064..663.603 rows=1600020 loops=1)
        Output: store_database.id, store_record_p_1.updated_at, store_record_p_1.id
        Inner Unique: true
        Hash Cond: (store_record_p_1.database_id = store_database.id)
        Buffers: shared hit=827 read=9367 dirtied=5099 written=4145
        ->  Append  (cost=0.00..39893.73 rows=1979382 width=14) (actual time=0.014..390.152 rows=1600000 loops=1)
              Buffers: shared hit=826 read=9367 dirtied=5099 written=4145
              ->  Seq Scan on public.store_record_p_1  (cost=0.00..8035.02 rows=530202 width=14) (actual time=0.014..77.130 rows=429068 loops=1)
                    Output: store_record_p_1.updated_at, store_record_p_1.id, store_record_p_1.database_id
                    Buffers: shared read=2733 dirtied=1367 written=1335
              ->  Seq Scan on public.store_record_p_41  (cost=0.00..8067.36 rows=532336 width=14) (actual time=0.017..75.193 rows=430684 loops=1)
                    Output: store_record_p_41.updated_at, store_record_p_41.id, store_record_p_41.database_id
                    Buffers: shared read=2744 dirtied=1373 written=1341
              ->  Seq Scan on public.store_record_p_81  (cost=0.00..8029.14 rows=529814 width=14) (actual time=0.017..74.583 rows=428682 loops=1)
                    Output: store_record_p_81.updated_at, store_record_p_81.id, store_record_p_81.database_id
                    Buffers: shared read=2731 dirtied=1366 written=1334
              ->  Seq Scan on public.store_record_p_121  (cost=0.00..5835.90 rows=385090 width=14) (actual time=0.016..45.407 rows=311566 loops=1)
                    Output: store_record_p_121.updated_at, store_record_p_121.id, store_record_p_121.database_id
                    Buffers: shared hit=826 read=1159 dirtied=993 written=135
              ->  Seq Scan on public.store_record_p_161  (cost=0.00..29.40 rows=1940 width=14) (actual time=0.008..0.008 rows=0 loops=1)
                    Output: store_record_p_161.updated_at, store_record_p_161.id, store_record_p_161.database_id
        ->  Hash  (cost=37.20..37.20 rows=2720 width=2) (actual time=0.041..0.042 rows=169 loops=1)
              Output: store_database.id
              Buckets: 4096  Batches: 1  Memory Usage: 38kB
              Buffers: shared hit=1
              ->  Seq Scan on public.store_database  (cost=0.00..37.20 rows=2720 width=2) (actual time=0.012..0.021 rows=169 loops=1)
                    Output: store_database.id
                    Buffers: shared hit=1
Planning Time: 0.292 ms
Execution Time: 910.811 ms

Demo. It’s best to test what granularity gives the best performance on your setup. You can even test sub-partioning, giving each store_record.database_id a partition, that is then sub-partitioned into date ranges, simplifying access to most recent entries.