feat: bind number and true/false loop variables in generated Python operators - #9359

A property of an operator inside a loop block (LoopStart ... LoopEnd) may
name a loop variable as `$K`. The frontend keeps sending the literal
string whatever the property's type; the backend binds it in three steps.

Parse. LogicalOp is now a StateReferencing descriptor, with a
`stateReferences` sidecar (JSON pointer -> variable name) that the
property panel never shows. StateReferenceModule, registered on
JSONUtils.objectMapper and on the workflow-compiling-service mapper, adds
a Jackson problem handler: when a whole-string `$name` cannot be converted
to an integral, floating-point or boolean property, it returns the
placeholder 0 / 0.0 / false and records the value's pointer, relative to
the operator, in the operator's sidecar. A thin wrapper around each
concrete StateReferencing deserializer opens that recording frame. An
object that a polymorphic type deserializer hands over mid-way is first
copied into a buffer of its own, because Jackson replays the fields before
a late type id from a buffer whose parse context copies the first field's
value, not the object. String properties keep the literal, and any other
target (an enum, ...) keeps Jackson's ordinary error.

Compile. WorkflowCompiler works out which operators lie inside a loop
block (LoopBlockMembership, the frontend's block walk) and makes the
sidecar final before the physical plan serializes it into the executor's
descString. Inside a block it adds every whole-string `$name` of the
operator's JSON. Outside every block a string such as "$AAPL" stays the
literal it is on main, and a typed placeholder is a compile error naming
the property and the variable. Inside a block, an operator whose executor
runs generated code, or with an input fed from outside the block, is a
compile error too.

Worker. ExecFactory hands out a LateBoundExecutor when the descString's
sidecar is not empty. It keeps and forwards each state message, and binds
when the real executor is first needed, at the first tuple or finishing
port: the channel has delivered its state messages by then, so their
values are merged with the last one winning, and an inner loop's variable
shadows an outer one of the same name. It patches the placeholders,
coerced to their JSON type (a string property takes only a string, a
number or a boolean), builds the real executor through the normal factory
path, opens it and replays the kept state messages into it in order, so an
executor that acts on them (If) behaves as without references. Every later
call goes to that executor, which is never rebound. A variable still
missing then is an error naming it.

Closes apache#8635

The attribute the runtime registers on every operator right before
process_state is now `state` instead of `loop_state`.

LoopStartOperator and LoopEndOperator already used `self.state` for their
own loop variables, so registering the incoming message there would have
merged it into itself (LoopStart) and made the back-edge write the message
instead of the updated variables (LoopEnd). Their own field is renamed to
`variables`, and the back-edge in MainLoop._jump_to_loop_start now writes
`executor.variables`. Loop operator behavior is unchanged.

Two new tests pin the split: a LoopStart run through
DataProcessor.process_state keeps its variables apart from the registered
state, and the LoopEnd back-edge writes the updated variables rather than
the registered state.

The attribute the runtime registers before process_state is now `state`
(apache#8642), so the method that registers it follows: Operator.register_loop_state
is now Operator.register_state. DataProcessor.process_state and the deferred
consume in MainLoop call it under the new name, and its docstring and the
tests say `state` / `register_state` throughout. The loop-variable reads,
loop_variable_text and loop_variable_value, and their _loop_variables are
unchanged.

…-loop-refs

…nding gaps

SplitOpExec.open() resets its generator again, as before the simplification,
so a reopened executor repeats its seeded split; the reseed test is back.

New tests, each checked against a mutant that the old suites let through:
- a reference on an operator fed by another body operator compiles
  (LoopStart -> Projection -> Limit($n) -> LoopEnd);
- the compiling service's Dropwizard mapper parses a typed "$n";
- a control operator is not inside its own block when an outer block is
  left open, in both directions;
- ExecFactory finds a setting declared only in a superclass, and tells two
  equal settings apart by identity;
- a sidecar pointer that is not a JSON pointer is reported at binding.

The Filter($i) test runs the compiled executor again.

…-loop-refs

…ock around it

The outside-input check counted any operator inside some block as carrying the state, so an
inner block's operator could read first from the outer body or a sibling block, whose tuples
come before the inner loop's state. LogicalPlan.enclosingLoopStarts gives the LoopStarts
around each operator, checked against the panel's walk, and a link carries the state only when
its source is inside all of the operator's blocks or opens the innermost. Also pins an input
with one link from inside the block and one from outside.

…-loop-refs

- LogicalPlan: derive the operators inside loop blocks from
  enclosingLoopStarts, so the block walk runs once and is cached.
- WorkflowCompiler: normalizeStateReferences reads membership from the
  plan instead of a separate parameter.
- OperatorExecutor: two weak maps (states, bindings) replace the
  Registration holder; StateReferenceBinding.bind() replaces `unbound`.
- Shorter docs on ExecFactory and JSONUtils.registerWorkflowModules.
- Tests: set-only port of the panel walk, merged overlapping compiler,
  binding and mapper cases, EndChannelHandlerSpec reuses the
  DataProcessorSpec fixture, and the single-loop Filter($i) run is
  dropped (the branching run covers it). Pin a variable written into
  every property that names it, and a refused message writing and
  registering nothing.

…-loop-refs

# Conflicts:
#	amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala

…fs-backend

# Conflicts:
#	amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala

…perators

A number or true/false property of an operator whose Scala descriptor
generates Python can now refer to a loop variable inside a control
block. The code is generated before the loop runs, so
PythonOperatorDescriptor.loopVariableBinding generates it from copies of
the descriptor with probe values in the reference's place: 3 and 5, then
70 and 90 for an integer; 0.25 and 0.75, then 2.5 and 7.5 for a number;
True and False. Where the two codes differ by exactly the two values,
and nowhere else, the code calls self.loop_variable_value(name, kind)
there instead: 'integer' for an int property, 'long' for one that takes
a whole number past 32 bits. The value must also land as it is at 0, 1,
-1, 1000000 and the other probes. When no probe pair generates the
code, the properties' defaults and the values next to them are tried,
so a descriptor's own check never reports a probe value. A value the
descriptor checks or adjusts, one that lands in a text, a comment or a
name, one the code computes with, never holds or reads when the
operator starts (in open or __init__, before any state message), and
one that decides the output schema leave the reference a compile error
that names the property and why.

WorkflowCompiler probes the code once per compile: the physical
operator runs that binding's code, and unbindableStateReferences
decides every reason from the same binding, the code's before the
schema's, and the schema one only once every input has a schema, as
the schema propagation does.

Operator.loop_variable_value reads the variable's text as
loop_variable_text spells it, from the values bound at the operator's
first tuple, and converts it as a JVM operator's property binds it
(StateReferenceBinding.coerce): trimmed as String.trim, an integer as
new BigDecimal(text).longValueExact() reads it, and within 32 bits for
an int property, as Jackson writes it there; a number as
Double.parseDouble does (2.5d and 0x1p3, but not inf, nan, 1_000 or
fullwidth digits), and a boolean as true or false in any case. A whole
double past 10^16 reads as the whole number its fewest digits spell, as
JDK 19 and later bind it; JDK 17 spells it with other digits
(JDK-4511638). An integer past 64 bits is refused, as
loop_variable_text refuses it.

The executor manager also hands each Python operator the values its
code reads through loop_variable_value, so binding fails, naming each,
for a required value no state message carried or one its property
cannot take, as the JVM's binding does, whether or not the code reads
the value then.

…o values

Since main's apache#8642, reading a name that got two different values from one
depth fails. bind_loop_variables copies loop_variables with dict(), which reads
every name, so a generated operator fails at its first tuple, or when an input
ends, as soon as any name has two values, even one no property refers to. A JVM
operator fails only for a name its setting refers to. main's
test_main_loop_thread_can_process_state (four values of one name, then
EndChannel) fails the same way.

Pin that binding fails only for a required name, listed with the other
problems, and that the bound values keep the two values: reading that name
fails, the other names read, and a deeper value before binding replaces the
two.

… values

bind_loop_variables now copies the values and the names that got two different
values, as loop_variables holds them, instead of reading every name through
dict(). Binding fails for a required name with two values, listed with the
other problems, and reading such a name from the bound values fails with the
same error, while the other names read.

…ber-refs

loop_variable_value reads a variable's text through _loop_variable_text,
and binding checks each required value the same way, so since
feat/generated-op-loop-refs merged in, both follow its rule for a name
that got two different values from one depth: binding fails only for a
required name, listed with the other problems in order, and the bound
values keep the two values.

Pin it for numeric and boolean properties: an operator whose code reads
only n binds and converts n while i has two values; a required i with
two values fails binding with StateReferenceBinding's message whether or
not the values fit the property's kind, as a JVM operator fails before
it converts either; reading i as any kind fails before and after
binding, and still after a deeper i arrives; the other names convert as
before; a deeper value before binding replaces the two.

The merge commit already passes these tests, so no source change
follows.

Binding checks every read of a loop variable the generated code makes:
once as a text, and once for each kind it is read as. A name that got
two different values from one depth fails each of those reads with the
same error, so code that reads it as an integer, a number and a text
lists StateReferenceBinding's message three times. A JVM operator
throws that message once for the name (StateReferenceBinding.write goes
over each referenced name once).

Pin that binding names the two values once, where the name is first
read: for code that reads n three ways, for a text and several kinds,
and beside a missing and a refused value, which keep their place. A
value that two kinds refuse is still named once for each kind.

bind_loop_variables now names a required loop variable that got two
different values once, where the code first reads it: a value read of
that name adds StateReferenceBinding's message only when no text or
earlier kind has added it. Every read is still checked, and missing and
refused values are named as before.

… the loop runs

Inside a loop block, SVM Regressor took a Ground Truth Attribute Column
"$y" beside the Selected Feature "Unit Price". Its check that the ground
truth is no feature runs in getOutputSchemas, before the loop runs, so
it saw "$y", and when y was "Unit Price" the run fitted Unit Price on
itself, which the operator refuses outside a loop. Sklearn Prediction
failed a ground truth "$g" with "Ground Truth column '$g' is not in the
input table", and Machine Learning Scorer an Actual Value "$a" the same
way, neither saying that a loop variable cannot go there.

These cases pin the reason "it checks the text before the loop runs"
for the three, a feature "$f", a classic trainer's target with Count
Vectorizer on and both of the Scorer's columns included, and "the text
decides its output schema" for a test-only operator that types its
output column like the input column its text names. The same operator
pins a check in its code generation against another of its texts, one
naming an input column and one naming none; a check against the empty
text, which no copy tries, so the text stays bound; and a null setting,
which the copies keep rather than take the property's default. As the
schema propagation, the check waits until every input has a schema.

…the loop runs

PythonOperatorDescriptor.textsReadBeforeTheLoop tells which of a
generated operator's text references it checks, or computes its output
schemas from, before the loop runs. It builds copies of the descriptor
that hold another text in the reference's place: the first column of
each type on each input, and each other non-empty text among its
settings, which a check commonly compares the text to. Where only one
of a copy and the copy it is compared with computes its output schemas,
or the copy fails to generate its code where the other does not, the
operator checks the text; where both compute them and they differ, the
text decides them. The copy compared with holds every reference's
"$name", unless its schemas fail; then it is the first copy that
computes them with such texts in the references' places, so that one
reference that fails a check cannot hide another's.

WorkflowCompiler asks it once every input has a schema, for the texts
the generated code would otherwise bind, and refuses them with "it
checks the text before the loop runs" or "the text decides its output
schema", much as apache#9359 refuses a number it checks or computes its
schema from.

Inside a loop this refuses the ground truth and Selected Features of
the KNN Classifier, KNN Regressor, SVC and SVR trainers, and the Target
and Text Attribute of the classic sklearn trainers with Count Vectorizer
on, which each check against the other. Sklearn Prediction's ground
truth and the Machine Learning Scorer's columns, which failed before
with "... is not in the input table", now get this reason. Sweeping
every generated operator, one text property at a time, finds no other
text it refuses.

…ber-refs

On this branch a generated operator's code and output schemas come from
the binding's baseline copy, which holds a probe value for each numeric
or boolean reference, while the descriptor holds its placeholder (0,
0.0 or false). apache#8671's check of the texts runs on the descriptor, so a
placeholder that fails the operator's schema makes every copy fail
alike and hides the check of a text. This case pins it with a test-only
operator whose schema requires its ratio, "$r", to be positive and its
column, "$c", not to be "total": "$c" is refused, and with nothing to
avoid both references are read from the loop state.

LoopVariableBinding now holds the copy its code and output schemas come
from, the descriptor itself or the probing's baseline, in place of that
copy's schema function, and asks that copy which text references it
checks or computes its schemas from (textsReadBeforeTheLoop). The
compiler asks the binding rather than the descriptor, whose
placeholders for the numeric and boolean references may fail a check
whatever the text is.

…p-refs

…ber-refs