Proposal: Statement-level Python-to-Texera workflow compiler MVP

[carloea2]

Summary

This proposal adds a first end-to-end path that translates Python source into an ordinary Texera workflow of Python UDF operators. It reuses the existing compiler kernel and its working source model, dependency analysis, checking, realization, and rendering algorithms. The implementation is a smaller modular composition of that compiler, not a second compiler and not a rewrite from scratch.

The initial boundary is about where the compiler may split a program, not which normal Python statements a user may write.

What the compiler does

The user-facing pipeline has four stable stages:

Python source
    -> ANALYZE: understand complete statements and their dependencies
    -> GROUP: choose which statements should share an operator
    -> VERIFY: prove that every value is either local or can cross a boundary
    -> BUILD: generate Python UDFs, ports, links, and Texera workflow JSON

An evaluation atom is simply a set of program operations that cannot be separated safely. The initial composition closes those atoms under each complete top-level statement. Therefore the smallest placement unit visible to the grouping strategy is one complete statement.

Statement-level does not mean straight-line-only

Control flow, functions, and classes are not rejected merely because of their syntax. A complete top-level if, for, while, try, with, function definition, or class definition is admitted as one opaque placement unit. Python executes its body normally inside one generated operator.

What is deferred is decomposing the interior of those statements across multiple operators. Future modules may add the dependency and execution protocols required to split loops, calls, exceptions, or recursion without changing the grouping interface, checker, renderer, or workflow builder.

Invalid Python still fails during parsing, and a proposed operator boundary still fails verification when no registered realization can implement it.

Example

Given:

data = load("a.csv")
for row in data:
    clean(row)
model = train(data)
save(model)

the initial placement units are:

S1 = data = load("a.csv")
S2 = for row in data: ...       # one complete, indivisible unit
S3 = model = train(data)
S4 = save(model)

A grouping strategy may propose:

Operator A = {S1, S2}
Operator B = {S3, S4}

Verification then proves whether the value required downstream can cross from A to B. If it can, the selected boundary realization emits explicit export/import actions. If it cannot, that cut is illegal and a coarser legal grouping must be selected. The compiler never splits S2 or invents transport as a repair.

Conceptually, the generated workflow is:

Python UDF A                         Python UDF B
-------------------------------     --------------------------
data = load("a.csv")                 data = import boundary
for row in data:                     model = train(data)
    clean(row)                       save(model)
export boundary(data)       ---->   terminal result

Deterministic grouping strategies

The same analyzed program can be compiled with either strategy:

1. One statement per operator proposes the finest statement-level grouping and serves as a diagnostic/reference strategy.
2. Square-root contiguous grouping targets
   K = min(statement_count, ceil(sqrt(physical_LOC)))
   and chooses deterministic contiguous, acyclic groups close to equal physical-LOC partitions.

Neither strategy may override dependency, transport, or realizability constraints. One authoritative checker certifies the final proposal.

Internal architecture

source
  -> one parse: source inventory + existing Action/Parameter forest
  -> installed analysis modules
  -> one authoritative dependence graph
  -> evaluation atoms
  -> statement placement view
  -> grouping strategy
  -> expanded placement
  -> one checker
  -> selected internal and boundary realizations
  -> Action projections
  -> one UDF composer and one workflow renderer
  -> ordinary Texera workflow

The forest retains exact Python source and structural information. The dependence graph records which operations produce and consume semantic values. The statement placement view is the only solver-facing projection. The checker owns final legality. Realizations explain how certified local code and boundaries are materialized. The renderer only composes already-certified projections.

Boundary transport and Amber integration

• The PythonValue boundary realization determines exactly which required values cross each operator boundary.
• A shared generic PyTexera runtime performs export/import with Cloudpickle so aliases, cycles, and supported callables survive a process boundary.
• Generated UDFs import that runtime instead of embedding another runtime or serializer in every operator.
• The boundary envelope contains only explicitly selected fields; it never sends the complete Python namespace.
• The result uses Texera's ordinary Python source and tuple operators.
• No Amber scheduler, coordinator, recovery, materialization, worker, or protocol redesign is required.

MOSAIC owns source semantics, dependency analysis, grouping, and realization selection. Amber/PyTexera owns generic execution and transport primitives; it remains unaware of MOSAIC-specific atoms, carriers, colors, or solver rules.

Extensibility rule

Optional capabilities are installed as modules with explicit dependencies and contributions. A provider and the projector that interprets its facts are owned together. Adding support for distributed loops, calls, exceptions, files, consoles, or resources must add analysis evidence and/or a boundary method while preserving the same four-stage pipeline.

