GitHub user carloea2 edited a discussion: Proposal: Statement-level
Python-to-Texera workflow compiler MVP
## 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:
```text
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:
```python
data = load("a.csv")
for row in data:
clean(row)
model = train(data)
save(model)
```
the initial placement units are:
```text
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:
```text
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:
```text
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
```text
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.
<details>
<summary>Wine input program</summary>
```python
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)
```
</details>
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.
<details>
<summary>UDF 001 — imports and dataset loading</summary>
```python
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
```
</details>
<details>
<summary>UDF 004 — dependency join, train/test split, and export</summary>
```python
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
```
</details>
<details>
<summary>UDF 008 — terminal reporting</summary>
```python
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
```
</details>
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?
GitHub link: https://github.com/apache/texera/discussions/8160
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
