/* * Function to create id partitions */ CREATE FUNCTION create_id_partition (p_parent_table text, p_control text, p_interval bigint, p_partition_ids bigint[]) RETURNS text LANGUAGE plpgsql SECURITY DEFINER AS $$ DECLARE v_all text[] := ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER']; v_grantees text[]; v_job_id bigint; v_jobmon_schema text; v_old_search_path text; v_parent_grant record; v_parent_owner text; v_parent_schema text; v_parent_tablename text; v_partition_name text; v_revoke text[]; v_step_id bigint; v_tablename text; v_id bigint; BEGIN SELECT nspname INTO v_jobmon_schema FROM pg_namespace n, pg_extension e WHERE e.extname = 'pg_jobmon' AND e.extnamespace = n.oid; IF v_jobmon_schema IS NOT NULL THEN SELECT current_setting('search_path') INTO v_old_search_path; EXECUTE 'SELECT set_config(''search_path'',''@extschema@,'||v_jobmon_schema||''',''false'')'; END IF; SELECT tableowner, schemaname, tablename INTO v_parent_owner, v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table; FOREACH v_id IN ARRAY p_partition_ids LOOP v_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_id::text, TRUE); -- If child table already exists, skip creation SELECT tablename INTO v_tablename FROM pg_catalog.pg_tables WHERE schemaname ||'.'|| tablename = v_partition_name; IF v_tablename IS NOT NULL THEN CONTINUE; END IF; IF v_jobmon_schema IS NOT NULL THEN v_job_id := add_job('PARTMAN CREATE TABLE: '||p_parent_table); v_step_id := add_step(v_job_id, 'Creating new partition '||v_partition_name||' with interval from '||v_id||' to '||(v_id + p_interval)-1); END IF; EXECUTE 'CREATE TABLE '||v_partition_name||' (LIKE '||p_parent_table||' INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING INDEXES INCLUDING STORAGE INCLUDING COMMENTS)'; SELECT tablename INTO v_tablename FROM pg_catalog.pg_tables WHERE schemaname ||'.'|| tablename = v_partition_name; EXECUTE 'ALTER TABLE '||v_partition_name||' ADD CONSTRAINT '||v_tablename||'_partition_check CHECK ('||p_control||'>='||quote_literal(v_id)||' AND '||p_control||'<'||quote_literal(v_id + p_interval)||')'; EXECUTE 'ALTER TABLE '||v_partition_name||' INHERIT '||p_parent_table; FOR v_parent_grant IN SELECT array_agg(DISTINCT privilege_type::text ORDER BY privilege_type::text) AS types, grantee FROM information_schema.table_privileges WHERE table_schema ||'.'|| table_name = p_parent_table GROUP BY grantee LOOP EXECUTE 'GRANT '||array_to_string(v_parent_grant.types, ',')||' ON '||v_partition_name||' TO '||v_parent_grant.grantee; SELECT array_agg(r) INTO v_revoke FROM (SELECT unnest(v_all) AS r EXCEPT SELECT unnest(v_parent_grant.types)) x; IF v_revoke IS NOT NULL THEN EXECUTE 'REVOKE '||array_to_string(v_revoke, ',')||' ON '||v_partition_name||' FROM '||v_parent_grant.grantee||' CASCADE'; END IF; v_grantees := array_append(v_grantees, v_parent_grant.grantee::text); END LOOP; -- Revoke all privileges from roles that have none on the parent IF v_grantees IS NOT NULL THEN SELECT array_agg(r) INTO v_revoke FROM ( SELECT DISTINCT grantee::text AS r FROM information_schema.table_privileges WHERE table_schema ||'.'|| table_name = v_partition_name EXCEPT SELECT unnest(v_grantees)) x; IF v_revoke IS NOT NULL THEN EXECUTE 'REVOKE ALL ON '||v_partition_name||' FROM '||array_to_string(v_revoke, ','); END IF; END IF; EXECUTE 'ALTER TABLE '||v_partition_name||' OWNER TO '||v_parent_owner; IF v_jobmon_schema IS NOT NULL THEN PERFORM update_step(v_step_id, 'OK', 'Done'); PERFORM close_job(v_job_id); END IF; END LOOP; IF v_jobmon_schema IS NOT NULL THEN EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')'; END IF; RETURN v_partition_name; EXCEPTION WHEN OTHERS THEN IF v_jobmon_schema IS NOT NULL THEN EXECUTE 'SELECT set_config(''search_path'',''@extschema@,'||v_jobmon_schema||''',''false'')'; IF v_job_id IS NULL THEN v_job_id := add_job('PARTMAN CREATE TABLE: '||p_parent_table); v_step_id := add_step(v_job_id, 'Partition maintenance for table '||p_parent_table||' failed'); ELSIF v_step_id IS NULL THEN v_step_id := add_step(v_job_id, 'EXCEPTION before first step logged'); END IF; PERFORM update_step(v_step_id, 'BAD', 'ERROR: '||coalesce(SQLERRM,'unknown')); PERFORM fail_job(v_job_id); EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')'; END IF; RAISE EXCEPTION '%', SQLERRM; END $$;