\echo Use "ALTER EXTENSION pg_sift UPDATE TO '4.2'" to load this file. \quit

-- Per-table options of a sift_table (docs/COL-TAM.md section 8). A table access method has no reloptions before PostgreSQL 20, so the settings that act on one
-- table (sift.table_*) can be overridden for it with sift.table_set_option; the rows are in sift.table_option_values, keyed by the table.

-- The key of a table: its OID, written (by pg_dump, by COPY) as its qualified name and read back as the OID the table has in the database that reads it. A name
-- with no table reads as 0, and the row of 0 is skipped (the trigger below), so a restore that leaves a table out does not fail on its options. Binary compatible
-- with oid and regclass.
CREATE TYPE sift.tblref;

CREATE FUNCTION sift.tblref_in(cstring) RETURNS sift.tblref
	AS 'MODULE_PATHNAME', 'sift_tblref_in' LANGUAGE C STABLE STRICT;

CREATE FUNCTION sift.tblref_out(sift.tblref) RETURNS cstring
	AS 'MODULE_PATHNAME', 'sift_tblref_out' LANGUAGE C STABLE STRICT;

CREATE TYPE sift.tblref (INPUT = sift.tblref_in, OUTPUT = sift.tblref_out, INTERNALLENGTH = 4, PASSEDBYVALUE, ALIGNMENT = int4);

CREATE CAST (sift.tblref AS oid) WITHOUT FUNCTION AS IMPLICIT;
CREATE CAST (sift.tblref AS regclass) WITHOUT FUNCTION;
CREATE CAST (regclass AS sift.tblref) WITHOUT FUNCTION AS ASSIGNMENT;

CREATE TABLE sift.table_option_values (rel sift.tblref NOT NULL, name text NOT NULL, value text NOT NULL);
CREATE UNIQUE INDEX table_option_values_key ON sift.table_option_values (rel oid_ops, name);
-- dumped with the database; a row of a table that is gone (a stale OID prints as a number) is never written
SELECT pg_catalog.pg_extension_config_dump('sift.table_option_values', 'WHERE rel::pg_catalog.oid IN (SELECT oid FROM pg_catalog.pg_class)');

-- Skips the row of a table the database does not have, and replaces the entry of the table in the cache of every backend
CREATE FUNCTION sift.table_option_values_trg() RETURNS trigger
	AS 'MODULE_PATHNAME', 'sift_table_option_values_trg' LANGUAGE C;
CREATE TRIGGER table_option_values_trg BEFORE INSERT OR UPDATE OR DELETE ON sift.table_option_values
	FOR EACH ROW EXECUTE FUNCTION sift.table_option_values_trg();

-- The options go with the table.
CREATE FUNCTION sift.table_option_values_drop() RETURNS event_trigger
	LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, pg_temp AS $fn$
BEGIN
	DELETE FROM sift.table_option_values v USING pg_event_trigger_dropped_objects() d
		WHERE d.classid = 'pg_class'::regclass AND d.objsubid = 0 AND v.rel::oid = d.objid;
END
$fn$;

CREATE EVENT TRIGGER sift_table_option_values_drop ON sql_drop EXECUTE FUNCTION sift.table_option_values_drop();

-- The names of the per-table options: the sift.table_* settings without the prefix
CREATE FUNCTION sift._table_option_names() RETURNS text[]
	AS 'MODULE_PATHNAME', 'sift_table_option_names' LANGUAGE C IMMUTABLE STRICT;

-- Sets the option of the sift_table tbl; the value is checked as the setting of the same name checks it (type, range, unit). The owner of the table.
CREATE FUNCTION sift.table_set_option(tbl regclass, name text, value text) RETURNS void
	AS 'MODULE_PATHNAME', 'sift_table_set_option' LANGUAGE C VOLATILE;

-- Takes the option off the table (all its options when name is NULL): the setting applies again.
CREATE FUNCTION sift.table_reset_option(tbl regclass, name text DEFAULT NULL) RETURNS void
	AS 'MODULE_PATHNAME', 'sift_table_reset_option' LANGUAGE C VOLATILE;

-- The value in force on the table: its own option, else the setting.
CREATE FUNCTION sift.table_get_option(tbl regclass, name text) RETURNS text
	AS 'MODULE_PATHNAME', 'sift_table_get_option' LANGUAGE C STABLE STRICT;

-- The option as the table stores it; NULL when the table sets none.
CREATE FUNCTION sift._table_option_stored(tbl regclass, name text) RETURNS text
	AS 'MODULE_PATHNAME', 'sift_table_option_stored' LANGUAGE C STABLE STRICT;

