Flowfile Frame And Codegen

flowfile_frame and codegen

When NOT to use this skill

• Adding a brand-new core node type (settings class, NODE_TYPE_TO_SETTINGS_CLASS, template, frontend component) — that four-layer registration process, including the flowfile_frame decision-to-add-a-method checklist at a shallower level, is flowfile-node-development. Come here only for the internals of how flowfile_frame itself is built (Expr repr machinery, injection decorators, lazy semantics, connectors).
• Systematically closing native-vs-polars-code parity gaps as a campaign/roadmap (which ops to convert next, tracking issues) — flowfile-codegen-parity-campaign. This skill documents today's parity table and the mechanics of extending it; it is not the battle plan.
• General core/worker/kernel system map, the worker-offload wire protocol, $ffsec$ secrets format — flowfile-architecture-contract.
• Version bump rules, the stub gate as a release mechanic, Alembic migration discipline — flowfile-change-control (this skill covers the stub generators' mechanics; change-control covers when/why the gate blocks a release).
• Running the test suite in general, pytest markers, Docker fixture matrix — flowfile-testing-and-validation.
• CLI flow execution (flowfile run), on-disk storage layout, scheduler — flowfile-run-and-operate.
• Debugging a specific failure symptom you don't yet understand — flowfile-debugging-playbook.

All facts below verified in-repo as of 2026-07-03 (v0.12.7), branch feature/claude-skills. Re-verification commands are in the final section.

1. Mental model

flowfile_frame (import flowfile_frame as ff) gives a Polars-LazyFrame-shaped surface (FlowFrame, Expr/Column) where every method call appends a node to an in-process flowfile_core FlowGraph as a side effect, in addition to (usually) computing data.

It is not a client of any backend HTTP API. It imports flowfile_core directly — flowfile_frame/flowfile_frame/flow_frame.py imports FlowDataEngine, FlowGraph, add_connection, combine_flow_graphs_with_mapping, FlowNode, input_schema, transform_schema straight from flowfile_core. There is no port, no serialization boundary, no auth — it mutates Python objects in the calling process. The graph it builds is the same kind of DAG the Designer UI edits and core executes: it can be saved to a .flowfile, opened in the Designer (open_graph_in_editor, §7), or materialized with .collect().

utils.create_flow_graph() seeds every new graph with track_history=False and, critically, forces flow_graph.flow_settings.execution_location = "local" — the comment in source is explicit: "so that the run time does not attempt to use the flowfile_worker process." This one setting is why several connector reads (§8) don't need a running worker, and why write_*/pivot behave eagerly (§6).

The FlowFrame object

A FlowFrame is a thin handle around four fields, all set in __new__ (flow_frame.py:274):

All initialization happens in __new__; __init__ is a deliberate no-op (flow_frame.py:370, docstring: "Python automatically calls __init__ after __new__... this empty method catches that call and safely does nothing"). __new__ is a factory with three branches: flow_graph+node_id given → wrap an existing node; flow_graph given + non-LazyFrame data → create_from_any_type (new source node); otherwise → build a fresh graph, register a NodeManualInput (Python object) or a runtime dependency (add_dependency_on_polars_lazy_frame, for an existing pl.LazyFrame), then recurse into itself with the new graph/node.

2. The three node-emission paths

Every graph-mutating method picks one of three ways to represent the operation. Read this before writing or reviewing any flowfile_frame code.

1. Native node. When the arguments fit a dedicated core node schema, the method builds a Pydantic settings object (flowfile_core.schemas.input_schema / transform_schema) and calls the matching flow_graph.add_*() (add_select, add_sort, add_join, add_group_by, add_formula, add_output, …). Renders as a first-class editable node in the Designer.
2. Polars-code node. Complex cases call FlowFrame._add_polars_code (flow_frame.py:588-684), which creates a NodePolarsCode whose payload is a generated Polars source string (transform_schema.PolarsCodeInput(polars_code=...)). Core re-executes that string in a sandbox at run time (§6) — the string is load-bearing; a wrong one breaks graph execution silently, only surfacing when the graph actually runs.
3. Serialization fallback (inside _add_polars_code, edge case within path 2). If an expression can't be turned into code (convertable_to_code=False — e.g. a lambda whose source can't be recovered) or the generated code still contains a raw <lambda> at 0x...> repr, the operation runs directly on self.data (getattr(self.data, method_name)(...)) and the resulting LazyFrame is base64-serialized into the node. Source carries an explicit warning: "This will result in a breaking graph when using the the ui." If the result isn't a LazyFrame at all, the node degrades further to output_df = input_df and the real result is injected as a precomputed_result on the node's results.resulting_data (_create_child_frame, flow_frame.py:399-418) — the visual graph now lies about what the node computes, and only a logger.error hints at it.

