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>
Add rule chaining: later sequence rules can read earlier outputs
2026-04-17 07:48:23 -04:00 · 2026-04-16 23:04:14 -04:00
3 changed files with 234 additions and 126 deletions
--- a/database/functions.sql
+++ b/database/functions.sql
@ -187,17 +187,27 @@ CREATE OR REPLACE FUNCTION apply_transformations(
    p_record_ids  INTEGER[] DEFAULT NULL,  -- NULL = all eligible records
    p_overwrite   BOOLEAN   DEFAULT FALSE  -- FALSE = skip already-transformed, TRUE = overwrite all
 ) RETURNS JSON AS $$
-WITH
-- All records to process
-qualifying AS (
+DECLARE
+    v_seq        INT;
+    v_seq_count  INT;
+    v_count      INT := 0;
+BEGIN
+    -- 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))
-),
-- Mirror TPS rx: fan out one row per regex match, drive from rules → records
-rx AS (
+        ),
+        rx AS (
            SELECT
                q.id,
                r.name          AS rule_name,
@ -206,7 +216,6 @@ rx AS (
                r.retain,
                r.function_type,
                COALESCE(mt.rn, rp.rn, 1) AS result_number,
-        -- extract: build map_val and retain_val per match (mirrors TPS)
                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
@ -217,16 +226,11 @@ rx AS (
                WITH ORDINALITY AS rp(rp, rn) ON r.function_type = 'replace'
            WHERE r.source_name = p_source_name
              AND r.enabled = true
-),
-- Aggregate match rows back into one value per (record, rule) — mirrors TPS agg_to_target_items
-agg_matches AS (
+              AND (r.function_type != 'extract' OR mt.mt IS NOT NULL)
+        ),
+        agg_matches AS (
            SELECT
-        id,
-        rule_name,
-        sequence,
-        output_field,
-        retain,
-        function_type,
+                id, rule_name, sequence, output_field, retain, function_type,
                CASE function_type
                    WHEN 'replace' THEN jsonb_agg(replace_val) -> 0
                    ELSE
@ -237,52 +241,32 @@ agg_matches AS (
                END AS extracted
            FROM rx
            GROUP BY id, rule_name, sequence, output_field, retain, function_type
-),
-- Join with mappings to find mapped output — mirrors TPS link_map
-linked AS (
-    SELECT
-        a.id,
-        a.sequence,
-        a.output_field,
-        a.retain,
-        a.extracted,
-        m.output AS mapped
+        ),
+        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
-),
-- Build per-rule output JSONB:
--   mapped  → use mapping output; also write output_field if retain = true
--   no map  → write extracted value to output_field
-rule_output AS (
-    SELECT
-        id,
-        sequence,
+        ),
+        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
+                        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
-),
-- Merge all rule outputs per record in sequence order — mirrors TPS agg_to_id
-record_additions AS (
-    SELECT
-        id,
-        dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
+        ),
+        record_additions AS (
+            SELECT id, dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
            FROM rule_output
            GROUP BY id
-),
-- Update all qualifying records; records with no rule matches get transformed = data
-updated AS (
+        ),
+        updated AS (
            UPDATE dataflow.records rec
            SET transformed    = rec.data || COALESCE(ra.additions, '{}'::jsonb) || COALESCE(rec.overrides, '{}'::jsonb),
                transformed_at = CURRENT_TIMESTAMP
@ -290,12 +274,113 @@ updated AS (
            LEFT JOIN record_additions ra ON ra.id = q.id
            WHERE rec.id = q.id
            RETURNING rec.id
-)
-SELECT json_build_object('success', true, 'transformed', count(*))
-FROM updated
-$$ LANGUAGE sql;
+        )
+        SELECT count(*) INTO v_count FROM updated;

-COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records (set-based CTE)';
+        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
+    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));
+
+    FOR v_seq IN
+        SELECT DISTINCT sequence
+        FROM dataflow.rules
+        WHERE source_name = p_source_name AND enabled = true
+        ORDER BY sequence
+    LOOP
+        WITH
+        current AS (
+            SELECT id, data || additions AS current_data
+            FROM _xform_acc
+        ),
+        rx AS (
+            SELECT
+                c.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 current c ON (c.current_data ? r.field)
+            LEFT JOIN LATERAL regexp_matches(c.current_data ->> r.field, r.pattern, r.flags)
+                WITH ORDINALITY AS mt(mt, rn) ON r.function_type = 'extract'
+            LEFT JOIN LATERAL regexp_replace(c.current_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.sequence = v_seq
+              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
+        ),
+        seq_additions AS (
+            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 = acc.additions || COALESCE(sa.seq_adds, '{}'::jsonb)
+        FROM seq_additions sa
+        WHERE acc.id = sa.id;
+    END LOOP;
+
+    WITH updated AS (
+        UPDATE dataflow.records rec
+        SET transformed    = rec.data || acc.additions || COALESCE(rec.overrides, '{}'::jsonb),
+            transformed_at = CURRENT_TIMESTAMP
+        FROM _xform_acc acc
+        WHERE rec.id = acc.id
+        RETURNING rec.id
+    )
+    SELECT count(*) INTO v_count FROM updated;
+
+    RETURN json_build_object('success', true, 'transformed', v_count);
+END;
+$$ LANGUAGE plpgsql;
+
+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
--- a/database/queries/rules.sql
+++ b/database/queries/rules.sql
@ -86,21 +86,27 @@ CREATE OR REPLACE FUNCTION preview_rule(
    p_limit         INT  DEFAULT 20
 )
 RETURNS TABLE (id INT, raw_value TEXT, extracted_value JSONB) AS $$
+-- Field is resolved from data first, then transformed (supports chained rules whose
+-- input field was produced by an earlier-sequence rule rather than the raw import).
 BEGIN
    IF p_function_type = 'replace' THEN
        RETURN QUERY
        SELECT
            r.id,
-            r.data ->> p_field,
-            to_jsonb(regexp_replace(r.data ->> p_field, p_pattern, p_replace_value, p_flags))
+            COALESCE(r.data ->> p_field, r.transformed ->> p_field),
+            to_jsonb(regexp_replace(
+                COALESCE(r.data ->> p_field, r.transformed ->> p_field),
+                p_pattern, p_replace_value, p_flags
+            ))
        FROM dataflow.records r
-        WHERE source_name = p_source AND data ? p_field
+        WHERE source_name = p_source
+          AND (data ? p_field OR transformed ? p_field)
        ORDER BY r.id DESC LIMIT p_limit;
    ELSE
        RETURN QUERY
        SELECT
            r.id,
-            r.data ->> p_field,
+            COALESCE(r.data ->> p_field, r.transformed ->> p_field),
            CASE
                WHEN agg.match_count = 0 THEN NULL
                WHEN agg.match_count = 1 THEN agg.matches -> 0
@ -114,10 +120,14 @@ BEGIN
                    ORDER BY rn
                ) AS matches,
                count(*)::int AS match_count
-            FROM regexp_matches(r.data ->> p_field, p_pattern, p_flags)
+            FROM regexp_matches(
+                COALESCE(r.data ->> p_field, r.transformed ->> p_field),
+                p_pattern, p_flags
+            )
            WITH ORDINALITY AS m(mt, rn)
        ) agg