-- The options of the sift_tables the caller may see (or of tbl): the value in force, where it comes from (table: set on the table; setting: the sift.table_*
-- setting is not at its built-in default; default) and the built-in default.
CREATE FUNCTION sift.table_options(tbl regclass DEFAULT NULL)
	RETURNS TABLE (rel regclass, name text, value text, source text, default_value text)
	LANGUAGE plpgsql STABLE AS $fn$
BEGIN
	PERFORM sift._table_option_names();		-- loads the library, which defines the settings
	RETURN QUERY
	SELECT c.oid::regclass, o.name,
		   coalesce(sift._table_option_stored(c.oid, o.name), pg_catalog.current_setting('sift.table_' || o.name)),
		   CASE WHEN sift._table_option_stored(c.oid, o.name) IS NOT NULL THEN 'table' WHEN s.source <> 'default' THEN 'setting' ELSE 'default' END,
		   s.boot_val || coalesce(s.unit, '')
	FROM pg_catalog.pg_class c
	JOIN pg_catalog.pg_am a ON a.oid = c.relam AND a.amname = 'sift_table'
	CROSS JOIN LATERAL pg_catalog.unnest(sift._table_option_names()) WITH ORDINALITY AS o(name, ord)
	JOIN pg_catalog.pg_settings s ON s.name = 'sift.table_' || o.name
	WHERE c.relkind IN ('r', 'm') AND (tbl IS NULL OR c.oid = tbl)
	  AND pg_catalog.has_table_privilege(c.oid, 'SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER')
	ORDER BY c.oid::regclass::text, o.ord;
END
$fn$;

COMMENT ON TABLE sift.table_option_values IS 'The per-table options of the sift_tables, one row for each option a table sets (sift.table_set_option writes it, sift.table_options reads it); dumped with the database.';
COMMENT ON FUNCTION sift.table_set_option(regclass, text, text) IS 'Sets a per-table option of a sift_table (the name of a sift.table_* setting without the prefix); the value is checked as the setting checks it. The table option beats the setting whatever its source.';
COMMENT ON FUNCTION sift.table_reset_option(regclass, text) IS 'Takes an option (all of them when the name is NULL) off a sift_table, so that the setting applies again.';
COMMENT ON FUNCTION sift.table_get_option(regclass, text) IS 'The value of a per-table option in force on a sift_table: its own option, else the setting.';
COMMENT ON FUNCTION sift.table_options(regclass) IS 'The per-table options of the sift_tables the caller may see (or of one): value in force, source (table, setting, default) and built-in default.';
COMMENT ON FUNCTION sift._table_option_names() IS 'The names of the per-table options of a sift_table (the support of sift.table_options).';
COMMENT ON FUNCTION sift._table_option_stored(regclass, text) IS 'The option as the table stores it, NULL when the table sets none (the support of sift.table_options).';
COMMENT ON FUNCTION sift.table_option_values_trg() IS 'Trigger of sift.table_option_values: skips the row of a table the database does not have and replaces the entry of the table in the cache of every backend.';
COMMENT ON FUNCTION sift.tblref_in(cstring) IS 'Input of sift.tblref: the qualified name of a table, read as its OID here (0 for a table that is not there).';
COMMENT ON FUNCTION sift.tblref_out(sift.tblref) IS 'Output of sift.tblref: the qualified name of the table.';

-- What to change for each slug of sift.refusal_slugs() (the statement, the index or a setting: the HINT of the notice of sift.explain_refusals). A function of its own: the result of
-- refusal_slugs() cannot gain a column without dropping the function, and a view that reads it would fail the update.
CREATE FUNCTION sift.refusal_actions(OUT slug text, OUT action text)
	RETURNS SETOF record
	AS 'MODULE_PATHNAME', 'sift_agg_refusal_actions' LANGUAGE C STRICT VOLATILE;
COMMENT ON FUNCTION sift.refusal_actions() IS 'What to change for each slug of sift.refusal_slugs(): the statement, the index or a setting (the hint of the notice of sift.explain_refusals); join it to sift.explain_why on slug';
COMMENT ON FUNCTION sift.help(text) IS 'The row of docs/messages.tsv of a pg_sift message or choice by its slug, or the registry row of a refusal slug of sift.explain_why (all message rows without an argument)';