The general per-method skeleton

new_node_id = generate_node_id()                        # utils.py — process-global counter
# decide native vs polars-code (mirror an existing use_polars_code_path / can_use_native flag)
settings = input_schema.NodeXxx(
    flow_id=self.flow_graph.flow_id, node_id=new_node_id,
    depending_on_id=self.node_id, description=description, is_setup=True,
)
self.flow_graph.add_xxx(settings)                        # or self._add_polars_code(...)
return self._create_child_frame(new_node_id)              # wires the connection, re-reads data

_create_child_frame (flow_frame.py:399) calls _add_connection — an input_schema.NodeConnection.create_from_simple_input(...) + add_connection(self.flow_graph, ...), honoring self.output_handle — then immediately calls get_resulting_data() on the new node to obtain the next LazyFrame. Multi-input methods (join, concat, fuzzy_join) skip _create_child_frame and wire "main"/"right" connections manually before constructing the FlowFrame(...) directly.

Node ids are process-global, not per-graph. generate_node_id() (utils.py) increments a module-level dict data = {"c": 0}; set_node_id(n) resets it. Combining two graphs (e.g. via join across graphs, §3) bumps the counter by the combined node count. Consequence, verified live below: node ids inside a single graph can skip values — this is expected, not a bug.

Verified (isolated DB, FLOWFILE_DB_PATH set):

import flowfile_frame as ff
df = ff.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
out = (df.filter(ff.col("a") > 1)
         .with_columns((ff.col("a") * 2).alias("dbl"))
         .select("a", "dbl")
         .sort("a", descending=True))
for n in out.flow_graph.nodes:
    print(n.node_id, n.node_type, type(n.setting_input).__name__)

1 manual_input NodeManualInput
2 filter       NodeFilter          # filter — the predicate's formula form lowers onto the Filter node (see 3d)
4 formula      NodeFormula         # with_columns via _ff_repr, one native node per expr
5 select       NodeSelect
6 sort         NodeSort

Node id 3 never appears — with_columns burned an id internally before delegating to the formula path. out.collect() returns the correct 2-row pl.DataFrame.

3. Worked examples — read these to internalize the pattern

3a. sort — dual-path with an explicit trigger list (flow_frame.py:448)

• Goes native (NodeSort / transform_schema.SortByInput(column=…, how="asc"|"desc")) only when every sort key is a plain string or an unaltered Column, and nulls_last is falsy, and multithreaded=True, and maintain_order=False.
• Any Expr, an altered Column (aliased/cast), nulls_last, maintain_order, or multithreaded=False flips to polars-code: input_df.sort(<pure expr strs>, descending=…, …).
• Any collected function sources (from lambdas inside sort keys) get prepended as a definitions block, joined to the operation by the literal marker \n#─────SPLIT─────\n\n.

3b. select — native vs polars-code (flow_frame.py:1022)

• Strings and Column objects (including alias/cast via Column.to_select_input()) go native: builds a transform_schema.SelectInput list, adds unselected existing columns with a keep flag, emits input_schema.NodeSelect.
• Any general Expr, Selector, or expression with collected function sources → input_df.select([<expr strs>]) polars-code.
• Special case: select(len().alias("number_of_records")) is pattern-matched by string comparison of the repr (str(expr) == "pl.Expr(len()).alias('number_of_records')") and becomes a native NodeRecordCount.
• rename(mapping) is implemented as select([col(old).alias(new), …], _keep_missing=True) — its strict kwarg is accepted but silently ignored.

