[+] add admin.drop_source_partitions, closes #1340 (#1364)

* [+] implement admin.drop_source_partitions for proactive decommission cleanup, fixes #1340

* align drop_source_partitions with drop_old_time_partitions pattern

- Use oid::regclass for schema-qualified identifiers instead of relname
  with hardcoded schema prefixes.
- Remove unnecessary CASCADE from DROP TABLE after DETACH.
- Check parent table comment (pgwatch-generated-metric-lvl) instead of
  requiring a separate comment on each child partition.
- Trim verbose inline comments.
- Move to admin_functions.sql

---------

Co-authored-by: Pavlo Golub <[email protected]>

Author: RafayKhattak

internal/sinks/sql/admin_functions.sql

@@ -167,6 +167,53 @@ BEGIN
 END;
 $SQL$ LANGUAGE plpgsql;
 
+/*
+drop_source_partitions removes all metric data for a decommissioned source
+    p_source_name - the dbname (source) whose partitions should be dropped
+Returns the number of patritions dropped
+*/
+CREATE OR REPLACE FUNCTION admin.drop_source_partitions(p_source_name text)
+RETURNS int AS
+$SQL$
+DECLARE
+  r record;
+  v_dropped_count int := 0;
+BEGIN
+  FOR r IN
+    SELECT
+      c.oid::regclass::text      AS partition_name,
+      parent.oid::regclass::text AS metric_table,
+      parent.relname             AS metric_name
+    FROM pg_catalog.pg_class c
+    JOIN pg_catalog.pg_inherits i      ON c.oid = i.inhrelid
+    JOIN pg_catalog.pg_class parent    ON parent.oid = i.inhparent
+    WHERE c.relnamespace = 'subpartitions'::regnamespace
+      AND pg_catalog.obj_description(parent.oid, 'pg_class') = 'pgwatch-generated-metric-lvl'
+      AND (regexp_match(
+             pg_catalog.pg_get_expr(c.relpartbound, c.oid),
+             E'FOR VALUES IN \\(''(.+?)''\\)'
+           ))[1] = p_source_name
+  LOOP
+    -- advisory lock matching ensure_partition_metric_dbname_time
+    PERFORM pg_advisory_xact_lock(
+      regexp_replace(md5(r.metric_name), E'\\D', '', 'g')::varchar(10)::int8
+    );
+
+    RAISE NOTICE 'detaching partition: % from %', r.partition_name, r.metric_table;
+    EXECUTE 'ALTER TABLE ' || r.metric_table || ' DETACH PARTITION ' || r.partition_name;
+
+    RAISE NOTICE 'dropping partition:  %', r.partition_name;
+    EXECUTE 'DROP TABLE IF EXISTS ' || r.partition_name;
+
+    v_dropped_count := v_dropped_count + 1;
+  END LOOP;
+
+  DELETE FROM admin.all_distinct_dbname_metrics WHERE dbname = p_source_name;
+
+  RETURN v_dropped_count;
+END;
+$SQL$ LANGUAGE plpgsql;
+
 /*
 maintain_unique_sources() maintains a mapping of unique sources in each metric table 
 in admin.all_distinct_dbname_metrics. This is used to avoid listing the same source