-        WHERE r.source_name = p_source AND r.data ? p_field
+        WHERE r.source_name = p_source
+          AND (r.data ? p_field OR r.transformed ? p_field)
        ORDER BY r.id DESC LIMIT p_limit;
    END IF;
 END;
--- a/ui/src/pages/Rules.jsx
+++ b/ui/src/pages/Rules.jsx
@ -42,7 +42,7 @@ function PreviewModal({ rows, onClose }) {
  )
 }

-function FormPanel({ form, setForm, editing, error, loading, fields, source, onSubmit, onCancel }) {
+function FormPanel({ form, setForm, editing, error, loading, fields, rules, source, onSubmit, onCancel }) {
  const [preview, setPreview] = useState([])
  const [previewing, setPreviewing] = useState(false)
  const [modalOpen, setModalOpen] = useState(false)
@ -91,15 +91,28 @@ function FormPanel({ form, setForm, editing, error, loading, fields, source, onS
        <div className="grid grid-cols-2 gap-3">
          <div>
            <label className="text-xs text-gray-500 block mb-1">Input field</label>
-            {fields.length > 0 ? (
+            {fields.length > 0 ? (() => {
+              // Output fields from rules at a lower sequence — available as chained inputs
+              const chainedFields = [...new Set(
+                (rules || [])
+                  .filter(r => r.sequence < form.sequence && r.output_field && (!editing || r.id !== editing))
+                  .map(r => r.output_field)
+              )].filter(f => !fields.includes(f))
+              return (
                <select
                  className="w-full border border-gray-200 rounded px-3 py-1.5 text-sm focus:outline-none focus:border-blue-400"
                  value={form.field} onChange={e => setForm(f => ({ ...f, field: e.target.value }))}
                >
                  <option value="">— select field —</option>
                  {fields.map(f => <option key={f} value={f}>{f}</option>)}
+                  {chainedFields.length > 0 && (
+                    <optgroup label="from earlier rules">
+                      {chainedFields.map(f => <option key={f} value={f}>{f}</option>)}
+                    </optgroup>
+                  )}
                </select>
-            ) : (
+              )
+            })() : (
              <input
                className="w-full border border-gray-200 rounded px-3 py-1.5 text-sm focus:outline-none focus:border-blue-400"
                value={form.field} onChange={e => setForm(f => ({ ...f, field: e.target.value }))}
@ -322,7 +335,7 @@ export default function Rules({ source }) {
      {creating && (
        <FormPanel
          form={form} setForm={setForm} editing={false}
-          error={error} loading={loading} fields={fields} source={source}
+          error={error} loading={loading} fields={fields} rules={rules} source={source}
          onSubmit={handleSubmit}
          onCancel={() => { setCreating(false); setError('') }}
        />
@ -377,8 +390,8 @@ export default function Rules({ source }) {
                  </div>
                  <div className="px-4 pb-4">
                    <FormPanel
-                      form={form} setForm={setForm} editing={true}
-                      error={error} loading={loading} fields={fields} source={source}
+                      form={form} setForm={setForm} editing={rule.id}
+                      error={error} loading={loading} fields={fields} rules={rules} source={source}
                      onSubmit={e => handleSubmit(e, rule.id)}
                      onCancel={() => { setEditing(null); setExpanded(null) }}
                    />