Feature-specific conditionals must not be scattered through dependence-graph construction, checking, or rendering, and no module may introduce a second authoritative graph, checker, or renderer.

End-to-end result: Wine classification

The current MVP was exercised with a realistic scikit-learn Wine classification program. This is the actual compiler input and the generated Python UDF code below is copied from the resulting Texera workflow JSON.

Result	Value
Grouping strategy	Square-root contiguous
Placement unit	Complete top-level statement
Generated workflow	1 entry operator + 8 Python UDF operators
Workflow links	12
Boundary payload	Only values required by downstream operators

Workflow screenshot: insert the Texera canvas capture here.

Wine input program

import numpy as np
import pandas as pd
from sklearn.datasets import load_wine
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.model_selection import train_test_split

wine = load_wine(as_frame=True)
frame = wine.frame.copy()
features = frame.drop(columns="target")
target = frame["target"].astype(np.int64)
class_profile = frame.groupby("target").mean(numeric_only=True)

lower_limits = features.quantile(0.01)
upper_limits = features.quantile(0.99)
curated = features.clip(lower=lower_limits, upper=upper_limits, axis=1)
curated["phenol_balance"] = curated["total_phenols"] / curated["nonflavanoid_phenols"].clip(lower=0.01)
curated["color_hue_ratio"] = curated["color_intensity"] / curated["hue"].clip(lower=0.01)
curated["alcohol_malic_interaction"] = curated["alcohol"] * curated["malic_acid"]

X_train, X_test, y_train, y_test = train_test_split(
    curated,
    target,
    test_size=0.25,
    random_state=42,
    stratify=target,
)

training_mean = X_train.mean(axis=0)
training_variance = X_train.var(axis=0, ddof=0)
training_scale = np.sqrt(training_variance).replace(0.0, 1.0)
X_train_scaled = (X_train - training_mean) / training_scale
X_test_scaled = (X_test - training_mean) / training_scale

model = RandomForestClassifier(
    n_estimators=300,
    max_depth=7,
    min_samples_leaf=2,
    random_state=42,
    n_jobs=-1,
)
model.fit(X_train_scaled, y_train)
prediction = model.predict(X_test_scaled)
class_probability = model.predict_proba(X_test_scaled)

quality_report = classification_report(y_test, prediction, output_dict=True, zero_division=0)
confusion = confusion_matrix(y_test, prediction)
accuracy = float(quality_report["accuracy"])
macro_f1 = float(quality_report["macro avg"]["f1-score"])
mean_confidence = float(np.mean(np.max(class_probability, axis=1)))

importance = pd.Series(model.feature_importances_, index=X_train_scaled.columns, name="importance")
top_features = importance.nlargest(8)
summary = pd.DataFrame({
    "metric": ["accuracy", "macro_f1", "mean_confidence"],
    "value": [accuracy, macro_f1, mean_confidence],
})
print(summary.to_string(index=False))
print(top_features.to_string())
print(confusion)

The generated DAG exposes parallel branches where the statement dependencies permit them and joins those branches through ordinary Texera input ports. Port labels use the producer identifier (for example, 001, 004, and 007) rather than internal operator class names.

Representative generated UDFs

These are complete code fields from three operators in the generated workflow: the initial Wine loader, a two-input training/preprocessing operator, and the terminal reporting operator.

UDF 001 — imports and dataset loading

import pytexera.workflow as _mosaic_workflow_module_0

_mosaic_runtime_0 = _mosaic_workflow_module_0.Runtime(
    input_ports=(),
    outgoing=('boundary_0000', 'boundary_0001', 'boundary_0002', 'boundary_0003', 'boundary_0004', 'boundary_0005'),
)

@_mosaic_runtime_0.driver
def _mosaic_driver_0(_mosaic_heap_0):
    import numpy as np
    import pandas as pd
    from sklearn.datasets import load_wine
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.metrics import classification_report, confusion_matrix
    from sklearn.model_selection import train_test_split
    wine = load_wine(as_frame=True)
    _mosaic_boundary_0 = 'boundary_0000'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    _mosaic_boundary_0 = 'boundary_0001'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ('wine',), locals())
    _mosaic_boundary_0 = 'boundary_0002'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    _mosaic_boundary_0 = 'boundary_0003'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    _mosaic_boundary_0 = 'boundary_0004'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    _mosaic_boundary_0 = 'boundary_0005'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    return _mosaic_heap_0

class _mosaic_operator_0(_mosaic_workflow_module_0.TupleOperator):
    runtime = _mosaic_runtime_0

UDF 004 — dependency join, train/test split, and export

