CREATE 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 , p_order text DEFAULT 'ASC' , p_analyze boolean DEFAULT true) RETURNS bigint LANGUAGE plpgsql SECURITY DEFINER AS $$ DECLARE v_control text; v_control_type text; v_datetime_string text; v_current_partition_name text; v_epoch text; v_last_partition text; v_lock_iter int := 1; v_lock_obtained boolean := FALSE; v_max_partition_timestamp timestamptz; v_min_partition_timestamp timestamptz; v_new_search_path text := '@extschema@,pg_temp'; v_old_search_path text; v_parent_schema text; v_parent_tablename text; v_partition_expression text; v_partition_interval interval; v_partition_suffix text; v_partition_timestamp timestamptz[]; v_rowcount bigint; v_start_control timestamptz; v_total_rows bigint := 0; v_type text; BEGIN /* * Populate the child table(s) of a time-based partition set with old data from the original parent */ SELECT partition_type , partition_interval::interval , control , datetime_string , epoch INTO v_type , v_partition_interval , v_control , v_datetime_string , v_epoch FROM @extschema@.part_config WHERE parent_table = p_parent_table AND partition_type IN ('partman', 'time-custom'); IF NOT FOUND THEN RAISE EXCEPTION 'ERROR: No entry in part_config found for non-native partitioning for given table: %', p_parent_table; END IF; SELECT schemaname, tablename INTO v_parent_schema, v_parent_tablename FROM pg_catalog.pg_tables WHERE schemaname = split_part(p_parent_table, '.', 1)::name AND tablename = split_part(p_parent_table, '.', 2)::name; SELECT general_type INTO v_control_type FROM @extschema@.check_control_type(v_parent_schema, v_parent_tablename, v_control); IF v_control_type <> 'time' THEN IF (v_control_type = 'id' AND v_epoch = 'none') OR v_control_type <> 'id' THEN RAISE EXCEPTION 'Cannot run on partition set without time based control column or epoch flag set with an id column. Found control: %, epoch: %', v_control_type, v_epoch; END IF; END IF; SELECT current_setting('search_path') INTO v_old_search_path; EXECUTE format('SELECT set_config(%L, %L, %L)', 'search_path', v_new_search_path, 'false'); IF p_batch_interval IS NULL OR p_batch_interval > v_partition_interval THEN p_batch_interval := v_partition_interval; END IF; SELECT partition_tablename INTO v_last_partition FROM @extschema@.show_partitions(p_parent_table, 'DESC') LIMIT 1; v_partition_expression := CASE WHEN v_epoch = 'seconds' THEN format('to_timestamp(%I)', v_control) WHEN v_epoch = 'milliseconds' THEN format('to_timestamp((%I/1000)::float)', v_control) ELSE format('%I', v_control) END; FOR i IN 1..p_batch_count LOOP IF p_order = 'ASC' THEN EXECUTE format('SELECT min(%s) FROM ONLY %I.%I', v_partition_expression, v_parent_schema, v_parent_tablename) INTO v_start_control; ELSIF p_order = 'DESC' THEN EXECUTE format('SELECT max(%s) FROM ONLY %I.%I', v_partition_expression, v_parent_schema, v_parent_tablename) INTO v_start_control; ELSE RAISE EXCEPTION 'Invalid value for p_order. Must be ASC or DESC'; END IF; IF v_start_control IS NULL THEN EXIT; END IF; IF v_type = 'partman' THEN CASE WHEN v_partition_interval = '15 mins' THEN v_min_partition_timestamp := date_trunc('hour', v_start_control) + '15min'::interval * floor(date_part('minute', v_start_control) / 15.0); WHEN v_partition_interval = '30 mins' THEN v_min_partition_timestamp := date_trunc('hour', v_start_control) + '30min'::interval * floor(date_part('minute', v_start_control) / 30.0); WHEN v_partition_interval = '1 hour' THEN v_min_partition_timestamp := date_trunc('hour', v_start_control); WHEN v_partition_interval = '1 day' THEN v_min_partition_timestamp := date_trunc('day', v_start_control); WHEN v_partition_interval = '1 week' THEN v_min_partition_timestamp := date_trunc('week', v_start_control); WHEN v_partition_interval = '1 month' THEN v_min_partition_timestamp := date_trunc('month', v_start_control); WHEN v_partition_interval = '3 months' THEN v_min_partition_timestamp := date_trunc('quarter', v_start_control); WHEN v_partition_interval = '1 year' THEN v_min_partition_timestamp := date_trunc('year', v_start_control); END CASE; ELSIF v_type = 'time-custom' THEN SELECT child_start_time INTO v_min_partition_timestamp FROM @extschema@.show_partition_info(v_parent_schema||'.'||v_last_partition , v_partition_interval::text , p_parent_table); v_max_partition_timestamp := v_min_partition_timestamp + v_partition_interval; LOOP IF v_start_control >= v_min_partition_timestamp AND v_start_control < v_max_partition_timestamp THEN EXIT; ELSE BEGIN IF v_start_control > v_max_partition_timestamp THEN -- Keep going forward in time, checking if child partition time interval encompasses the current v_start_control value v_min_partition_timestamp := v_max_partition_timestamp; v_max_partition_timestamp := v_max_partition_timestamp + v_partition_interval; ELSE -- Keep going backwards in time, checking if child partition time interval encompasses the current v_start_control value v_max_partition_timestamp := v_min_partition_timestamp; v_min_partition_timestamp := v_min_partition_timestamp - v_partition_interval; END IF; 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_timestamp; END; END IF; END LOOP; END IF; v_partition_timestamp := ARRAY[v_min_partition_timestamp]; IF p_order = 'ASC' THEN -- Ensure batch interval given as parameter doesn't cause maximum to overflow the current partition maximum IF (v_start_control + p_batch_interval) >= (v_min_partition_timestamp + v_partition_interval) THEN v_max_partition_timestamp := v_min_partition_timestamp + v_partition_interval; ELSE v_max_partition_timestamp := v_start_control + p_batch_interval; END IF; ELSIF p_order = 'DESC' THEN -- Must be greater than max value still in parent table since query below grabs < max v_max_partition_timestamp := v_min_partition_timestamp + v_partition_interval; -- Ensure batch interval given as parameter doesn't cause minimum to underflow current partition minimum IF (v_start_control - p_batch_interval) >= v_min_partition_timestamp THEN v_min_partition_timestamp = v_start_control - p_batch_interval; END IF; ELSE RAISE EXCEPTION 'Invalid value for p_order. Must be ASC or DESC'; 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 EXECUTE format('SELECT * FROM ONLY %I.%I WHERE %s >= %L AND %3$s < %5$L FOR UPDATE NOWAIT' , v_parent_schema , v_parent_tablename , v_partition_expression , v_min_partition_timestamp , v_max_partition_timestamp); 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; PERFORM @extschema@.create_partition_time(p_parent_table, v_partition_timestamp, p_analyze); -- This suffix generation code is in create_partition_time() as well v_partition_suffix := to_char(v_min_partition_timestamp, v_datetime_string); v_current_partition_name := @extschema@.check_name_length(v_parent_tablename, v_partition_suffix, TRUE); EXECUTE format('WITH partition_data AS ( DELETE FROM ONLY %I.%I WHERE %s >= %L AND %3$s < %5$L RETURNING *) INSERT INTO %I.%I SELECT * FROM partition_data' , v_parent_schema , v_parent_tablename , v_partition_expression , v_min_partition_timestamp , v_max_partition_timestamp , v_parent_schema , v_current_partition_name); GET DIAGNOSTICS v_rowcount = ROW_COUNT; v_total_rows := v_total_rows + v_rowcount; IF v_rowcount = 0 THEN EXIT; END IF; END LOOP; PERFORM @extschema@.create_function_time(p_parent_table); EXECUTE format('SELECT set_config(%L, %L, %L)', 'search_path', v_old_search_path, 'false'); RETURN v_total_rows; END $$;