test(verify): run a Python operator the way the engine runs it - #8357

A Python operator is not called; it is handed to an interpreter that imports
the generated module and drives it through the same open, process and close
the engine uses. `PyOpExecHarness` writes that module, starts the driver, and
reads back what the operator emitted.

The driver is the engine's side of the contract written out plainly: it is
what makes the answer this side produces the engine's answer rather than an
approximation of it.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

Eight lines this change is already in the file for. sbt lifts a `.value`
written inside a lambda to the top of the task, so the options were read once
rather than per suite either way; written where they were, they said the other
thing, and sbt warned on it in every build.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

The two helpers here were a second copy of OpExecHarness's, carrying a note that
they were kept inline so each harness read on its own and would be consolidated
if a third arrived. Two copies of the same twenty lines is already the cost that
note was deferring: they have to be changed together, and nothing says so at
either site.

Preparing the plan does not vary with the executor, so it happens once now. What
differs between the harnesses stays here: how the prepared op is run.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

`PhysicalPlan` is named only by `getPhysicalPlan` and by a comment, so scalafix
reports the import and the check fails before any test runs.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

The driver projected the operator's output onto the schema's attribute
names and coerced each value with int/float/str. That is looser than the
engine, which hands every yielded value to all_output_to_tuple and
finalizes the result: an operator putting 1.5 in an INTEGER field, or a
row with a field the schema does not have, or one missing a field, all
raise there and none of the three raised here. A reference path that
accepts output the engine rejects cannot say what the engine would do,
so call the engine's own two functions instead of re-deriving them.

finalize pickles a BINARY field's object itself and tags it, where the
standalone side pickles the same object untagged, so drop the tag on the
way out and both sides carry the same payload.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

The driver collected everything the operator yielded and converted the
list afterwards. `DataProcessor` reads a yield where it happens, turning
it into tuples and finalizing them on the spot, so a UDF that yields one
dict twice and changes it in between is recorded as the two values it
produced. The driver recorded the last one twice.

The operator now takes a callback and hands each value over at the
yield, before the generator is advanced.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

apache#8488 is closed: a table with no rows carries no schema by
design, so there is nothing for the driver to hand the operator and no
engine behavior here to match.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

The worker's ArrowTableTupleProvider unpickles a BINARY cell that starts
with the cast's pickle marker, so a model column reaches the next
operator as a model. The driver handed the operator the bytes instead,
and a predict on them raised AttributeError.

Decoding the cell alone was not enough. The driver then finalized each
input tuple, and the cast in finalize pickles any non-bytes value back
into bytes. The worker's InputManager attaches the schema without
finalizing, so the driver now builds input tuples the same way.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

…and refuse a row the schema does not describe

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Co-Authored-By: Claude Opus 5.5 <[email protected]>

Windows drops the quotes from a multiline `-c` argument, so the bytes
literal reached Python as `bpickle` and the fixture failed before the
harness ran. Write the script to a temporary .py file and run that.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

As the worker's EndChannelHandler does since apache#8506: a HashJoin probe that
received no build rows names its right columns from the build port's
declared schema.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…iter does

json.dumps writes a bare Infinity, which is not JSON; the engine's writer spells
it as text. A NaN still leaves as null, as the worker casts it.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

…rator-python

# Conflicts:
#	build.sbt

…rator-python

# Conflicts:
#	build.sbt