-- sift.workers also lists the sift_table: its column is named index and holds the table, kind is sift_table, and its segments and sealed write buffers are as for an index.
-- The background of a table keeps no note of a failed step, so its failures, failed_at, next_retry, error_code and last_error are 0 and NULL, and due is NULL.
CREATE OR REPLACE VIEW sift.workers AS
	SELECT c.oid::regclass AS index, a.amname::text AS kind, w.segments, w.sealed, w.running, w.failures, w.failed_at, w.next_retry,
		   w.error_code, w.last_error,
		   CASE WHEN a.amname = 'sift_col' THEN sift.col_maintenance_due(c.oid) END AS due
	FROM pg_class c
	JOIN pg_am a ON a.oid = c.relam AND a.amname IN ('sift', 'sift_col')
	JOIN pg_index i ON i.indexrelid = c.oid
	CROSS JOIN LATERAL sift.worker_state(c.oid) w
	WHERE c.relkind = 'i' AND has_table_privilege(i.indrelid, 'SELECT') AND w.segments IS NOT NULL
	UNION ALL
	SELECT c.oid::regclass, a.amname::text, w.segments, w.sealed, w.running, w.failures, w.failed_at, w.next_retry,
		   w.error_code, w.last_error, NULL::text
	FROM pg_class c
	JOIN pg_am a ON a.oid = c.relam AND a.amname = 'sift_table'
	CROSS JOIN LATERAL sift.worker_state(c.oid) w
	WHERE c.relkind IN ('r', 'm') AND has_table_privilege(c.oid, 'SELECT') AND w.segments IS NOT NULL;
COMMENT ON VIEW sift.workers IS 'One row per sift and sift_col index and per sift_table whose table you can read (index: the index, or the table): segments, sealed write buffers, whether maintenance runs, the failures in a row of the background, when, the next retry, the SQLSTATE and the last error (a sift_table keeps none), and why a sift_col index would be rewritten (due).';

-- The option rollup_dims is rollup_columns since 4.2 (the older name stays accepted): the list matches both spellings, so the command is the same whichever an index carries.
CREATE OR REPLACE FUNCTION sift.upgrade_commands(OUT kind text, OUT object regclass, OUT reason text, OUT command text, OUT required bool)
	RETURNS SETOF record
	LANGUAGE plpgsql SET lock_timeout = '100ms' AS $fn$
DECLARE
	r record;
	lo int;
	why text;