import pytexera.workflow as _mosaic_workflow_module_0

_mosaic_runtime_0 = _mosaic_workflow_module_0.Runtime(
    input_ports=(_mosaic_workflow_module_0.InputPort(boundaries=('boundary_0003',)), _mosaic_workflow_module_0.InputPort(boundaries=('boundary_0008', 'boundary_0009')),),
    outgoing=('boundary_0010', 'boundary_0011', 'boundary_0012'),
)

@_mosaic_runtime_0.driver
def _mosaic_driver_0(_mosaic_heap_0):
    _mosaic_boundary_0 = 'boundary_0003'
    _mosaic_runtime_0.import_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    from sklearn.model_selection import train_test_split
    _mosaic_heap_0.train_test_split = train_test_split
    _mosaic_boundary_0 = 'boundary_0008'
    _mosaic_runtime_0.import_boundary(_mosaic_heap_0, _mosaic_boundary_0, ('curated', 'target'))
    _mosaic_boundary_0 = 'boundary_0009'
    _mosaic_runtime_0.import_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    X_train, X_test, y_train, y_test = _mosaic_heap_0.train_test_split(
        _mosaic_heap_0.curated,
        _mosaic_heap_0.target,
        test_size=0.25,
        random_state=42,
        stratify=_mosaic_heap_0.target,
    )
    training_mean = X_train.mean(axis=0)
    _mosaic_boundary_0 = 'boundary_0010'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ('X_train', 'training_mean'), locals())
    _mosaic_boundary_0 = 'boundary_0011'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    _mosaic_boundary_0 = 'boundary_0012'
    _mosaic_runtime_0.export_boundary(_mosaic_heap_0, _mosaic_boundary_0, ('X_test', 'y_test', 'y_train'), locals())
    return _mosaic_heap_0

class _mosaic_operator_0(_mosaic_workflow_module_0.TupleOperator):
    runtime = _mosaic_runtime_0

UDF 008 — terminal reporting

import pytexera.workflow as _mosaic_workflow_module_0

_mosaic_runtime_0 = _mosaic_workflow_module_0.Runtime(
    input_ports=(_mosaic_workflow_module_0.InputPort(boundaries=('boundary_0017', 'boundary_0018')),),
    outgoing=(),
)

@_mosaic_runtime_0.driver
def _mosaic_driver_0(_mosaic_heap_0):
    _mosaic_boundary_0 = 'boundary_0017'
    _mosaic_runtime_0.import_boundary(_mosaic_heap_0, _mosaic_boundary_0, ('confusion', 'summary', 'top_features'))
    _mosaic_boundary_0 = 'boundary_0018'
    _mosaic_runtime_0.import_boundary(_mosaic_heap_0, _mosaic_boundary_0, ())
    print(_mosaic_heap_0.summary.to_string(index=False))
    print(_mosaic_heap_0.top_features.to_string())
    print(_mosaic_heap_0.confusion)
    return _mosaic_heap_0

class _mosaic_operator_0(_mosaic_workflow_module_0.TupleOperator):
    runtime = _mosaic_runtime_0

This example demonstrates compilation and workflow construction. Runtime performance claims require separate controlled execution measurements and are not inferred from the graph shape.

Validation plan

The initial implementation will cover:

• complete-statement atomicity for assignments, expressions, control flow, functions, and classes;
• statement-unit closure and deterministic grouping;
• positive and negative boundary-realization cases;
• exact required-value and Cloudpickle cross-process transport;
• semantic parity between original Python execution and the generated workflow;
• a real Amber integration test using ordinary Python UDF operators;
• a representative data-science program, with the Wine pipeline expected to produce roughly 7–10 operators under square-root contiguous grouping.

Non-goals

This first integration does not distribute the interior of control-flow statements, functions, classes, recursion, exceptions, or individual expressions. It also excludes ML-based grouping, whole-namespace transport, and an Amber engine redesign.

The existing complex-case compiler work remains the reference for later modules. It is not being discarded or reimplemented.

Questions for review

1. Is complete top-level statement atomicity the right first placement boundary?
2. Should both deterministic grouping strategies be exposed initially, or should one-statement-per-operator remain diagnostic only?
3. Is explicit required-value transport through the shared Cloudpickle runtime sufficient for the first integration?
4. Which end-to-end Python examples should block the first PR?

[chenlica]

Per our offline discussions, I support this design. Here are my answers to the questions:

1. Yes.
2. Either way is fine. We can choose one based on the quality of generated workflows. Deciding a good strategy requires more future effort.
3. Yes.
4. We can select some sample Python programs, e.g., those from GitHub.