3c. join — native vs code, and cross-graph combining (flow_frame.py:686)

• _should_use_polars_code_for_join (flow_frame.py:781): any of maintain_order, coalesce, nulls_equal=True, validate, or a non-default suffix forces polars-code: input_df_1.join(other=input_df_2, how=…, …) with depending_on_ids=[self.node_id, other.node_id] and two "main" connections.
• Otherwise the native path builds JoinMaps and emits NodeJoin (or NodeCrossJoin for how="cross"), with auto_generate_selection=True, verify_integrity=True; right-side join keys get keep=False. Connections: left → "main", right → "right".
• _ensure_same_graph (flow_frame.py:791): if self and other live in different FlowGraphs, combine_flow_graphs_with_mapping merges them and mutates both frames' node_id/flow_graph in place. Any third FlowFrame still holding a reference into one of the old graphs is left stale — no re-mapping happens for frames the join doesn't touch directly. Keep frame references you plan to reuse after a cross-graph join to a minimum, or re-derive them from the merged graph.
• fuzzy_join(other, fuzzy_mappings: list[FuzzyMapping]) (flow_frame.py:3292) is a Flowfile-only extension with no Polars equivalent → NodeFuzzyMatch.

3d. group_by() → GroupByFrame (flow_frame.py:2361, flowfile_frame/flowfile_frame/group_frame.py)

• group_by() itself creates no node — it returns a GroupByFrame carrying the future node id.
• .agg(...) tries to convert every aggregation to a native transform_schema.AggColl row (agg="groupby" for the grouping keys). Conversion requires maintain_order=False, no complex expressions, no Selectors, and every agg_func in the fixed set _NATIVE_AGG_FUNCS = {sum, max, mean, median, min, count, n_unique, first, last, std, var, concat} (group_frame.py:20). Named aggs also accept name=("col", "aggstr") tuples. Success → NodeGroupBy; anything outside the set → polars-code input_df.group_by([...], maintain_order=…).agg(...).
• Direct .sum()/.mean()/.median() generate …agg(cs.numeric().sum()) and aggregate numeric columns only — a deliberate deviation from Polars' GroupBy.sum(), which aggregates everything it can (_NUMERIC_ONLY_METHODS, group_frame.py:235). .len()/.count()/.head()/.tail()/.first()/.last()/.min()/.max() map straight to .group_by(...).method(...).

Also note filter: when every predicate (and **constraints equality) has an _ff_repr — comparisons, &/|/~, is_in, is_null, the mapped .str/.dt helpers — the predicates are joined with and, checked with _formula_parses, and emitted as a native NodeFilter(advanced) whose expression is that formula (the flowfile_formula: str kwarg is always native). Lambdas, unmapped methods, or a formula core cannot parse fall back to a polars-code input_df.filter(...) node. filter_split builds its predicate the same way but raises ValueError when no formula form exists — the engine turns an unparseable advanced_filter into "keep nothing" with only a warning, and a two-output node has no Polars-code fallback. drop is hand-written too: plain names lower onto NodeSelect with keep=False (see §9).

4. The expression system (flowfile_frame/flowfile_frame/expr.py, ~1.7k lines)

Core state on every Expr

Every Expr carries two parallel representations plus metadata (expr.py:423-493):

• expr: pl.Expr | None — the live Polars expression, used for schema propagation and the serialization fallback; may be None for selector-aggregations.
• _repr_str: str — the load-bearing generated Polars source, e.g. "(pl.col('a') * 2).alias('dbl')". Every method returns a new Expr with the repr extended (_create_next_expr appends .method(args_repr); binary ops produce (left op right) and clear aggregation state).
• _ff_repr: str | None — an optional flowfile-formula representation ([colname], "str", (a + b), to_integer(...), uppercase(...), …) maintained through operators, casts, and a subset of .str/.dt methods. When every expression in a with_columns call still has _ff_repr + a resolvable column_name, the whole call becomes a chain of native NodeFormula nodes instead of one polars-code node — this is the mechanism behind the formula node in the worked example above.
• column_name / _initial_column_name (rename tracking), agg_func (drives GroupBy AggColl conversion), is_complex (blocks native group-by conversion), convertable_to_code (False → serialization fallback, §2), _function_sources: list[str] (extracted function/lambda definitions to prepend to generated code).