BEGIN
	kind := 'reindex'; required := true;
	FOR r IN SELECT c.oid AS relid, format('%I.%I', n.nspname, c.relname) AS name
			 FROM pg_class c JOIN pg_am am ON am.oid = c.relam JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_index i ON i.indexrelid = c.oid
			 WHERE am.amname = 'sift' AND c.relkind = 'i' AND has_table_privilege(i.indrelid, 'MAINTAIN') AND 'analyzer=keyword' <> ALL (coalesce(c.reloptions, '{}')) ORDER BY n.nspname, c.relname
	LOOP
		object := r.relid; command := format('REINDEX INDEX CONCURRENTLY %s;', r.name);
		reason := 'long words (over 255 bytes) are found only after REINDEX: an index built before 3.3 holds such a word cut at 255 bytes, so a query for it finds nothing';
		RETURN NEXT;
	END LOOP;
	required := false;
	FOR r IN SELECT c.oid AS relid, c.reloptions AS opts, format('%I.%I', n.nspname, c.relname) AS name
			 FROM pg_class c JOIN pg_am am ON am.oid = c.relam JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_index i ON i.indexrelid = c.oid
			 WHERE am.amname = 'sift_col' AND c.relkind = 'i' AND has_table_privilege(i.indrelid, 'MAINTAIN') ORDER BY n.nspname, c.relname
	LOOP
		BEGIN
			SELECT min((to_jsonb(s)->>'format')::int) INTO lo FROM sift.col_segments(r.relid) s;
		EXCEPTION WHEN OTHERS THEN
			IF SQLSTATE IN ('42883', '58P01', '55P03') THEN RAISE; END IF;
			CONTINUE;
		END;
		CONTINUE WHEN lo IS NULL;
		-- a segment is minor 9, 10, 13, 14, 15 or 20 when some unit holds the section (for 14, when the segment holds a point directory), so a lower minor means none does
		-- cube_minmax and superunit are on unless the index says otherwise (dflt): the units written before 3.3 lack their sections
		SELECT string_agg(format('%s with column format %s: %s', v.what, lo, v.why), '; ' ORDER BY v.minor) INTO why FROM (VALUES
				('^paths=', 6, 'JSON paths', false,
				 'segments before format 6 may lack the kinds of a JSON null or of a type; a predicate that needs them uses the normal plan, and a REINDEX writes them'),
				('^cube_minmax=(false|off)$', 7, 'cube_minmax', true,
				 'the units written before the option scan; a REINDEX writes the minimum and maximum of every unit'),
				('^(rollup_dims|rollup_columns)=.', 9, 'rollup_columns', false,
				 'the units written before the option have no rollup of their rows and are scanned (nor has a unit over the caps of the rollup, which a REINDEX does not change); a REINDEX writes it where it fits'),
				('^key_columns=.', 10, 'key_columns', false,
				 'the units written before the option have no keys of the column (a summary, or a key list for a count(DISTINCT)), so an equality or a distinct count reads them whole (nor has a unit whose keys would take more than 3% of its bytes, which a REINDEX does not change); a REINDEX writes them where they fit'),
				('^superunit=0$', 12, 'superunit', true,
				 'the units written before the option have no summary of a group of units, so a scan of the whole index reads and folds every unit; a REINDEX writes it'),
				('^order_columns=.', 13, 'order_columns', false,
				 'the units written before the option have no sorted run, so an ORDER BY .. LIMIT on the column reads them whole; a REINDEX writes it'),
				('^point_columns=.', 14, 'point_columns', false,
				 'the segments written before the option have no point directory, so an equality or IN list on the column reads their units (zones and summaries cut most); a REINDEX writes it'),
				('^point_tid_columns=.', 16, 'point_tid_columns', false,
				 'the segments written before the option list the rows of the column by their number in the segment, so an equality or IN list on the column reads the units of the rows it finds; a REINDEX lists their heap addresses'),
				('^key_cut=(?!off$|auto$)', 17, 'key_cut', false,
				 'the units written before the option are cut by superblock, so a lookup or a range of the key reads units from every part of the heap; a REINDEX cuts them by the key'),
				('^hll_columns=.', 20, 'hll_columns', false,
				 'the units written before the option have no sketch of the column, so sift.approx_count_distinct reads their rows; a REINDEX writes it'),
				('^order_array_columns=(?!off$|auto$)', 21, 'order_array_columns', false,
				 'the segments written before the option have no global order of the columns, so an ORDER BY .. LIMIT above 32 rows on one reads their units; a REINDEX writes it')
			 ) v(pat, minor, what, dflt, why)
			 WHERE lo < v.minor AND EXISTS (SELECT 1 FROM unnest(r.opts) opt WHERE opt ~ v.pat) <> v.dflt;
		CONTINUE WHEN why IS NULL;
		object := r.relid; reason := why; command := format('REINDEX INDEX CONCURRENTLY %s;', r.name);
		RETURN NEXT;
	END LOOP;
	kind := 'rewrite';
	FOR r IN SELECT c.oid AS relid, format('%I.%I', n.nspname, c.relname) AS name,
					EXISTS (SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped
							AND a.atttypid IN ('date'::regtype, 'timestamp'::regtype, 'timestamptz'::regtype)) AS calendar,
					EXISTS (SELECT 1 FROM sift.table_paths_maintained p WHERE p.relid = c.oid) AS paths
			 FROM pg_class c JOIN pg_am am ON am.oid = c.relam JOIN pg_namespace n ON n.oid = c.relnamespace
			 WHERE am.amname = 'sift_table' AND c.relkind IN ('r', 'm') AND has_table_privilege(c.oid, 'MAINTAIN') ORDER BY n.nspname, c.relname
	LOOP
		BEGIN
			SELECT min(s.format) INTO lo FROM sift.table_segments(r.relid) s WHERE s.kind = 'columns';
		EXCEPTION WHEN OTHERS THEN
			IF SQLSTATE IN ('42883', '58P01', '55P03') THEN RAISE; END IF;
			CONTINUE;
		END;
		-- the segments of a table with declared paths are never minor 8 (src/col/seg_write.c)
		why := concat_ws('; ',
				CASE WHEN r.calendar AND NOT r.paths AND lo < 8 THEN
					format('column format %s: the segments have no calendar cells of the date and timestamp columns, so a GROUP BY on a day or a month of them reads the rows', lo) END,
				CASE WHEN r.paths AND lo < 6 THEN
					format('column format %s: the segments may lack the kinds of a JSON null or of a type of the declared paths, and a predicate that needs them uses the normal plan', lo) END);
		CONTINUE WHEN why = '';
		object := r.relid; reason := why || '. It takes an ACCESS EXCLUSIVE lock and rewrites the table in the format of sift.col_format_minor: schedule it';
		command := format('VACUUM FULL %s;', r.name);
		RETURN NEXT;
	END LOOP;
END
$fn$;