CREATE TABLE daily_metrics (
id BIGGENERATED ALWAYS AS IDENTITY PRIMARY KEY,
tenant_id BIGINT NOT NULL,
metric TEXT NOT NULL,
day DATE NOT NULL,
count BIGINT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- The conflict target for every upsert below.
CREATE UNIQUE INDEX daily_metrics_unique_idx
ON daily_metrics (tenant_id, metric, day);
-- Supports range queries over a tenant's activity.
CREATE INDEX daily_metrics_tenant_day_idx
ON daily_metrics (tenant_id, day);INSERT INTO daily_metrics (tenant_id, metric, day, count, updated_at)
VALUES (:tenant_id, :metric, :day, :delta, now())
ON CONFLICT (tenant_id, metric, day)
DO UPDATE
SET count = daily_metrics.count + EXCLUDED.count,
updated_at = now()
WHERE EXCLUDED.count <> 0
RETURNING count AS total;-- Merge a pre-aggregated buffer of events in a single round trip.
-- Each row must be unique per key; aggregate upstream if events repeat.
INSERT INTO daily_metrics (tenant_id, metric, day, count, updated_at)
VALUES
(:t1, :m1, :d1, :c1, now()),
(:t2, :m2, :d2, :c2, now()),
(:t3, :m3, :d3, :c3, now())
ON CONFLICT (tenant_id, metric, day)
DO UPDATE
SET count = daily_metrics.count + EXCLUDED.count,
updated_at = now()
RETURNING tenant_id, metric, day, count AS total;Maintaining running counters — page views, per-tenant usage, daily event tallies — is deceptively hard under concurrency. The naive SELECT then UPDATE-or-INSERT flow has a race: two sessions can both miss the row, both insert, and one hits a duplicate-key error or clobbers the other's increment. Postgres solves this cleanly with INSERT ... ON CONFLICT DO UPDATE, an atomic upsert that either creates the counter row or increments the existing one in a single statement, with the whole thing serialized by the unique index.
The counters schema tab defines the shape the upsert depends on. daily_metrics stores one row per (tenant_id, metric, day), and the crucial piece is the UNIQUE constraint daily_metrics_unique_idx over those three columns. ON CONFLICT can only fire against a real unique or exclusion constraint, so the index is not just an optimization — it is what makes the conflict target valid. A plain count column defaults to zero, and updated_at tracks the last touch.
The increment upsert tab shows the core statement. On insert it seeds the row with the incoming :delta; on conflict it runs SET count = daily_metrics.count + EXCLUDED.count. The EXCLUDED pseudo-row holds the values that would have been inserted, so referencing EXCLUDED.count reuses the same parameter without binding it twice. Qualifying daily_metrics.count on the left disambiguates the existing row from EXCLUDED. The WHERE in the conflict clause makes the update conditional — it skips a redundant write when the delta is zero, avoiding index churn and dead tuples.
Because the statement is atomic, no explicit locking or retry loop is needed; concurrent increments queue on the row lock and each applies its delta exactly once. The RETURNING clause hands back the post-update total so callers avoid a follow-up read.
The batch merge tab scales this to bulk ingestion: a VALUES list pre-aggregated per key merges in one round trip, which matters when flushing a buffer of events. One pitfall worth noting — ON CONFLICT targets a single arbiter constraint, so overlapping unique indexes require care, and upserts still generate dead tuples under heavy churn, making autovacuum tuning relevant at scale.
Related snips
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
