Flowfile Codegen Parity Campaign

Flowfile codegen parity campaign

Read this first. Flowfile ships two executors of the same node settings: the live DAG engine (what the Designer runs) and the code generator (export_flow_to_polars / export_flow_to_flowframe), which emits standalone Python. When they disagree, the exported code runs clean but produces a different answer — silent, order-dependent, data-corrupting bugs. The sort bug in incident #544 (a "desc" vs "Descending" string mismatch) shipped exactly because generation was never tested against execution. This skill is the runbook to find and close those gaps measurably — never "by eye".

The maintainer named this the single hardest live problem in the repo (2026-07-03).

Status 2026-09-12 (branch fix/xfails): rows S1, X1, X2 and D1 are closed — the three edge-case markers are deleted, make_unique treats columns=[] as all-columns (keeping keep=strategy), and concat aggregations emit str.join(',') through the shared transform_schema.STRING_CONCAT_DELIMITER (the engine's string_concat uses the same constant). Scoreboard: edge-case 49 passed, 0 xfailed, 0 xpassed, corpus 692 passed, custom-nodes 20 passed. Phases 3.1, 3.2 and 3.4 below are historical; D2 and F1–F4 remain open.

What "parity" means (the only success criterion)

Parity holds for a flow when, for every output node:

collect( execute_flow(flow).node[N] )  ==  collect( exec(export_code(flow))["run_etl_pipeline"]() )

i.e. same input flow → exported code → executed result equals the in-engine result, compared with polars.testing.assert_frame_equal(..., check_column_order=False, check_row_order=False). The repo encodes this exact check in assert_flow_result_matches_generated(flow, output_node_id, code) (flowfile_core/tests/flowfile/test_code_generator_edge_cases.py:65). Every parity test builds a flow, runs both executors, and asserts frame-equality.

Campaign success is binary and machine-checked: zero xfailed and zero xpassed in the codegen edge-case suite, plus the full test_code_generator.py corpus (parameterized over both exporters) still green, plus make check_stubs clean. If you cannot state the pass/xfail numbers, you are not done.

When NOT to use this skill

• Adding a FlowFrame method / understanding how the emitter chooses native-node vs polars-code / Expr._repr_str mechanics → use flowfile-frame-and-codegen (that is the home for emitter internals; this skill only fixes divergences).
• Building a brand-new visual node → flowfile-node-development.
• The narrative history of the #544 sort bug and its sibling codegen bugs → flowfile-failure-archaeology (this skill uses #544 only as the cautionary rule).
• PR mechanics, drift gates, branch policy → flowfile-change-control (routed to, not duplicated).
• Isolated-DB / env-flag rationale → flowfile-config-and-flags and flowfile-testing-and-validation (this skill states the one command it needs and cross-refs there).
• make stubs internals / build orchestration → flowfile-build-and-env.

Phase 0 — Environment (do this once, every session)

The code generator lives entirely in core. Importing flowfile_core auto-runs Alembic against the catalog DB, so isolate the DB on every test invocation or you will corrupt the shared catalog.

# Pick a scratch DB path OUTSIDE the repo, one per parallel run. NEVER omit this.
export PARITY_DB=/tmp/ff_parity_$$.db

Hard rules (see flowfile-config-and-flags for why):

• Always prefix pytest with FLOWFILE_DB_PATH=$PARITY_DB.
• Never combine a fresh FLOWFILE_DB_PATH with FLOWFILE_SKIP_STARTUP_MIGRATION=1 — the DB will have no tables and you get no such table: users cascades.
• Run from the repo root with poetry run.

Phase 1 — BASELINE (reproduce, then freeze the numbers)

Run these two commands now, before changing anything, and confirm you see the embedded numbers. If your numbers differ, the inventory below is stale — STOP and re-verify against the source (Phase 2 line refs) before proceeding.

1a. The codegen edge-case suite (the campaign scoreboard)

FLOWFILE_DB_PATH=$PARITY_DB poetry run pytest \
  flowfile_core/tests/flowfile/test_code_generator_edge_cases.py \
  -q -p no:cacheprovider -rX

Expected baseline (2026-09-12): 49 passed, with no xfailed or xpassed lines.

The campaign rule: an XPASS in this repo means "delete the marker". Non-strict markers keep CI green, so nobody notices the rot.

Branches:

• no such table: users (or similar OperationalError cascade) → FLOWFILE_SKIP_STARTUP_MIGRATION=1 is set in your environment alongside a fresh DB. unset FLOWFILE_SKIP_STARTUP_MIGRATION, pick a new $PARITY_DB, re-run. Never combine that flag with a fresh test DB.
• Any xpassed → a marker went stale; each one is a delete-the-marker row. -rX names them all.
• Any failed or fewer than 49 passed → the tree drifted from this frozen baseline; re-verify the Phase 2 anchors with the Provenance greps before touching anything.

1b. The full parity corpus (regression floor — must stay green)

FLOWFILE_DB_PATH=${PARITY_DB}.main poetry run pytest \
  flowfile_core/tests/flowfile/test_code_generator.py \
  -q -p no:cacheprovider
# Expected baseline (2026-09-12): 692 passed (no skips locally)

The round-trip flow-builders are parameterized @pytest.mark.parametrize("export_func", [export_flow_to_polars, export_flow_to_flowframe], ids=["polars", "flowframe"]) — the parametrize sites produce the bulk of the corpus — so the corpus is the sample-flow set for round-trip equivalence across both export frameworks. (The un-parameterized remainder are exporter-specific tests such as test_flowframe_* at :1326+.) That count is your floor. Any fix that drops it has regressed parity.

Branches:

• Any failed → re-run with -rf to name them; a fix that breaks the corpus is a parity regression, full stop — revert or fix before proceeding.
• More than 692 passed, no failures → a merge added parity tests; re-freeze the number in your notes and use the observed count as the new floor.

There is no directory of .flowfile fixtures for parity — the corpus is these programmatic flow-builders. Add new sample flows there, next to create_sample_dataframe_node (test_code_generator.py:318), not as loose files.

A third file exists: test_code_generator_custom_nodes.py (baseline: 21 passed, <1s). It covers custom-node emission on the Polars exporter and the FlowFrame exporter; run it too whenever you touch code_generator.py's custom-node path.

Phase 2 — INVENTORY (the worklist)

Every known divergence, with a status column. Work top-down; each row links to a fix phase.

Two secondary emitter defects that are not yet guarded by a parity test (open, lower priority): param codegen accepts Python keywords as parameter names → invalid signature (param_codegen.py:38); typed flow-parameter empty default exports "" for int/float params (flowfile_core/flowfile_core/flowfile/param_types.py:96 — note: one level above code_generator/). Add a failing parity test before fixing either.

The FlowFrame framework-prefix trap underlies F1–F4: the same converter emits both pl.-prefixed Polars and ff.-prefixed FlowFrame code. export_flow_to_flowframe is newer and less complete than export_flow_to_polars. When a bug is FlowFrame-only, fix the FlowFrame converter path — do not "fix" it by making the Polars path emit FlowFrame syntax.

Phase 3 — Fix phases (ordered by leverage ÷ risk)

Do them in order. Each phase names the command, the exact expected observation, and a if you see X → do Y branch. Do not batch phases; land and verify one at a time.

Phase 3.1 — Delete the stale XPASS markers (highest leverage, lowest risk)

Removing a stale xfail is pure debt-payoff: the test already passes, so deleting the decorator makes it a normal green test that will now catch regressions.

1. Edit test_code_generator_edge_cases.py:201 — delete the @pytest.mark.xfail(...) decorator on test_in_operator_numeric (row S1). Leave the test body and docstring; optionally trim the "will fail until fixed" line in the docstring.
2. (If working near node_designer) delete the decorator at node_designer/test_node_designer.py:516 (row S2).

Verify:

FLOWFILE_DB_PATH=$PARITY_DB poetry run pytest \
  flowfile_core/tests/flowfile/test_code_generator_edge_cases.py \
  -q -p no:cacheprovider -rX

• Expect: 47 passed, 2 xfailed and NO xpassed line.
• If you see 1 xpassed still → you deleted the wrong decorator; the XPASS is test_in_operator_numeric.
• If you see 1 failed → the underlying fix regressed between baseline and now; re-run Phase 1a, and if it reproduces, that IN-filter split (filter_expressions.py:155) has been reverted — treat as a new bug, not a stale marker.

Phase 3.2 — X2 + D1: group_by concat delimiter (emitter fix)

The emitter must emit the flow's actual concat delimiter. Row X2 is a correctness bug; D1 is the paired deprecation. Fix them together (the delimiter must be explicit either way, because str.concat defaults to - and str.join defaults to "", but the flow uses ,).

1. The delimiter is not a stored node setting. The engine hard-codes it in transform_schema.string_concat (transform_schema.py:117-119: .str.concat(delimiter=",")), selected by AggColl.agg_func when agg == "concat" (transform_schema.py:886). The emitter maps "concat" → "str.concat" in _get_agg_function (expression_helpers.py:152) and the call site emits the mapped name with no arguments (transform_handlers.py:28: f"{self.framework}.col({old}).{agg_func}().alias({new})") — so Polars' default - wins. Fix: make the concat aggregation emit str.join(",") (modern name, delimiter explicit to match the engine). The map returns bare method names that get () appended, so special-case concat at the emit site or let the map carry an argument string.
2. Update D1's guard test_groupby_concat_uses_deprecated_str_concat (:1004): flip its TODO at :1054 — assert uses_modern and not uses_deprecated.
3. Delete the @pytest.mark.xfail on test_groupby_with_concat_aggregation (:766).

Verify (both classes — the D1 guard lives in TestDeprecatedMethodUsage, not TestGroupByEdgeCases):

FLOWFILE_DB_PATH=$PARITY_DB poetry run pytest \
  "flowfile_core/tests/flowfile/test_code_generator_edge_cases.py::TestGroupByEdgeCases" \
  "flowfile_core/tests/flowfile/test_code_generator_edge_cases.py::TestDeprecatedMethodUsage" \
  -q -p no:cacheprovider -rX

• Expect: 3 passed, 0 xfailed, 0 xpassed. (Baseline for this pair is 2 passed, 1 xfailed; TestGroupByEdgeCases holds exactly one test today.)
• If test_groupby_with_concat_aggregation still xfailed → your emitted call still lacks the explicit delimiter. Print export_flow_to_polars(flow) inside the test to see the emitted str.join/str.concat call.
• If test_groupby_concat_uses_deprecated_str_concat fails → you changed the emission but did not flip the guard's assertion at :1054 (or vice versa) — the two must land together.
• If assert_frame_equal fails with values like 'y,x' vs 'x,y' → the element order inside the concatenated string differs (the helper already ignores frame row/column order); ensure your emitted expression preserves the flow's group-member order — do not add a sort the flow doesn't have.

Phase 3.3 — D2: with_row_count → with_row_index (emitter fix, semantics-identical)

with_row_index(name=, offset=) is the drop-in modern replacement. Swap it at transform_handlers.py:370, then flip the D2 guard TODO at :1002.

Verify:

# D1 and D2 guards both live in TestDeprecatedMethodUsage (edge-case file:953).
# The round-trip guard for record-id semantics is TestRecordIdDeprecation (file:416).
FLOWFILE_DB_PATH=$PARITY_DB poetry run pytest \
  "flowfile_core/tests/flowfile/test_code_generator_edge_cases.py::TestDeprecatedMethodUsage" \
  "flowfile_core/tests/flowfile/test_code_generator_edge_cases.py::TestRecordIdDeprecation" \
  -q -p no:cacheprovider

• Expect: 3 passed, 0 failed (2 guards in TestDeprecatedMethodUsage + 1 round-trip in TestRecordIdDeprecation; the pair is also 3 passed at baseline — the swap must keep it so).
• If test_record_id_uses_deprecated_with_row_count fails → you swapped the emission but did not flip the guard TODO at :1002 (they must land together).
• If the round-trip in TestRecordIdDeprecation fails → with_row_index defaults differ in your Polars pin — check offset handling before shipping.

Phase 3.4 — X1: unique(columns=[]) (executor fix — the flow is wrong)

This is the only row where the emitter is correct and the flow engine is buggy. The unique node calls FlowDataEngine.make_unique (flow_graph.py:2415 → flow_data_engine.py:2686). Its guard only special-cases columns is None; an empty list columns=[] falls into the subset branch and emits .unique([], keep=...), which Polars rejects ("at least one key is required in a group_by operation"). Fix the executor: treat columns in (None, []) as all-columns → .unique(keep=unique_input.strategy) (mirror the emitted code; note the current None branch also drops keep= — preserve the strategy). Then delete the xfail at :459.

Because you are changing runtime engine semantics, you must keep the flow-vs-generated equivalence test as the regression guard (it already exists — that is the point). Do not fix the executor without it.

Verify:

FLOWFILE_DB_PATH=$PARITY_DB poetry run pytest \
  "flowfile_core/tests/flowfile/test_code_generator_edge_cases.py::TestUniqueOperationVariations" \
  -q -p no:cacheprovider -rX

• Expect: 2 passed, 0 xfailed (baseline for this class: 1 passed, 1 xfailed).
• If the other unique test in the class now fails → your executor change altered the non-empty-columns path. Scope the fix to the columns == [] branch only.
• Also re-run the whole engine suite before promoting — the exact command is Phase 5 step 3; an executor change touches every flow, not just codegen.