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>

database/functions.sql

@@ -184,118 +184,203 @@ CREATE AGGREGATE dataflow.jsonb_concat_obj(JSONB) (
 DROP FUNCTION IF EXISTS apply_transformations(TEXT, INTEGER[]);
 CREATE OR REPLACE FUNCTION apply_transformations(
     p_source_name TEXT,
-    p_record_ids INTEGER[] DEFAULT NULL,  -- NULL = all eligible records
+    p_record_ids  INTEGER[] DEFAULT NULL,  -- NULL = all eligible records
-    p_overwrite   BOOLEAN  DEFAULT FALSE  -- FALSE = skip already-transformed, TRUE = overwrite all
+    p_overwrite   BOOLEAN   DEFAULT FALSE  -- FALSE = skip already-transformed, TRUE = overwrite all
 ) RETURNS JSON AS $$
-WITH
+DECLARE
--- All records to process
+    v_seq        INT;
-qualifying AS (
+    v_seq_count  INT;
-    SELECT id, data
+    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))
+        ),
+        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
     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))
+      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 (
-    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,
-        -- 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
-    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
-),
--- Aggregate match rows back into one value per (record, rule) — mirrors TPS agg_to_target_items
-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
-),
--- 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
-    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,
-        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
-),
--- 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
-    FROM rule_output
-    GROUP BY id
-),
--- Update all qualifying records; records with no rule matches get transformed = data
-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 json_build_object('success', true, 'transformed', count(*))
-FROM updated
-$$ LANGUAGE sql;
 
-COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records (set-based CTE)';
+    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

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,
+            COALESCE(r.data ->> p_field, r.transformed ->> p_field),
-            to_jsonb(regexp_replace(r.data ->> p_field, p_pattern, p_replace_value, p_flags))
+            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;

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 ? (() => {
-              <select
+              // Output fields from rules at a lower sequence — available as chained inputs
-                className="w-full border border-gray-200 rounded px-3 py-1.5 text-sm focus:outline-none focus:border-blue-400"
+              const chainedFields = [...new Set(
-                value={form.field} onChange={e => setForm(f => ({ ...f, field: e.target.value }))}
+                (rules || [])
-              >
+                  .filter(r => r.sequence < form.sequence && r.output_field && (!editing || r.id !== editing))
-                <option value="">— select field —</option>
+                  .map(r => r.output_field)
-                {fields.map(f => <option key={f} value={f}>{f}</option>)}
+              )].filter(f => !fields.includes(f))
-              </select>
+              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}
+                      form={form} setForm={setForm} editing={rule.id}
-                      error={error} loading={loading} fields={fields} source={source}
+                      error={error} loading={loading} fields={fields} rules={rules} source={source}
                       onSubmit={e => handleSubmit(e, rule.id)}
                       onCancel={() => { setEditing(null); setExpanded(null) }}
                     />