Fast-path single-pass CTE when no chaining is needed
If all rules share one sequence value, skip the loop and temp table and use the original single-pass CTE that the planner can fully optimize. The loop path only runs when multiple distinct sequence values exist (i.e. chaining is actually being used). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
fce427ba95
commit
d7f6e60040
@ -188,11 +188,100 @@ CREATE OR REPLACE FUNCTION apply_transformations(
|
||||
p_overwrite BOOLEAN DEFAULT FALSE -- FALSE = skip already-transformed, TRUE = overwrite all
|
||||
) RETURNS JSON AS $$
|
||||
DECLARE
|
||||
v_seq INT;
|
||||
v_count INT := 0;
|
||||
v_seq INT;
|
||||
v_seq_count INT;
|
||||
v_count INT := 0;
|
||||
BEGIN
|
||||
-- Accumulator: one row per qualifying record, additions built up across sequence steps.
|
||||
-- Each sequence step reads from data || additions so later rules can reference earlier outputs.
|
||||
-- Fast path: if all rules share one sequence value, no chaining is needed —
|
||||
-- use the original single-pass CTE which the planner can fully optimize.
|
||||
SELECT count(DISTINCT sequence) INTO v_seq_count
|
||||
FROM dataflow.rules
|
||||
WHERE source_name = p_source_name AND enabled = true;
|
||||
|
||||
IF v_seq_count <= 1 THEN
|
||||
WITH
|
||||
qualifying AS (
|
||||
SELECT id, data
|
||||
FROM dataflow.records
|
||||
WHERE source_name = p_source_name
|
||||
AND (p_overwrite OR transformed IS NULL)
|
||||
AND (p_record_ids IS NULL OR id = ANY(p_record_ids))
|
||||
),
|
||||
rx AS (
|
||||
SELECT
|
||||
q.id,
|
||||
r.name AS rule_name,
|
||||
r.sequence,
|
||||
r.output_field,
|
||||
r.retain,
|
||||
r.function_type,
|
||||
COALESCE(mt.rn, rp.rn, 1) AS result_number,
|
||||
CASE WHEN array_length(mt.mt, 1) = 1 THEN to_jsonb(mt.mt[1]) ELSE to_jsonb(mt.mt) END AS match_val,
|
||||
to_jsonb(rp.rp) AS replace_val
|
||||
FROM dataflow.rules r
|
||||
INNER JOIN qualifying q ON q.data ? r.field
|
||||
LEFT JOIN LATERAL regexp_matches(q.data ->> r.field, r.pattern, r.flags)
|
||||
WITH ORDINALITY AS mt(mt, rn) ON r.function_type = 'extract'
|
||||
LEFT JOIN LATERAL regexp_replace(q.data ->> r.field, r.pattern, r.replace_value, r.flags)
|
||||
WITH ORDINALITY AS rp(rp, rn) ON r.function_type = 'replace'
|
||||
WHERE r.source_name = p_source_name
|
||||
AND r.enabled = true
|
||||
AND (r.function_type != 'extract' OR mt.mt IS NOT NULL)
|
||||
),
|
||||
agg_matches AS (
|
||||
SELECT
|
||||
id, rule_name, sequence, output_field, retain, function_type,
|
||||
CASE function_type
|
||||
WHEN 'replace' THEN jsonb_agg(replace_val) -> 0
|
||||
ELSE
|
||||
CASE WHEN max(result_number) = 1
|
||||
THEN jsonb_agg(match_val ORDER BY result_number) -> 0
|
||||
ELSE jsonb_agg(match_val ORDER BY result_number)
|
||||
END
|
||||
END AS extracted
|
||||
FROM rx
|
||||
GROUP BY id, rule_name, sequence, output_field, retain, function_type
|
||||
),
|
||||
linked AS (
|
||||
SELECT a.id, a.sequence, a.output_field, a.retain, a.extracted, m.output AS mapped
|
||||
FROM agg_matches a
|
||||
LEFT JOIN dataflow.mappings m ON
|
||||
m.source_name = p_source_name
|
||||
AND m.rule_name = a.rule_name
|
||||
AND m.input_value = a.extracted
|
||||
WHERE a.extracted IS NOT NULL
|
||||
),
|
||||
rule_output AS (
|
||||
SELECT id, sequence,
|
||||
CASE
|
||||
WHEN mapped IS NOT NULL THEN
|
||||
mapped || CASE WHEN retain THEN jsonb_build_object(output_field, extracted) ELSE '{}'::jsonb END
|
||||
ELSE
|
||||
jsonb_build_object(output_field, extracted)
|
||||
END AS output
|
||||
FROM linked
|
||||
),
|
||||
record_additions AS (
|
||||
SELECT id, dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
|
||||
FROM rule_output
|
||||
GROUP BY id
|
||||
),
|
||||
updated AS (
|
||||
UPDATE dataflow.records rec
|
||||
SET transformed = rec.data || COALESCE(ra.additions, '{}'::jsonb) || COALESCE(rec.overrides, '{}'::jsonb),
|
||||
transformed_at = CURRENT_TIMESTAMP
|
||||
FROM qualifying q
|
||||
LEFT JOIN record_additions ra ON ra.id = q.id
|
||||
WHERE rec.id = q.id
|
||||
RETURNING rec.id
|
||||
)
|
||||
SELECT count(*) INTO v_count FROM updated;
|
||||
|
||||
RETURN json_build_object('success', true, 'transformed', v_count);
|
||||
END IF;
|
||||
|
||||
-- Chaining path: multiple sequence groups — process in order so each group
|
||||
-- can read fields written by earlier groups.
|
||||
CREATE TEMP TABLE _xform_acc ON COMMIT DROP AS
|
||||
SELECT id, data, '{}'::jsonb AS additions
|
||||
FROM dataflow.records
|
||||
@ -200,8 +289,6 @@ BEGIN
|
||||
AND (p_overwrite OR transformed IS NULL)
|
||||
AND (p_record_ids IS NULL OR id = ANY(p_record_ids));
|
||||
|
||||
-- Process one sequence group at a time, in order.
|
||||
-- Rules at sequence N can read fields written by rules at sequence < N.
|
||||
FOR v_seq IN
|
||||
SELECT DISTINCT sequence
|
||||
FROM dataflow.rules
|
||||
@ -209,12 +296,10 @@ BEGIN
|
||||
ORDER BY sequence
|
||||
LOOP
|
||||
WITH
|
||||
-- Current view of each record: original data merged with accumulated outputs so far
|
||||
current AS (
|
||||
SELECT id, data || additions AS current_data
|
||||
FROM _xform_acc
|
||||
),
|
||||
-- Fan out one row per regex match for rules at this sequence level
|
||||
rx AS (
|
||||
SELECT
|
||||
c.id,
|
||||
@ -235,6 +320,7 @@ BEGIN
|
||||
WHERE r.source_name = p_source_name
|
||||
AND r.sequence = v_seq
|
||||
AND r.enabled = true
|
||||
AND (r.function_type != 'extract' OR mt.mt IS NOT NULL)
|
||||
),
|
||||
agg_matches AS (
|
||||
SELECT
|
||||
@ -251,9 +337,7 @@ BEGIN
|
||||
GROUP BY id, rule_name, sequence, output_field, retain, function_type
|
||||
),
|
||||
linked AS (
|
||||
SELECT
|
||||
a.id, a.sequence, a.output_field, a.retain, a.extracted,
|
||||
m.output AS mapped
|
||||
SELECT a.id, a.sequence, a.output_field, a.retain, a.extracted, m.output AS mapped
|
||||
FROM agg_matches a
|
||||
LEFT JOIN dataflow.mappings m ON
|
||||
m.source_name = p_source_name
|
||||
@ -272,17 +356,16 @@ BEGIN
|
||||
FROM linked
|
||||
),
|
||||
seq_additions AS (
|
||||
SELECT id, dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
|
||||
SELECT id, dataflow.jsonb_concat_obj(output ORDER BY sequence) AS seq_adds
|
||||
FROM rule_output
|
||||
GROUP BY id
|
||||
)
|
||||
UPDATE _xform_acc acc
|
||||
SET additions = additions || COALESCE(sa.additions, '{}'::jsonb)
|
||||
SET additions = acc.additions || COALESCE(sa.seq_adds, '{}'::jsonb)
|
||||
FROM seq_additions sa
|
||||
WHERE acc.id = sa.id;
|
||||
END LOOP;
|
||||
|
||||
-- Write final result: original data + all accumulated rule outputs + any manual overrides
|
||||
WITH updated AS (
|
||||
UPDATE dataflow.records rec
|
||||
SET transformed = rec.data || acc.additions || COALESCE(rec.overrides, '{}'::jsonb),
|
||||
@ -297,7 +380,7 @@ BEGIN
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records. Rules are processed in sequence order — a rule at sequence N can read fields written by rules at sequence < N (chaining).';
|
||||
COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records. Single-sequence sources use a fast single-pass CTE; multi-sequence sources use a loop so rules at sequence N can read outputs from sequence < N (chaining).';
|
||||
|
||||
------------------------------------------------------
|
||||
-- Function: get_all_values
|
||||
|
||||
Loading…
Reference in New Issue
Block a user