Simulating Row-Level TTL in YSQL

				
					-- yb_ttl_delete_bucket: delete up to p_batch_limit expired rows
-- from a single yb_hash_code slice [p_lo, p_hi].
-- Uses only HASH-key columns by default; can be overridden with p_hash_cols.

CREATE OR REPLACE FUNCTION yb_ttl_delete_bucket(
  p_schema        text,           -- e.g. 'public'
  p_table         text,           -- e.g. 'events'
  p_ts_col        text,           -- TTL column, e.g. 'expires_at'
  p_retention     interval,       -- e.g. '0 days' if p_ts_col already encodes expiry
  p_lo            int,            -- inclusive hash lower bound (0..65535)
  p_hi            int,            -- inclusive hash upper bound
  p_batch_limit   int    DEFAULT 5000,
  p_hash_cols     text[] DEFAULT NULL  -- optional override: ARRAY['tenant_id','id']
) RETURNS integer
LANGUAGE plpgsql AS $$
DECLARE
  v_sql       text;
  v_deleted   int;
  v_hash_arg  text;   -- "col1,col2,...", quoted
  v_pkdef     text;   -- PRIMARY KEY definition
  v_csv       text;   -- HASH key CSV from PK def
BEGIN
  IF p_hash_cols IS NOT NULL AND array_length(p_hash_cols, 1) IS NOT NULL THEN
    SELECT string_agg(format('%I', col), ',')
      INTO v_hash_arg
    FROM unnest(p_hash_cols) AS col;

  ELSE
    SELECT pg_get_indexdef(i.indexrelid)
      INTO v_pkdef
    FROM pg_index i
    JOIN pg_class c ON c.oid = i.indrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = p_schema
      AND c.relname  = p_table
      AND i.indisprimary;

    IF v_pkdef IS NULL THEN
      RAISE EXCEPTION 'Table %.% has no PRIMARY KEY', p_schema, p_table;
    END IF;

    -- Prefer explicit HASH key columns if present: PRIMARY KEY((... ) HASH, ...)
    SELECT (regexp_matches(v_pkdef, '\(\(([^)]*)\)\s*HASH', 'i'))[1]
      INTO v_csv;

    IF v_csv IS NOT NULL THEN
      SELECT string_agg(format('%I', btrim(s)), ',')
        INTO v_hash_arg
      FROM unnest(string_to_array(v_csv, ',')) AS s;
    ELSE
      -- Fallback: all PK cols in key order (ORDER BY is inside string_agg)
      SELECT string_agg(format('%I', a.attname), ',' ORDER BY k.ord)
        INTO v_hash_arg
      FROM pg_index i
      JOIN pg_class c ON c.oid = i.indrelid
      JOIN pg_namespace n ON n.oid = c.relnamespace
      JOIN unnest(i.indkey) WITH ORDINALITY k(attnum, ord) ON TRUE
      JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum = k.attnum
      WHERE n.nspname = p_schema
        AND c.relname  = p_table
        AND i.indisprimary;
    END IF;
  END IF;

  IF v_hash_arg IS NULL OR v_hash_arg = '' THEN
    RAISE EXCEPTION
      'yb_ttl_delete_bucket: no hash columns resolved for %.%. Provide p_hash_cols or define a PRIMARY KEY.',
      p_schema, p_table;
  END IF;

  v_sql := format($f$
    WITH doomed AS (
      SELECT ybctid
      FROM %1$I.%2$I
      WHERE %3$I < now() - $1
        AND yb_hash_code(%4$s) BETWEEN $2 AND $3
      ORDER BY ybctid
      LIMIT $4
    )
    DELETE FROM %1$I.%2$I e
    USING doomed d
    WHERE e.ybctid = d.ybctid
    RETURNING 1
  $f$, p_schema, p_table, p_ts_col, v_hash_arg);

  EXECUTE v_sql USING p_retention, p_lo, p_hi, p_batch_limit
    INTO v_deleted;

  RETURN COALESCE(v_deleted, 0);
END $$;