text: -- A new partitioning type has been added to allow setting almost any desired time interval (time-custom). The smallest interval supported is 1 second and the upper limit is bounded by the minimum and maximum timestamp values that PostgreSQL supports (http://www.postgresql.org/docs/current/static/datatype-datetime.html). This feature uses the range data type for internal configuration management, so it is only supported in PostgreSQL 9.2+.
    -- The custom time interval is less efficient than both time-static and time-dynamic since it must use a lookup table. If your needed partitioning interval can fit in one of the pre-made intervals given in the documentation, it is highly recommended to use one of those for better performance. time-static is still the best method when performance of inserts is important. See the documentation for more details on this new partitioning type.
  -- New parameter to create_parent() that sets what the first partition in the set will be (p_start_partition). 
    -- Must be a valid timestamp (for time-based) or positive integer (for id-based) value. Be aware, though, the actual paramater data type is text. 
    -- For time-based partitioning, all partitions starting with the given timestamp up to CURRENT_TIMESTAMP (plus premake) will be created.
    -- For id-based partitioning, only the partition starting at the given value (plus premake) will be made. 
  -- pg_jobmon is now truly optional. Additonal configuration option for each individual partition set to turn it off and on. run_maintenance() now has an optional parameter to turn it off when being run. If you tried to partition pg_jobmon tables before, it would cause a permanent lockwait. Turn pg_jobmon off for those tables to avoid this.
  -- Fixed partition_data_time() & partition_data_id() functions to recreate the parent trigger function when static partitioning is used. Without this, partitioning more recent data that may have gotten into the parent table could possibly leave the function without conditions for the new partitions. run_maintenance() would eventually fix this for time partitioning, but id partitioning could be left in a broken state forever. (Github issue #16) 
  -- Fixed bug in partition_data_time() & partition_data_id() to reset the lock wait counter properly between loops. Bug reported & fixed by bougyman from #postgresql on Freenode.
  -- pg_partman only supports id intervals greater than 1. May see if I can get an interval of 1 working later, but changed create_parent() to check for this and not allow it since it won't work properly at this time. New partitions were not automatically created if interval was set to 1. (Github issue #15)
  -- Clarify in docs that the id interval value passed to create_parent() must actually be in text type format.
  -- Changed drop & undo partition functions to use transaction based advistory locks.
  -- Removed need for internally used function create_next_time_partition() and therefore dropped the function. 
  -- Simplified the create_time_partition() & create_id_partition() parameter lists.
  
  ALTER TABLE @extschema@.part_config ADD jobmon boolean DEFAULT true;
  
  CREATE TEMP TABLE partman_preserve_privs_temp (statement text);
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.create_parent(text, text, text, text, text[], int, text, boolean, boolean) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'create_parent'; 
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.create_time_partition (text, timestamp[]) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'create_time_partition'; 
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.create_id_partition (text, bigint[]) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'create_id_partition'; 
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.undo_partition(text, int, boolean, boolean) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'undo_partition'; 
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.create_id_function(text) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'create_id_function'; 
  
  INSERT INTO partman_preserve_privs_temp 
  SELECT 'GRANT EXECUTE ON FUNCTION @extschema@.run_maintenance(boolean) TO '||array_to_string(array_agg(grantee::text), ',')||';' 
  FROM information_schema.routine_privileges
  WHERE routine_schema = '@extschema@'
  AND routine_name = 'run_maintenance'; 
  
  DROP FUNCTION @extschema@.create_parent(text, text, text, text, text[], int , boolean);
  DROP FUNCTION @extschema@.create_time_partition (text, text, interval, text, timestamp[]);
  DROP FUNCTION @extschema@.create_id_partition (text, text, bigint, bigint[]);
  DROP FUNCTION @extschema@.undo_partition(text, int, boolean);
  DROP FUNCTION @extschema@.create_id_function(text, bigint);
  DROP FUNCTION @extschema@.run_maintenance();
  DROP FUNCTION @extschema@.create_next_time_partition(text);
  
  /*
   * Check PostgreSQL version number. Parameter must be full 3 point version.
   * Returns true if current version is greater than or equal to the parameter given.
   */
  CREATE FUNCTION check_version(p_check_version text) RETURNS boolean
      LANGUAGE plpgsql STABLE
      AS $$
  DECLARE
  
  v_check_version     text[];
  v_current_version   text[] := string_to_array(current_setting('server_version'), '.');
   
  BEGIN
  
  v_check_version := string_to_array(p_check_version, '.');
  
  IF v_current_version[1]::int > v_check_version[1]::int THEN
      RETURN true;
  END IF;
  IF v_current_version[1]::int = v_check_version[1]::int THEN
      IF v_current_version[2]::int > v_check_version[2]::int THEN
          RETURN true;
      END IF;
      IF v_current_version[2]::int = v_check_version[2]::int THEN
          IF v_current_version[3]::int >= v_check_version[3]::int THEN
              RETURN true;
          END IF; -- 0.0.x
      END IF; -- 0.x.0
  END IF; -- x.0.0
  
  RETURN false;
  
  END
  $$;
  
  /********* 9.2+ stuff ********/
  DO $$
  BEGIN
  IF @extschema@.check_version('9.2.0') THEN
      CREATE TABLE custom_time_partitions (
          parent_table text NOT NULL
          , child_table text NOT NULL
          , partition_range tstzrange NOT NULL
          , PRIMARY KEY (parent_table, child_table));
      CREATE INDEX custom_time_partitions_partition_range_idx ON custom_time_partitions USING gist (partition_range);
  END IF;
  END
  $$;
  /********* end 9.2+ stuff ********/
  
  
  /*
   * Check function for config table partition types
   */
  CREATE OR REPLACE FUNCTION check_partition_type (p_type text) RETURNS boolean
      LANGUAGE plpgsql IMMUTABLE SECURITY DEFINER
      AS $$
  DECLARE
  v_result    boolean;
  BEGIN
      SELECT p_type IN ('time-static', 'time-dynamic', 'time-custom', 'id-static', 'id-dynamic') INTO v_result;
      RETURN v_result;
  END
  $$;
  
  
  /*
   * Function to turn a table into the parent of a partition set
   */
  CREATE FUNCTION create_parent(
      p_parent_table text
      , p_control text
      , p_type text
      , p_interval text
      , p_constraint_cols text[] DEFAULT NULL 
      , p_premake int DEFAULT 4
      , p_start_partition text DEFAULT NULL
      , p_jobmon boolean DEFAULT true
      , p_debug boolean DEFAULT false) 
  RETURNS void
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_base_timestamp        timestamp;
  v_count                 int := 1;
  v_datetime_string       text;
  v_id_interval           bigint;
  v_job_id                bigint;
  v_jobmon_schema         text;
  v_last_partition_name   text;
  v_old_search_path       text;
  v_partition_time        timestamp;
  v_partition_time_array  timestamp[];
  v_partition_id          bigint[];
  v_max                   bigint;
  v_notnull               boolean;
  v_start_time            timestamp;
  v_starting_partition_id bigint;
  v_step_id               bigint;
  v_step_overflow_id      bigint;
  v_tablename             text;
  v_time_interval         interval;
  
  BEGIN
  
  IF position('.' in p_parent_table) = 0  THEN
      RAISE EXCEPTION 'Parent table must be schema qualified';
  END IF;
  
  SELECT tablename INTO v_tablename FROM pg_tables WHERE schemaname || '.' || tablename = p_parent_table;
      IF v_tablename IS NULL THEN
          RAISE EXCEPTION 'Please create given parent table first: %', p_parent_table;
      END IF;
  
  SELECT attnotnull INTO v_notnull FROM pg_attribute WHERE attrelid = p_parent_table::regclass AND attname = p_control;
      IF v_notnull = false OR v_notnull IS NULL THEN
          RAISE EXCEPTION 'Control column (%) for parent table (%) must be NOT NULL', p_control, p_parent_table;
      END IF;
  
  IF NOT @extschema@.check_partition_type(p_type) THEN
      RAISE EXCEPTION '% is not a valid partitioning type', p_type;
  END IF;
  
  IF p_type = 'time-custom' AND @extschema@.check_version('9.2.0') IS FALSE THEN
      RAISE EXCEPTION 'The "time-custom" type requires a minimum PostgreSQL version of 9.2.0';
  END IF;
  
  EXECUTE 'LOCK TABLE '||p_parent_table||' IN ACCESS EXCLUSIVE MODE';
  
  IF p_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN SETUP PARENT: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Creating initial partitions on new parent table: '||p_parent_table);
  END IF;
  
  IF p_type = 'time-static' OR p_type = 'time-dynamic' OR p_type = 'time-custom' THEN
  
      CASE
          WHEN p_interval = 'yearly' THEN
              v_time_interval := '1 year';
          WHEN p_interval = 'quarterly' THEN
              v_time_interval := '3 months';
          WHEN p_interval = 'monthly' THEN
              v_time_interval := '1 month';
          WHEN p_interval  = 'weekly' THEN
              v_time_interval := '1 week';
          WHEN p_interval = 'daily' THEN
              v_time_interval := '1 day';
          WHEN p_interval = 'hourly' THEN
              v_time_interval := '1 hour';
          WHEN p_interval = 'half-hour' THEN
              v_time_interval := '30 mins';
          WHEN p_interval = 'quarter-hour' THEN
              v_time_interval := '15 mins';
          ELSE
              IF p_type <> 'time-custom' THEN
                  RAISE EXCEPTION 'Must use a predefined time interval if not using type "time-custom". See documentation.';
              END IF;
              v_time_interval := p_interval::interval;
              IF v_time_interval < '1 second'::interval THEN
                  RAISE EXCEPTION 'Partitioning interval must be 1 second or greater';
              END IF;
      END CASE;
  
      -- First partition is either the min premake or p_start_partition
      v_start_time := COALESCE(p_start_partition::timestamp, CURRENT_TIMESTAMP - (v_time_interval * p_premake));
  
      IF v_time_interval >= '1 year' THEN
          v_base_timestamp := date_trunc('year', v_start_time);
          IF v_time_interval >= '10 years' THEN
              v_base_timestamp := date_trunc('decade', v_start_time);
              IF v_time_interval >= '100 years' THEN
                  v_base_timestamp := date_trunc('century', v_start_time);
                  IF v_time_interval >= '1000 years' THEN
                      v_base_timestamp := date_trunc('millennium', v_start_time);
                  END IF; -- 1000
              END IF; -- 100
          END IF; -- 10
      END IF; -- 1
     
      v_datetime_string := 'YYYY';
      IF v_time_interval < '1 year' THEN
          IF p_interval = 'quarterly' THEN
              v_base_timestamp := date_trunc('quarter', v_start_time);
              v_datetime_string = 'YYYY"q"Q';
          ELSE
              v_base_timestamp := date_trunc('month', v_start_time); 
              v_datetime_string := v_datetime_string || '_MM';
          END IF;
          IF v_time_interval < '1 month' THEN
              IF p_interval = 'weekly' THEN
                  v_base_timestamp := date_trunc('week', v_start_time);
                  v_datetime_string := 'IYYY"w"IW';
              ELSE 
                  v_base_timestamp := date_trunc('day', v_start_time);
                  v_datetime_string := v_datetime_string || '_DD';
              END IF;
              IF v_time_interval < '1 day' THEN
                  v_base_timestamp := date_trunc('hour', v_start_time);
                  v_datetime_string := v_datetime_string || '_HH24MI';
                  IF v_time_interval < '1 minute' THEN
                      v_base_timestamp := date_trunc('minute', v_start_time);
                      v_datetime_string := v_datetime_string || 'SS';
                  END IF; -- minute
              END IF; -- day
          END IF; -- month
      END IF; -- year
  
      v_partition_time_array := array_append(v_partition_time_array, v_base_timestamp);
      LOOP
          -- If current loop value is less than or equal to the value of the max premake, add time to array.
          IF (v_base_timestamp + (v_time_interval * v_count)) < (CURRENT_TIMESTAMP + (v_time_interval * p_premake)) THEN
              BEGIN
                  v_partition_time := (v_base_timestamp + (v_time_interval * v_count))::timestamp;
                  v_partition_time_array := array_append(v_partition_time_array, v_partition_time);
              EXCEPTION WHEN datetime_field_overflow THEN
                  RAISE WARNING 'Attempted partition time interval is outside PostgreSQL''s supported time range. 
                      Child partition creation after time % skipped', v_partition_time;
                  v_step_overflow_id := add_step(v_job_id, 'Attempted partition time interval is outside PostgreSQL''s supported time range.');
                  PERFORM update_step(v_step_overflow_id, 'CRITICAL', 'Child partition creation after time '||v_partition_time||' skipped');
                  CONTINUE;
              END;
          ELSE
              EXIT; -- all needed partitions added to array. Exit the loop.
          END IF;
          v_count := v_count + 1;        
      END LOOP;
  
      INSERT INTO @extschema@.part_config (parent_table, type, part_interval, control, premake, constraint_cols, datetime_string, jobmon) VALUES
          (p_parent_table, p_type, v_time_interval, p_control, p_premake, p_constraint_cols, v_datetime_string, p_jobmon);
      v_last_partition_name := @extschema@.create_time_partition(p_parent_table, v_partition_time_array);
      -- Doing separate update because create function requires in config table last_partition to be set
      UPDATE @extschema@.part_config SET last_partition = v_last_partition_name WHERE parent_table = p_parent_table;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Time partitions premade: '||p_premake);
      END IF;
  END IF;
  
  IF p_type = 'id-static' OR p_type = 'id-dynamic' THEN
      v_id_interval := p_interval::bigint;
      IF v_id_interval <= 1 THEN
          RAISE EXCEPTION 'Interval for serial partitioning must be greater than 1';
      END IF;
      
      -- If custom start partition is set, use that. 
      -- If custom start is not set and there is already data, start partitioning with the highest current value
      EXECUTE 'SELECT COALESCE('||quote_nullable(p_start_partition::bigint)||', max('||p_control||')::bigint, 0) FROM '||p_parent_table||' LIMIT 1' INTO v_max;
      v_starting_partition_id := v_max - (v_max % v_id_interval);
      FOR i IN 0..p_premake LOOP
          -- Only make previous partitions if ID value is less than the starting value and positive (and custom start partition wasn't set)
          IF p_start_partition IS NULL AND 
              (v_starting_partition_id - (v_id_interval*i)) > 0 
              AND (v_starting_partition_id - (v_id_interval*i)) < v_starting_partition_id 
          THEN
              v_partition_id = array_append(v_partition_id, (v_starting_partition_id - v_id_interval*i));
          END IF; 
          v_partition_id = array_append(v_partition_id, (v_id_interval*i) + v_starting_partition_id);
      END LOOP;
  
      INSERT INTO @extschema@.part_config (parent_table, type, part_interval, control, premake, constraint_cols, jobmon) VALUES
          (p_parent_table, p_type, v_id_interval, p_control, p_premake, p_constraint_cols, p_jobmon);
      v_last_partition_name := @extschema@.create_id_partition(p_parent_table, v_partition_id);
      -- Doing separate update because create function needs parent table in config table for apply_grants()
      UPDATE @extschema@.part_config SET last_partition = v_last_partition_name WHERE parent_table = p_parent_table;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'ID partitions premade: '||p_premake);
      END IF;
      
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Creating partition function');
  END IF;
  
  IF p_type = 'time-static' OR p_type = 'time-dynamic' OR p_type = 'time-custom' THEN
      PERFORM @extschema@.create_time_function(p_parent_table);
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Time function created');
      END IF;
  ELSIF p_type = 'id-static' OR p_type = 'id-dynamic' THEN
      PERFORM @extschema@.create_id_function(p_parent_table);  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'ID function created');
      END IF;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Creating partition trigger');
  END IF;
  
  PERFORM @extschema@.create_trigger(p_parent_table);
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Done');
      IF v_step_overflow_id IS NOT NULL THEN
          PERFORM fail_job(v_job_id);
      ELSE
          PERFORM close_job(v_job_id);
      END IF;
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE PARENT: '||p_parent_table||')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''Partition creation for table '||p_parent_table||' failed'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to create a child table in a time-based partition set
   */
  CREATE FUNCTION create_time_partition (p_parent_table text, p_partition_times timestamp[]) 
  RETURNS text
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_all                           text[] := ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER'];
  v_control                       text;
  v_grantees                      text[];
  v_job_id                        bigint;
  v_jobmon                        boolean;
  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_partition_suffix              text;
  v_parent_tablespace             text;
  v_part_interval                 interval;
  v_partition_timestamp_end       timestamp;
  v_partition_timestamp_start     timestamp;
  v_quarter                       text;
  v_revoke                        text[];
  v_step_id                       bigint;
  v_step_overflow_id              bigint;
  v_tablename                     text;
  v_trunc_value                   text;
  v_time                          timestamp;
  v_type                          text;
  v_year                          text;
  
  BEGIN
  
  SELECT type
      , control
      , part_interval
      , jobmon
  INTO v_type
      , v_control
      , v_part_interval
      , v_jobmon
  FROM @extschema@.part_config
  WHERE parent_table = p_parent_table
  AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom');
  
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  SELECT tableowner, schemaname, tablename, tablespace INTO v_parent_owner, v_parent_schema, v_parent_tablename, v_parent_tablespace FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  
  FOREACH v_time IN ARRAY p_partition_times LOOP    
  
      v_partition_suffix := to_char(v_time, 'YYYY');
      IF v_part_interval < '1 year' AND v_part_interval <> '1 week' THEN 
          v_partition_suffix := v_partition_suffix ||'_'|| to_char(v_time, 'MM');
          IF v_part_interval < '1 month' AND v_part_interval <> '1 week' THEN 
              v_partition_suffix := v_partition_suffix ||'_'|| to_char(v_time, 'DD');
              IF v_part_interval < '1 day' THEN
                  v_partition_suffix := v_partition_suffix || '_' || to_char(v_time, 'HH24MI');
                  IF v_part_interval < '1 minute' THEN
                      v_partition_suffix := v_partition_suffix || to_char(v_time, 'SS');
                  END IF; -- end < minute IF
              END IF; -- end < day IF      
          END IF; -- end < month IF
      END IF; -- end < year IF
  
      v_partition_timestamp_start := v_time;
      BEGIN
          v_partition_timestamp_end := v_time + v_part_interval;
      EXCEPTION WHEN datetime_field_overflow THEN
          RAISE WARNING 'Attempted partition time interval is outside PostgreSQL''s supported time range. 
              Child partition creation after time % skipped', v_time;
          v_step_overflow_id := add_step(v_job_id, 'Attempted partition time interval is outside PostgreSQL''s supported time range.');
          PERFORM update_step(v_step_overflow_id, 'CRITICAL', 'Child partition creation after time '||v_time||' skipped');
          CONTINUE;
      END;
  
      IF v_part_interval = '1 week' THEN
          v_partition_suffix := to_char(v_time, 'IYYY') || 'w' || to_char(v_time, 'IW');
      END IF;
   
      -- "Q" is ignored in to_timestamp, so handle special case
      IF v_part_interval = '3 months' AND (v_type = 'time-static' OR v_type = 'time-dynamic') THEN
          v_year := to_char(v_time, 'YYYY');
          v_quarter := to_char(v_time, 'Q');
          v_partition_suffix := v_year || 'q' || v_quarter;
      END IF;
  
      v_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_partition_suffix, TRUE);
  
      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_partition_timestamp_start||' to '||(v_partition_timestamp_end-'1sec'::interval));
      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;
      IF v_parent_tablespace IS NOT NULL THEN
          EXECUTE 'ALTER TABLE '||v_partition_name||' SET TABLESPACE '||v_parent_tablespace;
      END IF;
      EXECUTE 'ALTER TABLE '||v_partition_name||' ADD CONSTRAINT '||v_tablename||'_partition_check
          CHECK ('||v_control||'>='||quote_literal(v_partition_timestamp_start)||' AND '||v_control||'<'||quote_literal(v_partition_timestamp_end)||')';
      EXECUTE 'ALTER TABLE '||v_partition_name||' INHERIT '||p_parent_table;
  
      -- If custom time, set extra config options.
      IF v_type = 'time-custom' THEN
          INSERT INTO @extschema@.custom_time_partitions (parent_table, child_table, partition_range)
          VALUES ( p_parent_table, v_partition_name, tstzrange(v_partition_timestamp_start, v_partition_timestamp_end, '[)') );
      END IF;
  
      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');
          IF v_step_overflow_id IS NOT NULL THEN
              PERFORM fail_job(v_job_id);
          ELSE
              PERFORM close_job(v_job_id);
          END IF;
      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
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE TABLE: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Create the trigger function for the parent table of a time-based partition set
   */
  CREATE OR REPLACE FUNCTION create_time_function(p_parent_table text) RETURNS void
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_control                       text;
  v_count                         int;
  v_current_partition_name        text;
  v_current_partition_timestamp   timestamptz;
  v_datetime_string               text;
  v_final_partition_timestamp     timestamptz;
  v_function_name                 text;
  v_job_id                        bigint;
  v_jobmon                        boolean;
  v_jobmon_schema                 text;
  v_old_search_path               text;
  v_new_length                    int;
  v_next_partition_name           text;
  v_next_partition_timestamp      timestamptz;
  v_parent_schema                 text;
  v_parent_tablename              text;
  v_part_interval                 interval;
  v_premake                       int;
  v_prev_partition_name           text;
  v_prev_partition_timestamp      timestamptz;
  v_step_id                       bigint;
  v_trig_func                     text;
  v_type                          text;
  
  BEGIN
  
  SELECT type
      , part_interval::interval
      , control
      , premake
      , datetime_string
      , jobmon
  INTO v_type
      , v_part_interval
      , v_control
      , v_premake
      , v_datetime_string
      , v_jobmon
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table
  AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom');
  
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN CREATE FUNCTION: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Creating partition function for table '||p_parent_table);
  END IF;
  
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  
  v_function_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, '_part_trig_func', FALSE);
  
  IF v_type = 'time-static' THEN
      CASE
          WHEN v_part_interval = '15 mins' THEN
              v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP) + 
                  '15min'::interval * floor(date_part('minute', CURRENT_TIMESTAMP) / 15.0);
          WHEN v_part_interval = '30 mins' THEN
              v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP) + 
                  '30min'::interval * floor(date_part('minute', CURRENT_TIMESTAMP) / 30.0);
          WHEN v_part_interval = '1 hour' THEN
              v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP);
           WHEN v_part_interval = '1 day' THEN
              v_current_partition_timestamp := date_trunc('day', CURRENT_TIMESTAMP);
          WHEN v_part_interval = '1 week' THEN
              v_current_partition_timestamp := date_trunc('week', CURRENT_TIMESTAMP);
          WHEN v_part_interval = '1 month' THEN
              v_current_partition_timestamp := date_trunc('month', CURRENT_TIMESTAMP);
          -- Type time-static plus this interval is the special quarterly interval 
          WHEN v_part_interval = '3 months' THEN
              v_current_partition_timestamp := date_trunc('quarter', CURRENT_TIMESTAMP);
          WHEN v_part_interval = '1 year' THEN
              v_current_partition_timestamp := date_trunc('year', CURRENT_TIMESTAMP);
      END CASE;
  
      v_current_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, to_char(v_current_partition_timestamp, v_datetime_string), TRUE); 
      v_next_partition_timestamp := v_current_partition_timestamp + v_part_interval::interval;
  
      v_trig_func := 'CREATE OR REPLACE FUNCTION '||v_function_name||'() RETURNS trigger LANGUAGE plpgsql AS $t$ 
          BEGIN 
          IF TG_OP = ''INSERT'' THEN 
              IF NEW.'||v_control||' >= '||quote_literal(v_current_partition_timestamp)||' AND NEW.'||v_control||' < '||quote_literal(v_next_partition_timestamp)|| ' THEN 
                  INSERT INTO '||v_current_partition_name||' VALUES (NEW.*); ';
      FOR i IN 1..v_premake LOOP
          v_prev_partition_timestamp := v_current_partition_timestamp - (v_part_interval::interval * i);
          v_next_partition_timestamp := v_current_partition_timestamp + (v_part_interval::interval * i);
          v_final_partition_timestamp := v_next_partition_timestamp + (v_part_interval::interval);
          v_prev_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, to_char(v_prev_partition_timestamp, v_datetime_string), TRUE);
          v_next_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, to_char(v_next_partition_timestamp, v_datetime_string), TRUE);
  
          -- Check that child table exist before making a rule to insert to them.
          -- Handles edge case of changing premake immediately after running create_parent(). 
          SELECT count(*) INTO v_count FROM pg_catalog.pg_tables WHERE schemaname ||'.'||tablename = v_prev_partition_name;
          IF v_count > 0 THEN
              v_trig_func := v_trig_func ||'
              ELSIF NEW.'||v_control||' >= '||quote_literal(v_prev_partition_timestamp)||' AND NEW.'||v_control||' < '||
                      quote_literal(v_prev_partition_timestamp + v_part_interval::interval)|| ' THEN 
                  INSERT INTO '||v_prev_partition_name||' VALUES (NEW.*);';
          END IF;
          SELECT count(*) INTO v_count FROM pg_catalog.pg_tables WHERE schemaname ||'.'||tablename = v_next_partition_name;
          IF v_count > 0 THEN
              v_trig_func := v_trig_func ||' 
              ELSIF NEW.'||v_control||' >= '||quote_literal(v_next_partition_timestamp)||' AND NEW.'||v_control||' < '||
                  quote_literal(v_final_partition_timestamp)|| ' THEN 
                  INSERT INTO '||v_next_partition_name||' VALUES (NEW.*);';
          END IF;
  
      END LOOP;
      v_trig_func := v_trig_func ||' 
              ELSE 
                  RETURN NEW; 
              END IF; 
          END IF; 
          RETURN NULL; 
          END $t$;';
  
      EXECUTE v_trig_func;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Added function for current time interval: '||
              v_current_partition_timestamp||' to '||(v_final_partition_timestamp-'1sec'::interval));
      END IF;
  
  ELSIF v_type = 'time-dynamic' THEN
  
      v_trig_func := 'CREATE OR REPLACE FUNCTION '||v_function_name||'() RETURNS trigger LANGUAGE plpgsql AS $t$ 
          DECLARE
              v_count                 int;
              v_partition_name        text;
              v_partition_timestamp   timestamptz;
          BEGIN 
          IF TG_OP = ''INSERT'' THEN 
              ';
          CASE
              WHEN v_part_interval = '15 mins' THEN 
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''hour'', NEW.'||v_control||') + 
                      ''15min''::interval * floor(date_part(''minute'', NEW.'||v_control||') / 15.0);';
              WHEN v_part_interval = '30 mins' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''hour'', NEW.'||v_control||') + 
                      ''30min''::interval * floor(date_part(''minute'', NEW.'||v_control||') / 30.0);';
              WHEN v_part_interval = '1 hour' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''hour'', NEW.'||v_control||');';
               WHEN v_part_interval = '1 day' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''day'', NEW.'||v_control||');';
              WHEN v_part_interval = '1 week' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''week'', NEW.'||v_control||');';
              WHEN v_part_interval = '1 month' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''month'', NEW.'||v_control||');';
              WHEN v_part_interval = '3 months' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''quarter'', NEW.'||v_control||');';
              WHEN v_part_interval = '1 year' THEN
                  v_trig_func := v_trig_func||'v_partition_timestamp := date_trunc(''year'', NEW.'||v_control||');';
          END CASE;
          v_trig_func := v_trig_func||'
              v_partition_name := @extschema@.check_name_length('''||v_parent_tablename||''', '''||v_parent_schema||''', to_char(v_partition_timestamp, '||quote_literal(v_datetime_string)||'), TRUE);
              SELECT count(*) INTO v_count FROM pg_tables WHERE schemaname ||''.''|| tablename = v_partition_name;
              IF v_count > 0 THEN 
                  EXECUTE ''INSERT INTO ''||v_partition_name||'' VALUES($1.*)'' USING NEW;
              ELSE
                  RETURN NEW;
              END IF;
          END IF;
          
          RETURN NULL; 
          END $t$;';
  
      EXECUTE v_trig_func;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Added function for dynamic time table: '||p_parent_table);
      END IF;
  
  ELSIF v_type = 'time-custom' THEN
  
      v_trig_func := 'CREATE OR REPLACE FUNCTION '||v_function_name||'() RETURNS trigger LANGUAGE plpgsql AS $t$ 
          DECLARE
              v_child_table       text;
              v_count             int; 
          BEGIN 
  
          SELECT child_table INTO v_child_table
          FROM @extschema@.custom_time_partitions 
          WHERE partition_range @> NEW.'||v_control||' 
          AND parent_table = '||quote_literal(p_parent_table)||';
  
          SELECT count(*) INTO v_count FROM pg_tables WHERE schemaname ||''.''|| tablename = v_child_table;
          IF v_count > 0 THEN
              EXECUTE ''INSERT INTO ''||v_child_table||'' VALUES ($1.*)'' USING NEW;
          ELSE
              RETURN NEW;
          END IF;
  
          RETURN NULL; 
          END $t$;';
  
      EXECUTE v_trig_func;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Added function for custom time table: '||p_parent_table);
      END IF;
  
  ELSE
      RAISE EXCEPTION 'ERROR: Invalid time partitioning type given: %', v_type;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE FUNCTION: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Populate the child table(s) of a time-based partition set with old data from the original parent
   */
  CREATE OR REPLACE FUNCTION partition_data_time(p_parent_table text, p_batch_count int DEFAULT 1, p_batch_interval interval DEFAULT NULL, p_lock_wait numeric DEFAULT 0) RETURNS bigint
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_control                   text;
  v_datetime_string           text;
  v_last_partition_name       text;
  v_max_partition_timestamp   timestamp;
  v_last_partition            text;
  v_lock_iter                 int := 1;
  v_lock_obtained             boolean := FALSE;
  v_min_control               timestamp;
  v_min_partition_timestamp   timestamp;
  v_part_interval             interval;
  v_partition_timestamp       timestamp[];
  v_rowcount                  bigint;
  v_sql                       text;
  v_time_position             int;
  v_total_rows                bigint := 0;
  v_type                      text;
  
  BEGIN
  
  SELECT type
      , part_interval::interval
      , control
      , datetime_string
      , last_partition
  INTO v_type
      , v_part_interval
      , v_control
      , v_datetime_string
      , v_last_partition
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table
  AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom');
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF p_batch_interval IS NULL OR p_batch_interval > v_part_interval THEN
      p_batch_interval := v_part_interval;
  END IF;
  
  FOR i IN 1..p_batch_count LOOP
  
      EXECUTE 'SELECT min('||v_control||') FROM ONLY '||p_parent_table INTO v_min_control;
      IF v_min_control IS NULL THEN
          RETURN 0;
      END IF;
  
      IF v_type = 'time-static' OR v_type = 'time-dynamic' THEN
          CASE
              WHEN v_part_interval = '15 mins' THEN
                  v_min_partition_timestamp := date_trunc('hour', v_min_control) + 
                      '15min'::interval * floor(date_part('minute', v_min_control) / 15.0);
              WHEN v_part_interval = '30 mins' THEN
                  v_min_partition_timestamp := date_trunc('hour', v_min_control) + 
                      '30min'::interval * floor(date_part('minute', v_min_control) / 30.0);
              WHEN v_part_interval = '1 hour' THEN
                  v_min_partition_timestamp := date_trunc('hour', v_min_control);
              WHEN v_part_interval = '1 day' THEN
                  v_min_partition_timestamp := date_trunc('day', v_min_control);
              WHEN v_part_interval = '1 week' THEN
                  v_min_partition_timestamp := date_trunc('week', v_min_control);
              WHEN v_part_interval = '1 month' THEN
                  v_min_partition_timestamp := date_trunc('month', v_min_control);
              WHEN v_part_interval = '3 months' THEN
                  v_min_partition_timestamp := date_trunc('quarter', v_min_control);
              WHEN v_part_interval = '1 year' THEN
                  v_min_partition_timestamp := date_trunc('year', v_min_control);
          END CASE;
      ELSIF v_type = 'time-custom' THEN
          -- Keep going backwards, checking if the time interval encompases the current v_min_control value
          v_time_position := (length(v_last_partition) - position('p_' in reverse(v_last_partition))) + 2;
          v_min_partition_timestamp := to_timestamp(substring(v_last_partition from v_time_position), v_datetime_string);
          v_max_partition_timestamp := v_min_partition_timestamp + v_part_interval;
          LOOP
              IF v_min_control >= v_min_partition_timestamp AND v_min_control < v_max_partition_timestamp THEN
                  EXIT;
              ELSE
                  v_max_partition_timestamp := v_min_partition_timestamp;
                  BEGIN
                      v_min_partition_timestamp := v_min_partition_timestamp - v_part_interval;
                  EXCEPTION WHEN datetime_field_overflow THEN
                      RAISE EXCEPTION 'Attempted partition time interval is outside PostgreSQL''s supported time range. 
                          Unable to create partition with interval before timestamp % ', v_min_partition_interval;
                  END;
              END IF;
          END LOOP;
              
      END IF;
  
      v_partition_timestamp := ARRAY[v_min_partition_timestamp];
      IF (v_min_control + p_batch_interval) >= (v_min_partition_timestamp + v_part_interval) THEN
          v_max_partition_timestamp := v_min_partition_timestamp + v_part_interval;
      ELSE
          v_max_partition_timestamp := v_min_control + p_batch_interval;
      END IF;
  
  -- do some locking with timeout, if required
      IF p_lock_wait > 0  THEN
          v_lock_iter := 0;
          WHILE v_lock_iter <= 5 LOOP
              v_lock_iter := v_lock_iter + 1;
              BEGIN
                  v_sql := 'SELECT * FROM ONLY ' || p_parent_table ||
                  ' WHERE '||v_control||' >= '||quote_literal(v_min_control)||
                  ' AND '||v_control||' < '||quote_literal(v_max_partition_timestamp)
                  ||' FOR UPDATE NOWAIT';
                  EXECUTE v_sql;
                  v_lock_obtained := TRUE;
              EXCEPTION
                  WHEN lock_not_available THEN
                      PERFORM pg_sleep( p_lock_wait / 5.0 );
                      CONTINUE;
              END;
              EXIT WHEN v_lock_obtained;
          END LOOP;
          IF NOT v_lock_obtained THEN
             RETURN -1;
          END IF;
      END IF;
  
      v_last_partition_name := @extschema@.create_time_partition(p_parent_table, v_partition_timestamp);
  
      EXECUTE 'WITH partition_data AS (
              DELETE FROM ONLY '||p_parent_table||' WHERE '||v_control||' >= '||quote_literal(v_min_control)||
                  ' AND '||v_control||' < '||quote_literal(v_max_partition_timestamp)||' RETURNING *)
              INSERT INTO '||v_last_partition_name||' SELECT * FROM partition_data';
  
      GET DIAGNOSTICS v_rowcount = ROW_COUNT;
      v_total_rows := v_total_rows + v_rowcount;
      IF v_rowcount = 0 THEN
          EXIT;
      END IF;
  
  END LOOP; 
  
  IF v_type = 'time-static' THEN
          PERFORM @extschema@.create_time_function(p_parent_table);
  END IF;    
  
  RETURN v_total_rows;
  
  END
  $$;
  
  
  /*
   * Function to manage pre-creation of the next partitions in a time-based partition set.
   * Also manages dropping old partitions if the retention option is set.
   */
  CREATE FUNCTION run_maintenance(p_jobmon BOOLEAN DEFAULT true) RETURNS void 
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock                      boolean;
  v_create_count                  int := 0;
  v_current_partition             text;
  v_current_partition_timestamp   timestamp;
  v_datetime_string               text;
  v_drop_count                    int := 0;
  v_job_id                        bigint;
  v_jobmon                        boolean;
  v_jobmon_schema                 text;
  v_last_partition                text;
  v_last_partition_timestamp      timestamp;
  v_next_partition_timestamp      timestamp;
  v_old_search_path               text;
  v_premade_count                 int;
  v_quarter                       text;
  v_step_id                       bigint;
  v_step_overflow_id              bigint;
  v_row                           record;
  v_tablename                     text;
  v_time_position                 int;
  v_year                          text;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman run_maintenance'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'Partman maintenance already running.';
      RETURN;
  END IF;
  
  IF p_jobmon THEN
      SELECT nspname INTO v_jobmon_schema FROM pg_catalog.pg_namespace n, pg_catalog.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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN RUN MAINTENANCE');
      v_step_id := add_step(v_job_id, 'Running maintenance loop');
  END IF;
  
  FOR v_row IN 
  SELECT parent_table
      , type
      , part_interval::interval
      , control
      , premake
      , datetime_string
      , last_partition
      , undo_in_progress
  FROM @extschema@.part_config WHERE type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom'
  LOOP
  
      CONTINUE WHEN v_row.undo_in_progress;
      
      -- Double check that last created partition exists
      IF v_row.last_partition IS NOT NULL THEN
          SELECT tablename INTO v_tablename FROM pg_tables WHERE schemaname || '.' || tablename = v_row.last_partition;
          IF v_tablename IS NULL THEN
              RAISE EXCEPTION 'ERROR: Last known partition table missing for parent table %. Unable to determine next partition in sequence.', v_row.parent_table;
          END IF;
      ELSE
          RAISE EXCEPTION 'ERROR: Last known partition missing from config table for parent table %.', p_parent_table;
      END IF;
  
      IF v_row.type = 'time-static' OR v_row.type = 'time-dynamic' THEN
          CASE
              WHEN v_row.part_interval = '15 mins' THEN
                  v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP) + 
                      '15min'::interval * floor(date_part('minute', CURRENT_TIMESTAMP) / 15.0);
              WHEN v_row.part_interval = '30 mins' THEN
                  v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP) + 
                      '30min'::interval * floor(date_part('minute', CURRENT_TIMESTAMP) / 30.0);
              WHEN v_row.part_interval = '1 hour' THEN
                  v_current_partition_timestamp := date_trunc('hour', CURRENT_TIMESTAMP);
               WHEN v_row.part_interval = '1 day' THEN
                  v_current_partition_timestamp := date_trunc('day', CURRENT_TIMESTAMP);
              WHEN v_row.part_interval = '1 week' THEN
                  v_current_partition_timestamp := date_trunc('week', CURRENT_TIMESTAMP);
              WHEN v_row.part_interval = '1 month' THEN
                  v_current_partition_timestamp := date_trunc('month', CURRENT_TIMESTAMP);
              WHEN v_row.part_interval = '3 months' THEN
                  v_current_partition_timestamp := date_trunc('quarter', CURRENT_TIMESTAMP);
              WHEN v_row.part_interval = '1 year' THEN
                  v_current_partition_timestamp := date_trunc('year', CURRENT_TIMESTAMP);
          END CASE;
      ELSIF v_row.type = 'time-custom' THEN
          SELECT child_table INTO v_current_partition FROM @extschema@.custom_time_partitions 
              WHERE parent_table = v_row.parent_table AND partition_range @> CURRENT_TIMESTAMP;
          IF v_current_partition IS NULL THEN
              RAISE EXCEPTION 'Current time partition missing from custom_time_partitions config table for table % and timestamp %',
                   CURRENT_TIMESTAMP, v_row.parent_table;
          END IF;
          v_time_position := (length(v_current_partition) - position('p_' in reverse(v_current_partition))) + 2;
          v_current_partition_timestamp := to_timestamp(substring(v_current_partition from v_time_position), v_row.datetime_string);
      END IF;
  
      v_time_position := (length(v_row.last_partition) - position('p_' in reverse(v_row.last_partition))) + 2;
      IF v_row.part_interval <> '3 months' OR (v_row.part_interval = '3 months' AND v_row.type = 'time-custom') THEN
         v_last_partition_timestamp := to_timestamp(substring(v_row.last_partition from v_time_position), v_row.datetime_string);
      ELSE
          -- to_timestamp doesn't recognize 'Q' date string formater. Handle it
          v_year := split_part(substring(v_row.last_partition from v_time_position), 'q', 1);
          v_quarter := split_part(substring(v_row.last_partition from v_time_position), 'q', 2);
          CASE
              WHEN v_quarter = '1' THEN
                  v_last_partition_timestamp := to_timestamp(v_year || '-01-01', 'YYYY-MM-DD');
              WHEN v_quarter = '2' THEN
                  v_last_partition_timestamp := to_timestamp(v_year || '-04-01', 'YYYY-MM-DD');
              WHEN v_quarter = '3' THEN
                  v_last_partition_timestamp := to_timestamp(v_year || '-07-01', 'YYYY-MM-DD');
              WHEN v_quarter = '4' THEN
                  v_last_partition_timestamp := to_timestamp(v_year || '-10-01', 'YYYY-MM-DD');
          END CASE;
      END IF;
      
      -- Check and see how many premade partitions there are.
      v_premade_count = round(EXTRACT('epoch' FROM age(v_last_partition_timestamp, v_current_partition_timestamp)) / EXTRACT('epoch' FROM v_row.part_interval::interval));
      v_next_partition_timestamp := v_last_partition_timestamp;
      -- Loop premaking until config setting is met. Allows it to catch up if it fell behind or if premake changed.
      WHILE v_premade_count < v_row.premake LOOP
          BEGIN
              v_next_partition_timestamp := v_next_partition_timestamp + v_row.part_interval;
          EXCEPTION WHEN datetime_field_overflow THEN
              RAISE WARNING 'Attempted partition time interval is outside PostgreSQL''s supported time range. 
                  Child partition creation skipped for parent table %', v_row.parent_table;
              v_premade_count = v_row.premake; -- do this so it can exit the premake check loop and continue in the outer for loop 
              v_step_overflow_id := add_step(v_job_id, 'Attempted partition time interval is outside PostgreSQL''s supported time range.');
              PERFORM update_step(v_step_overflow_id, 'CRITICAL', 'Child partition creation skippd for parent table '||v_partition_time);
              CONTINUE;
          END;        
          
          v_last_partition := @extschema@.create_time_partition(v_row.parent_table, ARRAY[v_next_partition_timestamp]); 
          IF v_last_partition IS NOT NULL THEN
              UPDATE @extschema@.part_config SET last_partition = v_last_partition WHERE parent_table = v_row.parent_table;
          END IF;
  
          v_create_count := v_create_count + 1;
          IF v_row.type = 'time-static' THEN
              PERFORM @extschema@.create_time_function(v_row.parent_table);
          END IF;
  
          -- Manage additonal constraints if set
          PERFORM @extschema@.apply_constraints(v_row.parent_table);
          v_premade_count = round(EXTRACT('epoch' FROM age(v_next_partition_timestamp, v_current_partition_timestamp)) / EXTRACT('epoch' FROM v_row.part_interval::interval));
      END LOOP;
  
  END LOOP; -- end of creation loop
  
  -- Manage dropping old partitions if retention option is set
  FOR v_row IN 
      SELECT parent_table FROM @extschema@.part_config WHERE retention IS NOT NULL AND undo_in_progress = false AND 
          (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom')
  LOOP
      v_drop_count := v_drop_count + @extschema@.drop_partition_time(v_row.parent_table);   
  END LOOP; 
  FOR v_row IN 
      SELECT parent_table FROM @extschema@.part_config WHERE retention IS NOT NULL AND undo_in_progress = false AND (type = 'id-static' OR type = 'id-dynamic')
  LOOP
      v_drop_count := v_drop_count + @extschema@.drop_partition_id(v_row.parent_table);
  END LOOP; 
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Partition maintenance finished. '||v_create_count||' partitons made. '||v_drop_count||' partitions dropped.');
      IF v_step_overflow_id IS NOT NULL THEN
          PERFORM fail_job(v_job_id);
      ELSE
          PERFORM close_job(v_job_id);
      END IF;
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN RUN MAINTENANCE'')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to drop child tables from a time-based partition set.
   * Options to move table to different schema, drop only indexes or actually drop the table from the database.
   */
  CREATE OR REPLACE FUNCTION drop_partition_time(p_parent_table text, p_retention interval DEFAULT NULL, p_keep_table boolean DEFAULT NULL, p_keep_index boolean DEFAULT NULL, p_retention_schema text DEFAULT NULL) RETURNS int
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock                  boolean;
  v_child_table               text;
  v_datetime_string           text;
  v_drop_count                int := 0;
  v_index                     record;
  v_job_id                    bigint;
  v_jobmon                    boolean;
  v_jobmon_schema             text;
  v_old_search_path           text;
  v_part_interval             interval;
  v_partition_timestamp       timestamp;
  v_quarter                   text;
  v_retention                 interval;
  v_retention_keep_index      boolean;
  v_retention_keep_table      boolean;
  v_retention_schema          text;
  v_step_id                   bigint;
  v_time_position             int;
  v_type                      text;
  v_year                      text;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman drop_partition_time'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'drop_partition_time already running.';
      RETURN 0;
  END IF;
  
  -- Allow override of configuration options
  IF p_retention IS NULL THEN
      SELECT  
          type
          , part_interval::interval
          , retention::interval
          , retention_keep_table
          , retention_keep_index
          , datetime_string
          , retention_schema
          , jobmon
      INTO
          v_type
          , v_part_interval
          , v_retention
          , v_retention_keep_table
          , v_retention_keep_index
          , v_datetime_string
          , v_retention_schema
          , v_jobmon
      FROM @extschema@.part_config 
      WHERE parent_table = p_parent_table 
      AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom') 
      AND retention IS NOT NULL;
      
      IF v_part_interval IS NULL THEN
          RAISE EXCEPTION 'Configuration for given parent table with a retention period not found: %', p_parent_table;
      END IF;
  ELSE
      SELECT  
          type
          , part_interval::interval
          , retention_keep_table
          , retention_keep_index
          , datetime_string
          , retention_schema
          , jobmon
      INTO
          v_type
          , v_part_interval
          , v_retention_keep_table
          , v_retention_keep_index
          , v_datetime_string
          , v_retention_schema
          , v_jobmon
      FROM @extschema@.part_config 
      WHERE parent_table = p_parent_table 
      AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom'); 
      v_retention := p_retention;
      
      IF v_part_interval IS NULL THEN
          RAISE EXCEPTION 'Configuration for given parent table not found: %', p_parent_table;
      END IF;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF p_keep_table IS NOT NULL THEN
      v_retention_keep_table = p_keep_table;
  END IF;
  IF p_keep_index IS NOT NULL THEN
      v_retention_keep_index = p_keep_index;
  END IF;
  IF p_retention_schema IS NOT NULL THEN
      v_retention_schema = p_retention_schema;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN DROP TIME PARTITION: '|| p_parent_table);
  END IF;
  
  -- Loop through child tables of the given parent
  FOR v_child_table IN 
      SELECT n.nspname||'.'||c.relname FROM pg_inherits i join pg_class c ON i.inhrelid = c.oid join pg_namespace n ON c.relnamespace = n.oid WHERE i.inhparent::regclass = p_parent_table::regclass ORDER BY i.inhrelid ASC
  LOOP
      -- pull out datetime portion of partition's tablename to make the next one
      v_time_position := (length(v_child_table) - position('p_' in reverse(v_child_table))) + 2;
      IF v_part_interval <> '3 months' OR (v_part_interval = '3 months' AND v_type = 'time-custom') THEN
          v_partition_timestamp := to_timestamp(substring(v_child_table from v_time_position), v_datetime_string);
      ELSE
          -- to_timestamp doesn't recognize 'Q' date string formater. Handle it
          v_year := split_part(substring(v_child_table from v_time_position), 'q', 1);
          v_quarter := split_part(substring(v_child_table from v_time_position), 'q', 2);
          CASE
              WHEN v_quarter = '1' THEN
                  v_partition_timestamp := to_timestamp(v_year || '-01-01', 'YYYY-MM-DD');
              WHEN v_quarter = '2' THEN
                  v_partition_timestamp := to_timestamp(v_year || '-04-01', 'YYYY-MM-DD');
              WHEN v_quarter = '3' THEN
                  v_partition_timestamp := to_timestamp(v_year || '-07-01', 'YYYY-MM-DD');
              WHEN v_quarter = '4' THEN
                  v_partition_timestamp := to_timestamp(v_year || '-10-01', 'YYYY-MM-DD');
          END CASE;
      END IF;
  
      -- Add one interval since partition names contain the start of the constraint period
      IF v_retention < (CURRENT_TIMESTAMP - (v_partition_timestamp + v_part_interval)) THEN
          IF v_jobmon_schema IS NOT NULL THEN
              v_step_id := add_step(v_job_id, 'Uninherit table '||v_child_table||' from '||p_parent_table);
          END IF;
          EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Done');
          END IF;
          IF v_retention_schema IS NULL THEN
              IF v_retention_keep_table = false THEN
                  IF v_jobmon_schema IS NOT NULL THEN
                      v_step_id := add_step(v_job_id, 'Drop table '||v_child_table);
                  END IF;
                  EXECUTE 'DROP TABLE '||v_child_table;
                  IF v_jobmon_schema IS NOT NULL THEN
                      PERFORM update_step(v_step_id, 'OK', 'Done');
                  END IF;
              ELSIF v_retention_keep_index = false THEN
                  FOR v_index IN 
                      SELECT i.indexrelid::regclass AS name
                      , c.conname
                      FROM pg_catalog.pg_index i
                      LEFT JOIN pg_catalog.pg_constraint c ON i.indexrelid = c.conindid 
                      WHERE i.indrelid = v_child_table::regclass
                  LOOP
                      IF v_jobmon_schema IS NOT NULL THEN
                          v_step_id := add_step(v_job_id, 'Drop index '||v_index.name||' from '||v_child_table);
                      END IF;
                      IF v_index.conname IS NOT NULL THEN
                          EXECUTE 'ALTER TABLE '||v_child_table||' DROP CONSTRAINT '||v_index.conname;
                      ELSE
                          EXECUTE 'DROP INDEX '||v_index.name;
                      END IF;
                      IF v_jobmon_schema IS NOT NULL THEN
                          PERFORM update_step(v_step_id, 'OK', 'Done');
                      END IF;
                  END LOOP;
              END IF;
          ELSE -- Move to new schema
              IF v_jobmon_schema IS NOT NULL THEN
                  v_step_id := add_step(v_job_id, 'Moving table '||v_child_table||' to schema '||v_retention_schema);
              END IF;
  
              EXECUTE 'ALTER TABLE '||v_child_table||' SET SCHEMA '||v_retention_schema; 
              
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Done');
              END IF;
          END IF; -- End retention schema if
          
          v_drop_count := v_drop_count + 1;
      END IF; -- End retention check IF
  
  END LOOP; -- End child table loop
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Finished partition drop maintenance');
      PERFORM update_step(v_step_id, 'OK', v_drop_count||' partitions dropped.');
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  RETURN v_drop_count;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN DROP TIME PARTITION: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to undo time-based partitioning created by this extension
   */
  CREATE OR REPLACE FUNCTION undo_partition_time(p_parent_table text, p_batch_count int DEFAULT 1, p_batch_interval interval DEFAULT NULL, p_keep_table boolean DEFAULT true) RETURNS bigint
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock              boolean;
  v_batch_loop_count      int := 0;
  v_child_min             timestamptz;
  v_child_loop_total      bigint := 0;
  v_child_table           text;
  v_control               text;
  v_function_name         text;
  v_inner_loop_count      int;
  v_job_id                bigint;
  v_jobmon                boolean;
  v_jobmon_schema         text;
  v_move_sql              text;
  v_old_search_path       text;
  v_parent_schema         text;
  v_parent_tablename      text;
  v_part_interval         interval;
  v_row                   record;
  v_rowcount              bigint;
  v_step_id               bigint;
  v_total                 bigint := 0;
  v_trig_name             text;
  v_undo_count            int := 0;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman undo_time_partition'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'undo_time_partition already running.';
      RETURN 0;
  END IF;
  
  SELECT part_interval::interval
      , control
      , jobmon
  INTO v_part_interval
      , v_control
      , v_jobmon
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table 
  AND (type = 'time-static' OR type = 'time-dynamic' OR type = 'time-custom');
  
  IF v_part_interval IS NULL THEN
      RAISE EXCEPTION 'Configuration for given parent table not found: %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN UNDO PARTITIONING: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Undoing partitioning for table '||p_parent_table);
  END IF;
  
  IF p_batch_interval IS NULL THEN
      p_batch_interval := v_part_interval;
  END IF;
  
  -- Stops new time partitons from being made as well as stopping child tables from being dropped if they were configured with a retention period.
  UPDATE @extschema@.part_config SET undo_in_progress = true WHERE parent_table = p_parent_table;
  -- Stop data going into child tables.
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  v_trig_name := @extschema@.check_name_length(p_object_name := v_parent_tablename, p_suffix := '_part_trig'); 
  v_function_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, '_part_trig_func', FALSE);
  EXECUTE 'DROP TRIGGER IF EXISTS '||v_trig_name||' ON '||p_parent_table;
  EXECUTE 'DROP FUNCTION IF EXISTS '||v_function_name||'()';
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Stopped partition creation process. Removed trigger & trigger function');
  END IF;
  
  <<outer_child_loop>>
  WHILE v_batch_loop_count < p_batch_count LOOP 
      SELECT n.nspname||'.'||c.relname INTO v_child_table
      FROM pg_inherits i 
      JOIN pg_class c ON i.inhrelid = c.oid 
      JOIN pg_namespace n ON c.relnamespace = n.oid 
      WHERE i.inhparent::regclass = p_parent_table::regclass 
      ORDER BY i.inhrelid ASC;
  
      EXIT WHEN v_child_table IS NULL;
  
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing child partition: '||v_child_table);
      END IF;
  
      EXECUTE 'SELECT min('||v_control||') FROM '||v_child_table INTO v_child_min;
      IF v_child_min IS NULL THEN
          -- No rows left in this child table. Remove from partition set.
          EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
          IF p_keep_table = false THEN
              EXECUTE 'DROP TABLE '||v_child_table;
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table DROPPED. Moved '||v_child_loop_total||' rows to parent');
              END IF;
          ELSE
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table UNINHERITED, not DROPPED. Moved '||v_child_loop_total||' rows to parent');
              END IF;
          END IF;
          v_undo_count := v_undo_count + 1;
          CONTINUE outer_child_loop;
      END IF;
      v_inner_loop_count := 1;
      v_child_loop_total := 0;
      <<inner_child_loop>>
      LOOP
          -- Get everything from the current child minimum up to the multiples of the given interval
          v_move_sql := 'WITH move_data AS (DELETE FROM '||v_child_table||
                  ' WHERE '||v_control||' <= '||quote_literal(v_child_min + (p_batch_interval * v_inner_loop_count))||' RETURNING *)
              INSERT INTO '||p_parent_table||' SELECT * FROM move_data';
          EXECUTE v_move_sql;
          GET DIAGNOSTICS v_rowcount = ROW_COUNT;
          v_total := v_total + v_rowcount;
          v_child_loop_total := v_child_loop_total + v_rowcount;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Moved '||v_child_loop_total||' rows to parent.');
          END IF;
          EXIT inner_child_loop WHEN v_rowcount = 0; -- exit before loop incr if table is empty
          v_inner_loop_count := v_inner_loop_count + 1;
          v_batch_loop_count := v_batch_loop_count + 1;
          EXIT outer_child_loop WHEN v_batch_loop_count >= p_batch_count; -- Exit outer FOR loop if p_batch_count is reached
      END LOOP inner_child_loop;
  END LOOP outer_child_loop;
  
  IF v_batch_loop_count < p_batch_count THEN
      -- FOR loop never ran, so there's no child tables left.
      DELETE FROM @extschema@.part_config WHERE parent_table = p_parent_table;
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing config from pg_partman');
          PERFORM update_step(v_step_id, 'OK', 'Done');
      END IF;
  END IF;
  
  RAISE NOTICE 'Copied % row(s) to the parent. Removed % partitions.', v_total, v_undo_count;
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Final stats');
      PERFORM update_step(v_step_id, 'OK', 'Copied '||v_total||' row(s) to the parent. Removed '||v_undo_count||' partitions.');
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  RETURN v_total;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN UNDO PARTITIONING: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to create id partitions
   */
  CREATE FUNCTION create_id_partition (p_parent_table text, p_partition_ids bigint[]) RETURNS text
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_all               text[] := ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER'];
  v_control           text;
  v_grantees          text[];
  v_job_id            bigint;
  v_jobmon            boolean;
  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_parent_tablespace text;
  v_part_interval     bigint;
  v_partition_name    text;
  v_revoke            text[];
  v_step_id           bigint;
  v_tablename         text;
  v_id                bigint;
  
  BEGIN
  
  SELECT control
      , part_interval
      , jobmon
  INTO v_control
      , v_part_interval
      , v_jobmon
  FROM @extschema@.part_config
  WHERE parent_table = p_parent_table
  AND (type = 'id-static' OR type = 'id-dynamic');
  
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  SELECT tableowner, schemaname, tablename, tablespace INTO v_parent_owner, v_parent_schema, v_parent_tablename, v_parent_tablespace 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 + v_part_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;
      IF v_parent_tablespace IS NOT NULL THEN
          EXECUTE 'ALTER TABLE '||v_partition_name||' SET TABLESPACE '||v_parent_tablespace;
      END IF;
      EXECUTE 'ALTER TABLE '||v_partition_name||' ADD CONSTRAINT '||v_tablename||'_partition_check 
          CHECK ('||v_control||'>='||quote_literal(v_id)||' AND '||v_control||'<'||quote_literal(v_id + v_part_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
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE TABLE: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Create the trigger function for the parent table of an id-based partition set
   */
  CREATE OR REPLACE FUNCTION create_id_function(p_parent_table text) RETURNS void
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_control                       text;
  v_count                         int;
  v_current_partition_name        text;
  v_current_partition_id          bigint;
  v_datetime_string               text;
  v_final_partition_id            bigint;
  v_function_name                 text;
  v_job_id                        bigint;
  v_jobmon                        text;
  v_jobmon_schema                 text;
  v_last_partition                text;
  v_max                           bigint;
  v_next_partition_id             bigint;
  v_next_partition_name           text;
  v_old_search_path               text;
  v_parent_schema                 text;
  v_parent_tablename              text;
  v_part_interval                 bigint;
  v_premake                       int;
  v_prev_partition_id             bigint;
  v_prev_partition_name           text;
  v_step_id                       bigint;
  v_trig_func                     text;
  v_type                          text;
  
  BEGIN
  
  SELECT type
      , part_interval::bigint
      , control
      , premake
      , last_partition
      , jobmon
  INTO v_type
      , v_part_interval
      , v_control
      , v_premake
      , v_last_partition
      , v_jobmon
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table
  AND (type = 'id-static' OR type = 'id-dynamic');
  
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      SELECT nspname INTO v_jobmon_schema FROM pg_catalog.pg_namespace n, pg_catalog.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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN CREATE FUNCTION: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Creating partition function for table '||p_parent_table);
  END IF;
  
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_catalog.pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  v_function_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, '_part_trig_func', FALSE);
  
  IF v_type = 'id-static' THEN
      EXECUTE 'SELECT COALESCE(max('||v_control||'), 0) FROM '||p_parent_table INTO v_max;
      v_current_partition_id = v_max - (v_max % v_part_interval);
      v_next_partition_id := v_current_partition_id + v_part_interval;
      v_current_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_current_partition_id::text, TRUE);
  
      v_trig_func := 'CREATE OR REPLACE FUNCTION '||v_function_name||'() RETURNS trigger LANGUAGE plpgsql AS $t$ 
          DECLARE
              v_current_partition_id  bigint;
              v_last_partition        text := '||quote_literal(v_last_partition)||';
              v_id_position           int;
              v_next_partition_id     bigint;
              v_next_partition_name   text;         
          BEGIN
          IF TG_OP = ''INSERT'' THEN 
              IF NEW.'||v_control||' >= '||v_current_partition_id||' AND NEW.'||v_control||' < '||v_next_partition_id|| ' THEN 
                  INSERT INTO '||v_current_partition_name||' VALUES (NEW.*); ';
  
          FOR i IN 1..v_premake LOOP
              v_prev_partition_id := v_current_partition_id - (v_part_interval * i);
              v_next_partition_id := v_current_partition_id + (v_part_interval * i);
              v_final_partition_id := v_next_partition_id + v_part_interval;
              v_prev_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_prev_partition_id::text, TRUE);
              v_next_partition_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_next_partition_id::text, TRUE);
              
              -- Check that child table exist before making a rule to insert to them.
              -- Handles edge case of changing premake immediately after running create_parent(). 
              SELECT count(*) INTO v_count FROM pg_catalog.pg_tables WHERE schemaname ||'.'||tablename = v_prev_partition_name;
              IF v_count > 0 THEN
                  -- Only handle previous partitions if they're starting above zero
                  IF v_prev_partition_id >= 0 THEN
                      v_trig_func := v_trig_func ||'
              ELSIF NEW.'||v_control||' >= '||v_prev_partition_id||' AND NEW.'||v_control||' < '||v_prev_partition_id + v_part_interval|| ' THEN 
                  INSERT INTO '||v_prev_partition_name||' VALUES (NEW.*); ';
                  END IF;
              END IF;
              
              SELECT count(*) INTO v_count FROM pg_catalog.pg_tables WHERE schemaname ||'.'||tablename = v_next_partition_name;
              IF v_count > 0 THEN
                  v_trig_func := v_trig_func ||'
              ELSIF NEW.'||v_control||' >= '||v_next_partition_id||' AND NEW.'||v_control||' < '||v_final_partition_id|| ' THEN 
                  INSERT INTO '||v_next_partition_name||' VALUES (NEW.*); ';
              END IF;
          END LOOP;
  
          v_trig_func := v_trig_func ||'
              ELSE
                  RETURN NEW;
              END IF;
              v_current_partition_id := NEW.'||v_control||' - (NEW.'||v_control||' % '||v_part_interval||');
              IF (NEW.'||v_control||' % '||v_part_interval||') > ('||v_part_interval||' / 2) THEN
                  v_id_position := (length(v_last_partition) - position(''p_'' in reverse(v_last_partition))) + 2;
                  v_next_partition_id := (substring(v_last_partition from v_id_position)::bigint) + '||v_part_interval||';
                  WHILE ((v_next_partition_id - v_current_partition_id) / '||v_part_interval||') <= '||v_premake||' LOOP 
                      v_next_partition_name := @extschema@.create_id_partition('||quote_literal(p_parent_table)||', ARRAY[v_next_partition_id]);
                      UPDATE @extschema@.part_config SET last_partition = v_next_partition_name WHERE parent_table = '||quote_literal(p_parent_table)||';
                      PERFORM @extschema@.create_id_function('||quote_literal(p_parent_table)||');
                      PERFORM @extschema@.apply_constraints('||quote_literal(p_parent_table)||');
                      v_next_partition_id := v_next_partition_id + '||v_part_interval||';
                  END LOOP;
              END IF;
          END IF; 
          RETURN NULL; 
          END $t$;';
  
      EXECUTE v_trig_func;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Added function for current id interval: '||v_current_partition_id||' to '||v_final_partition_id-1);
      END IF;
  
  ELSIF v_type = 'id-dynamic' THEN
      -- The return inside the partition creation check is there to keep really high ID values from creating new partitions.
      v_trig_func := 'CREATE OR REPLACE FUNCTION '||v_function_name||'() RETURNS trigger LANGUAGE plpgsql AS $t$ 
          DECLARE
              v_count                     int;
              v_current_partition_id      bigint;
              v_current_partition_name    text;
              v_id_position               int;
              v_last_partition            text := '||quote_literal(v_last_partition)||';
              v_last_partition_id         bigint;
              v_next_partition_id         bigint;
              v_next_partition_name       text;   
          BEGIN 
          IF TG_OP = ''INSERT'' THEN 
              v_current_partition_id := NEW.'||v_control||' - (NEW.'||v_control||' % '||v_part_interval||');
              v_current_partition_name := @extschema@.check_name_length('''||v_parent_tablename||''', '''||v_parent_schema||''', v_current_partition_id::text, TRUE);
              IF (NEW.'||v_control||' % '||v_part_interval||') > ('||v_part_interval||' / 2) THEN
                  v_id_position := (length(v_last_partition) - position(''p_'' in reverse(v_last_partition))) + 2;
                  v_last_partition_id = substring(v_last_partition from v_id_position)::bigint;
                  v_next_partition_id := v_last_partition_id + '||v_part_interval||';
                  IF NEW.'||v_control||' >= v_next_partition_id THEN
                      RETURN NEW;
                  END IF;
                  WHILE ((v_next_partition_id - v_current_partition_id) / '||v_part_interval||') <= '||v_premake||' LOOP 
                      v_next_partition_name := @extschema@.create_id_partition('||quote_literal(p_parent_table)||', ARRAY[v_next_partition_id]);
                      IF v_next_partition_name IS NOT NULL THEN
                          UPDATE @extschema@.part_config SET last_partition = v_next_partition_name WHERE parent_table = '||quote_literal(p_parent_table)||';
                          PERFORM @extschema@.create_id_function('||quote_literal(p_parent_table)||');
                          PERFORM @extschema@.apply_constraints('||quote_literal(p_parent_table)||');
                      END IF;
                      v_next_partition_id := v_next_partition_id + '||v_part_interval||';
                  END LOOP;
              END IF;
              SELECT count(*) INTO v_count FROM pg_tables WHERE schemaname ||''.''|| tablename = v_current_partition_name;
              IF v_count > 0 THEN 
                  EXECUTE ''INSERT INTO ''||v_current_partition_name||'' VALUES($1.*)'' USING NEW;
              ELSE
                  RETURN NEW;
              END IF;
          END IF;
          
          RETURN NULL; 
          END $t$;';
  
      EXECUTE v_trig_func;
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Added function for dynamic id table: '||p_parent_table);
      END IF;
  
  ELSE
      RAISE EXCEPTION 'ERROR: Invalid id partitioning type given: %', v_type;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE FUNCTION: '||p_parent_table||')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''Partition function maintenance for table '||p_parent_table||' failed'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Populate the child table(s) of an id-based partition set with old data from the original parent
   */
  CREATE OR REPLACE FUNCTION partition_data_id(p_parent_table text, p_batch_count int DEFAULT 1, p_batch_interval int DEFAULT NULL, p_lock_wait numeric DEFAULT 0) RETURNS bigint
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_control                   text;
  v_last_partition_name       text;
  v_lock_iter                 int := 1;
  v_lock_obtained             boolean := FALSE;
  v_max_partition_id          bigint;
  v_min_control               bigint;
  v_min_partition_id          bigint;
  v_part_interval             bigint;
  v_partition_id              bigint[];
  v_rowcount                  bigint;
  v_sql                       text;
  v_total_rows                bigint := 0;
  v_type                      text;
  
  BEGIN
  
  SELECT type
      , part_interval::bigint
      , control
  INTO v_type
      , v_part_interval
      , v_control
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table
  AND (type = 'id-static' OR type = 'id-dynamic');
  IF NOT FOUND THEN
      RAISE EXCEPTION 'ERROR: no config found for %', p_parent_table;
  END IF;
  
  IF p_batch_interval IS NULL OR p_batch_interval > v_part_interval THEN
      p_batch_interval := v_part_interval;
  END IF;
  
  FOR i IN 1..p_batch_count LOOP
  
      EXECUTE 'SELECT min('||v_control||') FROM ONLY '||p_parent_table INTO v_min_control;
      IF v_min_control IS NULL THEN
          RETURN 0;
      END IF;
  
      v_min_partition_id = v_min_control - (v_min_control % v_part_interval);
  
      v_partition_id := ARRAY[v_min_partition_id];
  --    RAISE NOTICE 'v_partition_id: %',v_partition_id;
      IF (v_min_control + p_batch_interval) >= (v_min_partition_id + v_part_interval) THEN
          v_max_partition_id := v_min_partition_id + v_part_interval;
      ELSE
          v_max_partition_id := v_min_control + p_batch_interval;
      END IF;
  --    RAISE NOTICE 'v_max_partition_id: %',v_max_partition_id;
  
  -- do some locking with timeout, if required
      IF p_lock_wait > 0  THEN
          v_lock_iter := 0;
          WHILE v_lock_iter <= 5 LOOP
              v_lock_iter := v_lock_iter + 1;
              BEGIN
                  v_sql := 'SELECT * FROM ONLY ' || p_parent_table ||
                  ' WHERE '||v_control||' >= '||quote_literal(v_min_control)||
                  ' AND '||v_control||' < '||quote_literal(v_max_partition_id)
                  ||' FOR UPDATE NOWAIT';
                  EXECUTE v_sql;
                  v_lock_obtained := TRUE;
              EXCEPTION
                  WHEN lock_not_available THEN
                      PERFORM pg_sleep( p_lock_wait / 5.0 );
                      CONTINUE;
              END;
              EXIT WHEN v_lock_obtained;
          END LOOP;
          IF NOT v_lock_obtained THEN
             RETURN -1;
          END IF;
      END IF;
  
      v_last_partition_name := @extschema@.create_id_partition(p_parent_table, v_partition_id);
  
      EXECUTE 'WITH partition_data AS (
          DELETE FROM ONLY '||p_parent_table||' WHERE '||v_control||' >= '||v_min_control||
              ' AND '||v_control||' < '||v_max_partition_id||' RETURNING *)
          INSERT INTO '||v_last_partition_name||' SELECT * FROM partition_data';        
  
      GET DIAGNOSTICS v_rowcount = ROW_COUNT;
      v_total_rows := v_total_rows + v_rowcount;
      IF v_rowcount = 0 THEN
          EXIT;
      END IF;
  
  END LOOP; 
  
  IF v_type = 'id-static' THEN
          PERFORM @extschema@.create_id_function(p_parent_table);
  END IF;    
  
  RETURN v_total_rows;
  
  END
  $$;
  
  
  /*
   * Function to drop child tables from an id-based partition set. 
   * Options to move table to different schema, drop only indexes or actually drop the table from the database.
   */
  CREATE OR REPLACE FUNCTION drop_partition_id(p_parent_table text, p_retention bigint DEFAULT NULL, p_keep_table boolean DEFAULT NULL, p_keep_index boolean DEFAULT NULL, p_retention_schema text DEFAULT NULL) RETURNS int
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock                  boolean;
  v_child_table               text;
  v_control                   text;
  v_drop_count                int := 0;
  v_id_position               int;
  v_index                     record;
  v_job_id                    bigint;
  v_jobmon                    boolean;
  v_jobmon_schema             text;
  v_max                       bigint;
  v_old_search_path           text;
  v_part_interval             bigint;
  v_partition_id              bigint;
  v_retention                 bigint;
  v_retention_keep_index      boolean;
  v_retention_keep_table      boolean;
  v_retention_schema          text;
  v_step_id                   bigint;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman drop_partition_id'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'drop_partition_id already running.';
      RETURN 0;
  END IF;
  
  -- Allow override of configuration options
  IF p_retention IS NULL THEN
      SELECT  
          part_interval::bigint
          , control
          , retention::bigint
          , retention_keep_table
          , retention_keep_index
          , retention_schema
          , jobmon
      INTO
          v_part_interval
          , v_control
          , v_retention
          , v_retention_keep_table
          , v_retention_keep_index
          , v_retention_schema
          , v_jobmon
      FROM @extschema@.part_config 
      WHERE parent_table = p_parent_table 
      AND (type = 'id-static' OR type = 'id-dynamic') 
      AND retention IS NOT NULL;
  
      IF v_part_interval IS NULL THEN
          RAISE EXCEPTION 'Configuration for given parent table with a retention period not found: %', p_parent_table;
      END IF;
  ELSE
       SELECT  
          part_interval::bigint
          , control
          , retention_keep_table
          , retention_keep_index
          , retention_schema
          , jobmon
      INTO
          v_part_interval
          , v_control
          , v_retention_keep_table
          , v_retention_keep_index
          , v_retention_schema
          , v_jobmon
      FROM @extschema@.part_config 
      WHERE parent_table = p_parent_table 
      AND (type = 'id-static' OR type = 'id-dynamic'); 
      v_retention := p_retention;
  
      IF v_part_interval IS NULL THEN
          RAISE EXCEPTION 'Configuration for given parent table not found: %', p_parent_table;
      END IF;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF p_keep_table IS NOT NULL THEN
      v_retention_keep_table = p_keep_table;
  END IF;
  IF p_keep_index IS NOT NULL THEN
      v_retention_keep_index = p_keep_index;
  END IF;
  IF p_retention_schema IS NOT NULL THEN
      v_retention_schema = p_retention_schema;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN DROP ID PARTITION: '|| p_parent_table);
  END IF;
  
  EXECUTE 'SELECT max('||v_control||') FROM '||p_parent_table INTO v_max;
  
  -- Loop through child tables of the given parent
  FOR v_child_table IN 
      SELECT n.nspname||'.'||c.relname FROM pg_inherits i join pg_class c ON i.inhrelid = c.oid join pg_namespace n ON c.relnamespace = n.oid WHERE i.inhparent::regclass = p_parent_table::regclass ORDER BY i.inhrelid ASC
  LOOP
      v_id_position := (length(v_child_table) - position('p_' in reverse(v_child_table))) + 2;
      v_partition_id := substring(v_child_table from v_id_position)::bigint;
  
      -- Add one interval since partition names contain the start of the constraint period
      IF v_retention <= (v_max - (v_partition_id + v_part_interval)) THEN
          IF v_jobmon_schema IS NOT NULL THEN
              v_step_id := add_step(v_job_id, 'Uninherit table '||v_child_table||' from '||p_parent_table);
          END IF;
          EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Done');
          END IF;
          IF v_retention_schema IS NULL THEN
              IF v_retention_keep_table = false THEN
                  IF v_jobmon_schema IS NOT NULL THEN
                      v_step_id := add_step(v_job_id, 'Drop table '||v_child_table);
                  END IF;
                  EXECUTE 'DROP TABLE '||v_child_table;
                  IF v_jobmon_schema IS NOT NULL THEN
                      PERFORM update_step(v_step_id, 'OK', 'Done');
                  END IF;
              ELSIF v_retention_keep_index = false THEN
                  FOR v_index IN 
                      SELECT i.indexrelid::regclass AS name
                      , c.conname
                      FROM pg_catalog.pg_index i
                      LEFT JOIN pg_catalog.pg_constraint c ON i.indexrelid = c.conindid 
                      WHERE i.indrelid = v_child_table::regclass
                  LOOP
                      IF v_jobmon_schema IS NOT NULL THEN
                          v_step_id := add_step(v_job_id, 'Drop index '||v_index.name||' from '||v_child_table);
                      END IF;
                      IF v_index.conname IS NOT NULL THEN
                          EXECUTE 'ALTER TABLE '||v_child_table||' DROP CONSTRAINT '||v_index.conname;
                      ELSE
                          EXECUTE 'DROP INDEX '||v_index.name;
                      END IF;
                      IF v_jobmon_schema IS NOT NULL THEN
                          PERFORM update_step(v_step_id, 'OK', 'Done');
                      END IF;
                  END LOOP;
              END IF;
          ELSE -- Move to new schema
              IF v_jobmon_schema IS NOT NULL THEN
                  v_step_id := add_step(v_job_id, 'Moving table '||v_child_table||' to schema '||v_retention_schema);
              END IF;
  
              EXECUTE 'ALTER TABLE '||v_child_table||' SET SCHEMA '||v_retention_schema; 
              
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Done');
              END IF;
          END IF; -- End retention schema if
          
          v_drop_count := v_drop_count + 1;
      END IF; -- End retention check IF
  
  END LOOP; -- End child table loop
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Finished partition drop maintenance');
      PERFORM update_step(v_step_id, 'OK', v_drop_count||' partitions dropped.');
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  RETURN v_drop_count;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN DROP ID PARTITION: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /* 
   * Apply constraints managed by partman extension
   */
  CREATE OR REPLACE FUNCTION apply_constraints(p_parent_table text, p_child_table text DEFAULT NULL, p_debug BOOLEAN DEFAULT FALSE) RETURNS void
      LANGUAGE plpgsql
      AS $$
  DECLARE
  
  v_child_table                   text;
  v_child_tablename               text;
  v_col                           text;
  v_constraint_cols               text[];
  v_constraint_col_type           text;
  v_constraint_name               text;
  v_datetime_string               text;
  v_existing_constraint_name      text;
  v_job_id                        bigint;
  v_jobmon                        boolean;
  v_jobmon_schema                 text;
  v_last_partition                text;
  v_last_partition_id             int; 
  v_last_partition_timestamp      timestamp;
  v_constraint_values             record;
  v_old_search_path               text;
  v_parent_schema                 text;
  v_parent_tablename              text;
  v_part_interval                 text;
  v_partition_suffix              text;
  v_premake                       int;
  v_sql                           text;
  v_step_id                       bigint;
  v_suffix_position               int;
  v_type                          text;
  
  BEGIN
  
  SELECT type
      , part_interval
      , premake
      , datetime_string
      , last_partition
      , constraint_cols
      , jobmon
  INTO v_type
      , v_part_interval
      , v_premake
      , v_datetime_string
      , v_last_partition
      , v_constraint_cols
      , v_jobmon
  FROM @extschema@.part_config
  WHERE parent_table = p_parent_table;
  
  IF v_constraint_cols IS NULL THEN
      IF p_debug THEN
          RAISE NOTICE 'Given parent table (%) not set up for constraint management (constraint_cols is NULL)', p_parent_table;
      END IF;
      -- Returns silently to allow this function to be simply called by maintenance processes without having to check if config options are set.
      RETURN;
  END IF;
  
  IF v_jobmon THEN
      SELECT nspname INTO v_jobmon_schema FROM pg_catalog.pg_namespace n, pg_catalog.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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN CREATE CONSTRAINT: '||p_parent_table);
  END IF;
  
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  
  -- If p_child_table is null, figure out the partition that is the one right before the premake value backwards.
  IF p_child_table IS NULL THEN
      
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Automatically determining most recent child on which to apply constraints');
      END IF;
  
      v_suffix_position := (length(v_last_partition) - position('p_' in reverse(v_last_partition))) + 2;
  
      IF v_type IN ('time-static', 'time-dynamic') THEN
          v_last_partition_timestamp := to_timestamp(substring(v_last_partition from v_suffix_position), v_datetime_string);
          v_partition_suffix := to_char(v_last_partition_timestamp - (v_part_interval::interval * ((v_premake * 2)+1) ), v_datetime_string);
      ELSIF v_type IN ('id-static', 'id-dynamic') THEN
          v_last_partition_id := substring(v_last_partition from v_suffix_position)::int;
          v_partition_suffix := (v_last_partition_id - (v_part_interval::int * ((v_premake * 2)+1) ))::text; 
      END IF;
      
      v_child_table := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, v_partition_suffix, TRUE);
  
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Target child table: '||v_child_table);
      END IF;
  ELSE
      v_child_table := p_child_table;
  END IF;
      
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Checking if target child table exists');
  END IF;
  
  SELECT tablename INTO v_child_tablename FROM pg_catalog.pg_tables WHERE schemaname ||'.'|| tablename = v_child_table;
  IF v_child_tablename IS NULL THEN
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'NOTICE', 'Target child table ('||v_child_table||') does not exist. Skipping constraint creation.');
          PERFORM close_job(v_job_id);
          EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
      END IF;
      IF p_debug THEN
          RAISE NOTICE 'Target child table (%) does not exist. Skipping constraint creation.', v_child_table;
      END IF;
      RETURN;
  ELSE
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'OK', 'Done');
      END IF;
  END IF;
  
  FOREACH v_col IN ARRAY v_constraint_cols
  LOOP
      SELECT c.conname
      INTO v_existing_constraint_name
      FROM pg_catalog.pg_constraint c 
          JOIN pg_catalog.pg_attribute a ON c.conrelid = a.attrelid 
      WHERE conrelid = v_child_table::regclass 
          AND c.conname LIKE 'partmanconstr_%'
          AND c.contype = 'c' 
          AND a.attname = v_col
          AND ARRAY[a.attnum] <@ c.conkey 
          AND a.attisdropped = false;
  
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Applying new constraint on column: '||v_col);
      END IF;
  
      IF v_existing_constraint_name IS NOT NULL THEN
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'NOTICE', 'Partman managed constraint already exists on this table ('||v_child_table||') and column ('||v_col||'). Skipping creation.');
          END IF;
          RAISE WARNING 'Partman managed constraint already exists on this table (%) and column (%). Skipping creation.', v_child_table, v_col ;
          CONTINUE;
      END IF;
  
      -- Ensure column name gets put on end of constraint name to help avoid naming conflicts 
      v_constraint_name := @extschema@.check_name_length('partmanconstr_'||v_child_tablename, p_suffix := '_'||v_col);
  
      EXECUTE 'SELECT min('||v_col||')::text AS min, max('||v_col||')::text AS max FROM '||v_child_table INTO v_constraint_values;
  
      IF v_constraint_values IS NOT NULL THEN
          v_sql := concat('ALTER TABLE ', v_child_table, ' ADD CONSTRAINT ', v_constraint_name
              , ' CHECK (', v_col, ' >= ', quote_literal(v_constraint_values.min), ' AND '
              , v_col, ' <= ', quote_literal(v_constraint_values.max), ')' );
          IF p_debug THEN
              RAISE NOTICE 'Constraint creation query: %', v_sql;
          END IF;
          EXECUTE v_sql;
  
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'New constraint created: '||v_sql);
          END IF;
      ELSE
          IF p_debug THEN
              RAISE NOTICE 'Given column (%) contains all NULLs. No constraint created', v_col;
          END IF;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'NOTICE', 'Given column ('||v_col||') contains all NULLs. No constraint created');
          END IF;
      END IF;
  
  END LOOP;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN CREATE CONSTRAINT: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Drop constraints managed by pg_partman
   */
  CREATE OR REPLACE FUNCTION drop_constraints(p_parent_table text, p_child_table text, p_debug boolean DEFAULT false) RETURNS void
      LANGUAGE plpgsql
      AS $$
  DECLARE
  
  v_col                           text;
  v_constraint_cols               text[]; 
  v_existing_constraint_name      text;
  v_exists                        boolean := FALSE;
  v_job_id                        bigint;
  v_jobmon                        boolean;
  v_jobmon_schema                 text;
  v_old_search_path               text;
  v_sql                           text;
  v_step_id                       bigint;
  
  BEGIN
  
  SELECT constraint_cols 
      , jobmon
  INTO v_constraint_cols 
      , v_jobmon
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table;
  
  IF v_constraint_cols IS NULL THEN
      RAISE EXCEPTION 'Given parent table (%) not set up for constraint management (constraint_cols is NULL)', p_parent_table;
  END IF;
  
  IF v_jobmon THEN 
      SELECT nspname INTO v_jobmon_schema FROM pg_catalog.pg_namespace n, pg_catalog.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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN DROP CONSTRAINT: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Entering constraint drop loop');
      PERFORM update_step(v_step_id, 'OK', 'Done');
  END IF;
  
  
  FOREACH v_col IN ARRAY v_constraint_cols
  LOOP
  
      SELECT c.conname
      INTO v_existing_constraint_name
      FROM pg_catalog.pg_constraint c 
          JOIN pg_catalog.pg_attribute a ON c.conrelid = a.attrelid 
      WHERE conrelid = p_child_table::regclass 
          AND c.conname LIKE 'partmanconstr_%'
          AND c.contype = 'c' 
          AND a.attname = v_col
          AND ARRAY[a.attnum] <@ c.conkey 
          AND a.attisdropped = false;
  
      IF v_existing_constraint_name IS NOT NULL THEN
          v_exists := TRUE;
          IF v_jobmon_schema IS NOT NULL THEN
              v_step_id := add_step(v_job_id, 'Dropping constraint on column: '||v_col);
          END IF;
          v_sql := 'ALTER TABLE '||p_child_table||' DROP CONSTRAINT '||v_existing_constraint_name;
          IF p_debug THEN
              RAISE NOTICE 'Constraint drop query: %', v_sql;
          END IF;
          EXECUTE v_sql;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Drop constraint query: '||v_sql);
          END IF;
      END IF;
  
  END LOOP;
  
  IF v_jobmon_schema IS NOT NULL AND v_exists IS FALSE THEN
      v_step_id := add_step(v_job_id, 'No constraints found to drop on child table: '||p_child_table);
      PERFORM update_step(v_step_id, 'OK', 'Done');
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN DROP CONSTRAINT: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to re-apply ownership & privileges on all child tables in a partition set using parent table as reference
   */
  CREATE OR REPLACE FUNCTION reapply_privileges(p_parent_table text) RETURNS void
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_all               text[] := ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER'];
  v_child_owner       text;
  v_child_table       text;
  v_child_grant       record;
  v_grant             text;
  v_grantees          text[];
  v_job_id            bigint;
  v_jobmon            boolean;
  v_jobmon_schema     text;
  v_match             boolean;
  v_old_search_path   text;
  v_parent_owner      text;
  v_owner_sql         text;
  v_revoke            text[];
  v_parent_grant      record;
  v_sql               text;
  v_step_id           bigint;
  
  BEGIN
  
  SELECT jobmon INTO v_jobmon FROM @extschema@.part_config WHERE parent_table = p_parent_table;
  IF v_jobmon IS NULL THEN
      RAISE EXCEPTION 'Given table is not managed by this extention: %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN RE-APPLYING PRIVILEGES TO ALL CHILD TABLES OF: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Setting new child table privileges');
  END IF;
  
  SELECT tableowner INTO v_parent_owner FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  
  FOR v_child_table IN 
      SELECT n.nspname||'.'||c.relname FROM pg_inherits i join pg_class c ON i.inhrelid = c.oid join pg_namespace n ON c.relnamespace = n.oid WHERE i.inhparent::regclass = p_parent_table::regclass ORDER BY i.inhrelid ASC
  LOOP
      IF v_jobmon_schema IS NOT NULL THEN
          PERFORM update_step(v_step_id, 'PENDING', 'Currently on child partition in ascending order: '||v_child_table);
      END IF;
      v_grantees := NULL;
      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
          -- Compare parent & child grants. Don't re-apply if it already exists
          v_match := false;
          FOR v_child_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 = v_child_table
              GROUP BY grantee 
          LOOP
              IF v_parent_grant.types = v_child_grant.types AND v_parent_grant.grantee = v_child_grant.grantee THEN
                  v_match := true;
              END IF;
          END LOOP;
  
          IF v_match = false THEN
              EXECUTE 'GRANT '||array_to_string(v_parent_grant.types, ',')||' ON '||v_child_table||' 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_child_table||' FROM '||v_parent_grant.grantee||' CASCADE';
              END IF;
          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_child_table
              EXCEPT
              SELECT unnest(v_grantees)) x;
          IF v_revoke IS NOT NULL THEN
              EXECUTE 'REVOKE ALL ON '||v_child_table||' FROM '||array_to_string(v_revoke, ',');
          END IF;
      END IF;
  
      SELECT tableowner INTO v_child_owner FROM pg_tables WHERE schemaname ||'.'|| tablename = v_child_table;
      IF v_parent_owner <> v_child_owner THEN
          EXECUTE 'ALTER TABLE '||v_child_table||' OWNER TO '||v_parent_owner;
      END IF;
  
  END LOOP;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Done');
      PERFORM close_job(v_job_id);
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN RE-APPLYING PRIVILEGES TO ALL CHILD TABLES OF: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to undo id-based partitioning created by this extension
   */
  CREATE OR REPLACE FUNCTION undo_partition_id(p_parent_table text, p_batch_count int DEFAULT 1, p_batch_interval bigint DEFAULT NULL, p_keep_table boolean DEFAULT true) RETURNS bigint
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock              boolean;
  v_batch_loop_count      int := 0;
  v_child_loop_total      bigint := 0;
  v_child_min             bigint;
  v_child_table           text;
  v_control               text;
  v_function_name         text;
  v_inner_loop_count      int;
  v_job_id                bigint;
  v_jobmon                boolean;
  v_jobmon_schema         text;
  v_move_sql              text;
  v_old_search_path       text;
  v_parent_schema         text;
  v_parent_tablename      text;
  v_part_interval         bigint;
  v_row                   record;
  v_rowcount              bigint;
  v_step_id               bigint;
  v_trig_name             text;
  v_total                 bigint := 0;
  v_undo_count            int := 0;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman undo_id_partition'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'undo_id_partition already running.';
      RETURN 0;
  END IF;
  
  SELECT part_interval::bigint
      , control
      , jobmon
  INTO v_part_interval
      , v_control
      , v_jobmon
  FROM @extschema@.part_config 
  WHERE parent_table = p_parent_table 
  AND (type = 'id-static' OR type = 'id-dynamic');
  
  IF v_part_interval IS NULL THEN
      RAISE EXCEPTION 'Configuration for given parent table not found: %', p_parent_table;
  END IF;
  
  IF v_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN UNDO PARTITIONING: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Undoing partitioning for table '||p_parent_table);
  END IF;
  
  IF p_batch_interval IS NULL THEN
      p_batch_interval := v_part_interval;
  END IF;
  
  -- Stops new time partitons from being made as well as stopping child tables from being dropped if they were configured with a retention period.
  UPDATE @extschema@.part_config SET undo_in_progress = true WHERE parent_table = p_parent_table;
  -- Stop data going into child tables and stop new id partitions from being made.
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  v_trig_name := @extschema@.check_name_length(p_object_name := v_parent_tablename, p_suffix := '_part_trig'); 
  v_function_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, '_part_trig_func', FALSE);
  EXECUTE 'DROP TRIGGER IF EXISTS '||v_trig_name||' ON '||p_parent_table;
  EXECUTE 'DROP FUNCTION IF EXISTS '||v_function_name||'()';
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Stopped partition creation process. Removed trigger & trigger function');
  END IF;
  
  <<outer_child_loop>>
  WHILE v_batch_loop_count < p_batch_count LOOP 
      SELECT n.nspname||'.'||c.relname INTO v_child_table
      FROM pg_inherits i 
      JOIN pg_class c ON i.inhrelid = c.oid 
      JOIN pg_namespace n ON c.relnamespace = n.oid 
      WHERE i.inhparent::regclass = p_parent_table::regclass 
      ORDER BY i.inhrelid ASC;
  
      EXIT WHEN v_child_table IS NULL;
  
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing child partition: '||v_child_table);
      END IF;
  
      EXECUTE 'SELECT min('||v_control||') FROM '||v_child_table INTO v_child_min;
      IF v_child_min IS NULL THEN
          -- No rows left in this child table. Remove from partition set.
          EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
          IF p_keep_table = false THEN
              EXECUTE 'DROP TABLE '||v_child_table;
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table DROPPED. Moved '||v_child_loop_total||' rows to parent');
              END IF;
          ELSE
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table UNINHERITED, not DROPPED. Moved '||v_child_loop_total||' rows to parent');
              END IF;
          END IF;
          v_undo_count := v_undo_count + 1;
          CONTINUE outer_child_loop;
      END IF;
      v_inner_loop_count := 1;
      v_child_loop_total := 0;
      <<inner_child_loop>>
      LOOP
          -- Get everything from the current child minimum up to the multiples of the given interval
          v_move_sql := 'WITH move_data AS (DELETE FROM '||v_child_table||
                  ' WHERE '||v_control||' <= '||quote_literal(v_child_min + (p_batch_interval * v_inner_loop_count))||' RETURNING *)
              INSERT INTO '||p_parent_table||' SELECT * FROM move_data';
          EXECUTE v_move_sql;
          GET DIAGNOSTICS v_rowcount = ROW_COUNT;
          v_total := v_total + v_rowcount;
          v_child_loop_total := v_child_loop_total + v_rowcount;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Moved '||v_child_loop_total||' rows to parent.');
          END IF;
          EXIT inner_child_loop WHEN v_rowcount = 0; -- exit before loop incr if table is empty
          v_inner_loop_count := v_inner_loop_count + 1;
          v_batch_loop_count := v_batch_loop_count + 1;
          EXIT outer_child_loop WHEN v_batch_loop_count >= p_batch_count; -- Exit outer FOR loop if p_batch_count is reached
      END LOOP inner_child_loop;
  END LOOP outer_child_loop;
  
  IF v_batch_loop_count < p_batch_count THEN
      -- FOR loop never ran, so there's no child tables left.
      DELETE FROM @extschema@.part_config WHERE parent_table = p_parent_table;
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing config from pg_partman');
          PERFORM update_step(v_step_id, 'OK', 'Done');
      END IF;
  END IF;
  
  RAISE NOTICE 'Copied % row(s) to the parent. Removed % partitions.', v_total, v_undo_count;
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Final stats');
      PERFORM update_step(v_step_id, 'OK', 'Copied '||v_total||' row(s) to the parent. Removed '||v_undo_count||' partitions.');
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  RETURN v_total;
  
  EXCEPTION
      WHEN OTHERS THEN
          IF v_jobmon_schema IS NOT NULL THEN
              IF v_job_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_job(''PARTMAN UNDO PARTITIONING: '||p_parent_table||''')' INTO v_job_id;
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before job logging started'')' INTO v_step_id;
              ELSIF v_step_id IS NULL THEN
                  EXECUTE 'SELECT '||v_jobmon_schema||'.add_step('||v_job_id||', ''EXCEPTION before first step logged'')' INTO v_step_id;
              END IF;
              EXECUTE 'SELECT '||v_jobmon_schema||'.update_step('||v_step_id||', ''CRITICAL'', ''ERROR: '||coalesce(SQLERRM,'unknown')||''')';
              EXECUTE 'SELECT '||v_jobmon_schema||'.fail_job('||v_job_id||')';
          END IF;
          RAISE EXCEPTION '%', SQLERRM;
  END
  $$;
  
  
  /*
   * Function to undo partitioning. 
   * Will actually work on any parent/child table set, not just ones created by pg_partman.
   */
  CREATE FUNCTION undo_partition(p_parent_table text, p_batch_count int DEFAULT 1, p_keep_table boolean DEFAULT true, p_jobmon boolean DEFAULT true) RETURNS bigint
      LANGUAGE plpgsql SECURITY DEFINER
      AS $$
  DECLARE
  
  v_adv_lock              boolean;
  v_batch_loop_count      bigint := 0;
  v_child_count           bigint;
  v_child_table           text;
  v_copy_sql              text;
  v_function_name         text;
  v_job_id                bigint;
  v_jobmon_schema         text;
  v_old_search_path       text;
  v_parent_schema         text;
  v_parent_tablename      text;
  v_part_interval         interval;
  v_rowcount              bigint;
  v_step_id               bigint;
  v_total                 bigint := 0;
  v_trig_name             text;
  v_undo_count            int := 0;
  
  BEGIN
  
  v_adv_lock := pg_try_advisory_xact_lock(hashtext('pg_partman undo_partition'));
  IF v_adv_lock = 'false' THEN
      RAISE NOTICE 'undo_partition already running.';
      RETURN 0;
  END IF;
  
  IF p_jobmon THEN
      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;
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      v_job_id := add_job('PARTMAN UNDO PARTITIONING: '||p_parent_table);
      v_step_id := add_step(v_job_id, 'Undoing partitioning for table '||p_parent_table);
  END IF;
  
  -- Stops new time partitons from being made as well as stopping child tables from being dropped if they were configured with a retention period.
  UPDATE @extschema@.part_config SET undo_in_progress = true WHERE parent_table = p_parent_table;
  -- Stop data going into child tables and stop new id partitions from being made.
  SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_tables WHERE schemaname ||'.'|| tablename = p_parent_table;
  v_trig_name := @extschema@.check_name_length(p_object_name := v_parent_tablename, p_suffix := '_part_trig'); 
  v_function_name := @extschema@.check_name_length(v_parent_tablename, v_parent_schema, '_part_trig_func', FALSE);
  EXECUTE 'DROP TRIGGER IF EXISTS '||v_trig_name||' ON '||p_parent_table;
  EXECUTE 'DROP FUNCTION IF EXISTS '||v_function_name||'()';
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM update_step(v_step_id, 'OK', 'Stopped partition creation process. Removed trigger & trigger function');
  END IF;
  
  WHILE v_batch_loop_count < p_batch_count LOOP 
      SELECT n.nspname||'.'||c.relname INTO v_child_table
      FROM pg_inherits i 
      JOIN pg_class c ON i.inhrelid = c.oid 
      JOIN pg_namespace n ON c.relnamespace = n.oid 
      WHERE i.inhparent::regclass = p_parent_table::regclass 
      ORDER BY i.inhrelid ASC;
  
      EXIT WHEN v_child_table IS NULL;
  
      EXECUTE 'SELECT count(*) FROM '||v_child_table INTO v_child_count;
      IF v_child_count = 0 THEN
          -- No rows left in this child table. Remove from partition set.
          EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
          IF p_keep_table = false THEN
              EXECUTE 'DROP TABLE '||v_child_table;
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table DROPPED. Moved '||coalesce(v_rowcount, 0)||' rows to parent');
              END IF;
          ELSE
              IF v_jobmon_schema IS NOT NULL THEN
                  PERFORM update_step(v_step_id, 'OK', 'Child table UNINHERITED, not DROPPED. Copied '||coalesce(v_rowcount, 0)||' rows to parent');
              END IF;
          END IF;
          v_undo_count := v_undo_count + 1;
          CONTINUE;
      END IF;
  
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing child partition: '||v_child_table);
      END IF;
     
      v_copy_sql := 'INSERT INTO '||p_parent_table||' SELECT * FROM '||v_child_table;
      EXECUTE v_copy_sql;
      GET DIAGNOSTICS v_rowcount = ROW_COUNT;
      v_total := v_total + v_rowcount;
  
      EXECUTE 'ALTER TABLE '||v_child_table||' NO INHERIT ' || p_parent_table;
      IF p_keep_table = false THEN
          EXECUTE 'DROP TABLE '||v_child_table;
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Child table DROPPED. Moved '||v_rowcount||' rows to parent');
          END IF;
      ELSE
          IF v_jobmon_schema IS NOT NULL THEN
              PERFORM update_step(v_step_id, 'OK', 'Child table UNINHERITED, not DROPPED. Copied '||v_rowcount||' rows to parent');
          END IF;
      END IF;
      v_batch_loop_count := v_batch_loop_count + 1;
      v_undo_count := v_undo_count + 1;         
  END LOOP;
  
  IF v_undo_count = 0 THEN
      -- FOR loop never ran, so there's no child tables left.
      DELETE FROM @extschema@.part_config WHERE parent_table = p_parent_table;
      IF v_jobmon_schema IS NOT NULL THEN
          v_step_id := add_step(v_job_id, 'Removing config from pg_partman (if it existed)');
          PERFORM update_step(v_step_id, 'OK', 'Done');
      END IF;
  END IF;
  
  RAISE NOTICE 'Copied % row(s) from % child table(s) to the parent: %', v_total, v_undo_count, p_parent_table;
  IF v_jobmon_schema IS NOT NULL THEN
      v_step_id := add_step(v_job_id, 'Final stats');
      PERFORM update_step(v_step_id, 'OK', 'Copied '||v_total||' row(s) from '||v_undo_count||' child table(s) to the parent');
  END IF;
  
  IF v_jobmon_schema IS NOT NULL THEN
      PERFORM close_job(v_job_id);
      EXECUTE 'SELECT set_config(''search_path'','''||v_old_search_path||''',''false'')';
  END IF;
  
  RETURN v_total;
  
  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 UNDO PARTITIONING: '||p_parent_table);
                  v_step_id := add_step(v_job_id, 'Partition function 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, 'CRITICAL', '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
  $$;
  
  DO $$
  DECLARE
  v_row   record;
  BEGIN
      FOR v_row IN SELECT statement FROM partman_preserve_privs_temp LOOP
          IF v_row.statement IS NOT NULL THEN
              EXECUTE v_row.statement;
          END IF;
      END LOOP;
  END
  $$;
  
  DROP TABLE partman_preserve_privs_temp;
  