Subclasses & namespaces

• Column(Expr) — pl.col(name) plus a transform_schema.SelectInput; .alias()/.cast() return new Columns with is_altered/data_type_change flags so select/sort/unique can stay native when nothing structural changed.
• When(Expr) — when().then() mutate in place; .otherwise() returns a plain Expr carrying the full pl.when(...).then(...).otherwise(...) repr and an if c1 then v1 elseif c2 then v2 else v endif _ff_repr when every branch has one (a chain without otherwise renders else null), which is what lets with_columns lower a fluent when onto a Formula node.
• .str → StringMethods, .dt → DateTimeMethods (hand-written), .list → ExprListNameSpace, .name → ExprNameNameSpace.
• lit(value) uses pl.lit(value, allow_object=True) and repr(value); module-level @agg_function-decorated functions (max/min/first/last/mean/count/implode/explode/sum/corr/cov) render as pl.sum('a')-style calls and set agg_func.

Method injection (two decorators — most of the surface is not hand-written)

• add_expr_methods(Expr), invoked at expr.py:1403 (module import time). For every callable on pl.Expr not already defined on Expr and not a dunder/property, it installs a wrapper (adding_expr.py) that calls the real Polars method for schema/error-checking (failures set result_expr=None and log at debug level, not raise), classifies the call via hard-coded agg_methods/complex_methods sets, and appends it to _repr_str. PASSTHROUGH_METHODS = {"map_elements", "map_batches"} (adding_expr.py:16) get callable-source extraction; an unresolvable callable sets convertable_to_code=False with a logged warning.
• @add_lazyframe_methods on FlowFrame (lazy_methods.py:136, applied at flow_frame.py class definition). Explicitly hand-defined methods always win over the injected ones. Injected pl.LazyFrame methods split in two:
  • PASSTHROUGH_METHODS (lazy_methods.py:9-26) — collect, collect_async, profile, describe, explain, show_graph, fetch, collect_schema, columns, dtypes, schema, width, estimated_size, n_chunks, is_empty, chunk_lengths, get_meta — delegate straight to self.data, no node added. Calling describe() really executes on the underlying LazyFrame right there.
  • Everything else → a generic wrapper emitting output_df = input_df.<method>(<repr'd args>) as a polars-code node (verified: df.tail(2) → output_df = input_df.tail(2); drop used to be one of these but is now hand-written). Every injected method also gains an extra description: str | None kwarg.
  • lazy_methods.py:108-110: if any argument has convertable_to_code=False, the wrapper short-circuits into a new source frame built from the eagerly-computed result — graph lineage is severed at that point (a genuine edge case, not exercised in the test suite as of this writing).

Lambda / function source extraction (flowfile_frame/flowfile_frame/callable_utils.py)

• Named functions: inspect.getsource (_get_function_source, callable_utils.py:19), dedented and stored verbatim.
• Lambdas: _extract_lambda_source (callable_utils.py:52) re-synthesizes a named def _lambda_fn_<hash % 100000>(args): return <body> via AST, matching by argument names, and captures closure variables as constant assignments or nested function defs; an unresolvable closure returns (None, None) → convertable_to_code=False.
• Verified live, run from a file (has retrievable source): ff.col("n").map_elements(lambda x: x + 1, return_dtype=ff.Int64) produced a polars-code node body:

  def _lambda_fn_83562(x):
      return x + 1
  #─────SPLIT─────
  
  output_df = input_df.with_columns([pl.col('n').map_elements(_lambda_fn_83562, return_dtype=Int64).alias('m')])

  The same lambda run from stdin/a REPL (no retrievable source via inspect.getsource) falls into the base64-serialized-LazyFrame fallback (§2, path 3) instead — both variants .collect() to the same data, but only the file-based one produces a portable, UI-editable node. Note return_dtype=Int64 has no pl. prefix in the generated code — it resolves only because core's sandbox injects bare dtype names into its namespace (§6).

fold and the generic Polars-function wrapper

lazy.py provides polars_function_wrapper (lazy.py:490) — decorator machinery that turns arbitrary pl.* functions into FlowFrame/Expr-returning functions by deep-repr'ing their arguments. fold (lazy.py:681-682, @polars_function_wrapper("fold", return_type="Expr")) is the shipped example; use it as the template if you need to wrap another module-level Polars function the same way.

Selectors (selectors.py)

Selector mirrors polars.selectors; its repr_str renders pl.selectors.<name>()-style strings. Selector aggregation methods (.sum(), .mean(), .std(ddof), …) build an Expr(expr=None, selector=…, agg_func=…) whose repr is <selector_repr>.<func>(…). Selectors always force the polars-code path in select/group_by/unpivot — there is no native-node route for a Selector.

5. Lazy semantics — exactly when does anything run?

Graph building is schema-eager, data-lazy — with several genuinely eager exceptions worth memorizing.

Baseline: every _create_child_frame/FlowFrame(...) call immediately invokes the new node's get_resulting_data() under a lock. For ordinary transforms this only chains LazyFrame operations (cheap, schema-resolution only) — confirmed above, every node logs "getting resulting data" at build time, not at .collect() time. .collect(*args, **kwargs) (flow_frame.py:2446) just collects the accumulated local LazyFrame and returns a pl.DataFrame — it does not invoke core's run machinery or talk to the worker.

Things that look lazy but are not (all verified live in this session):

Readers (read_csv, read_parquet, scan_*) are genuinely lazy scans; scan_csv/scan_parquet are literal aliases of read_csv/read_parquet. read_csv uses a native ReceivedTable-based node when parameters fit a documented allowlist, else falls back to generated pl.scan_csv code. Passing a BytesIO as the CSV source degrades all the way to from_dict(pl.read_csv(source)) — fully eager, because a BytesIO handle can't be re-scanned lazily by a saved graph.

cache() (flow_frame.py:2616) sets cache_results=True on the node settings (an execution hint the graph runner honors) and calls self.data.cache() immediately.

6. The generated-code contract with core's sandbox

Polars-code nodes execute inside flowfile_core/flowfile_core/flowfile/flow_data_engine/polars_code_parser.py's PolarsCodeExecutor. When writing or reviewing a generated string, it must satisfy:

• safe_globals (polars_code_parser.py:129): __builtins__ = {}. Available names: pl, cs (= pl.selectors), col, lit, expr, every bare Polars dtype name (Int64, String, Datetime, Struct, …, no pl. prefix needed), a small basic-builtins set (print, len, range, enumerate, zip, list, dict, set, str, int, float, bool, True, False, None), plus time, BytesIO, base64, datetime.
• _validate_code (polars_code_parser.py:191): rejects any code containing an import statement, blocked calls (exec, eval, getattr, open, …), dunder attribute access, and string constants matching a __word__ pattern.
• _wrap_in_function (polars_code_parser.py:239): code is wrapped in _transform(input_df) (or _transform(input_df_1, …, input_df_N) for multi-input nodes). A single-line expression starting with pl./col(/input_df/expr( is returned as-is; any other body must assign the name output_df.
• Input naming convention (mirrored by frame codegen): 1 input → input_df; ≥2 inputs → input_df_1..N, in upstream-discovery order. concat also dedupes duplicate sources because add_connection is idempotent.
• The #─────SPLIT───── marker separating prepended function definitions from the operation is purely a flowfile_frame-side cosmetic convention for the UI code editor — comments are stripped before execution, and grep confirms no consumer in core parses that marker.

The rule that ties all of §2–§6 together (quoted verbatim from flowfile_frame/CLAUDE.md): "Source string is load-bearing. … A wrong string breaks graph execution silently."