From d7f6e60040e68ead76adcf9a82a7f2373b3a72e9 Mon Sep 17 00:00:00 2001 From: Paul Trowbridge Date: Fri, 17 Apr 2026 07:48:23 -0400 Subject: [PATCH] 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 --- database/functions.sql | 113 +++++++++++++++++++++++++++++++++++------ 1 file changed, 98 insertions(+), 15 deletions(-) diff --git a/database/functions.sql b/database/functions.sql index 82e5671..6493a39 100644 --- a/database/functions.sql +++ b/database/functions.sql @@ -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