Row-Based TTL in YSQL (When Your Table Has No Timestamp)

				
					CREATE OR REPLACE FUNCTION yb_ttl_delete_bucket_sidecar(
  p_schema_base   text,
  p_table_base    text,
  p_schema_ttl    text,
  p_table_ttl     text,
  p_retention     interval,        -- usually '0 days'
  p_lo            int,             -- inclusive 0..65535
  p_hi            int,             -- inclusive
  p_batch_limit   int DEFAULT 5000,
  p_hash_cols     text[] DEFAULT NULL   -- optional override of hash cols
) RETURNS integer
LANGUAGE plpgsql AS $$
DECLARE
  v_sql     text;
  v_deleted int;
  v_pkdef   text;
  v_csv     text;
  v_hasharg text;   -- "b.col1,b.col2,..."
  v_pklist  text;   -- "pk1,pk2,..."
BEGIN
  -- Resolve hash columns: explicit override wins, else derive from PK (prefer HASH key)
  IF p_hash_cols IS NOT NULL AND array_length(p_hash_cols,1) IS NOT NULL THEN
    SELECT string_agg(format('b.%I', col), ',') INTO v_hasharg
    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_base
      AND c.relname = p_table_base
      AND i.indisprimary;

    IF v_pkdef IS NULL THEN
      RAISE EXCEPTION '%.% has no PRIMARY KEY', p_schema_base, p_table_base;
    END IF;

    -- Try explicit HASH key first
    SELECT (regexp_matches(v_pkdef, '\(\(([^)]*)\)\s*HASH', 'i'))[1] INTO v_csv;

    IF v_csv IS NOT NULL THEN
      SELECT string_agg(format('b.%I', btrim(s)), ',') INTO v_hasharg
      FROM unnest(string_to_array(v_csv, ',')) AS s;
    ELSE
      -- Fallback: all PK cols in key order (ORDER BY inside string_agg)
      SELECT string_agg(format('b.%I', a.attname), ',' ORDER BY k.ord) INTO v_hasharg
      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_base
        AND c.relname = p_table_base
        AND i.indisprimary;
    END IF;
  END IF;

  IF v_hasharg IS NULL OR v_hasharg = '' THEN
    RAISE EXCEPTION 'No hash columns resolved for %.%. Provide p_hash_cols or define a PRIMARY KEY.',
      p_schema_base, p_table_base;
  END IF;

  -- Build USING(pk1,pk2,...) for the equality join (ORDER BY inside string_agg)
  SELECT string_agg(format('%I', a.attname), ',' ORDER BY k.ord) INTO v_pklist
  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_base
    AND c.relname = p_table_base
    AND i.indisprimary;

  v_sql := format($f$
    WITH doomed AS (
      SELECT b.ybctid
      FROM %1$I.%2$I b
      JOIN %3$I.%4$I t USING (%5$s)          -- USING(pk1, pk2, ...)
      WHERE t.expires_at < now() - $1
        AND yb_hash_code(%6$s) BETWEEN $2 AND $3
      ORDER BY b.ybctid
      LIMIT $4
    )
    DELETE FROM %1$I.%2$I b
    USING doomed d
    WHERE b.ybctid = d.ybctid
    RETURNING 1
  $f$, p_schema_base, p_table_base, p_schema_ttl, p_table_ttl, v_pklist, v_hasharg);

  EXECUTE v_sql USING p_retention, p_lo, p_hi, p_batch_limit INTO v_deleted;
  RETURN COALESCE(v_deleted, 0);
END $$;