feat(workflow-operator): export the join and set operators as Python

amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala

@@ -40,6 +40,9 @@ trait EndChannelHandler {
     val portId = dp.inputGateway.getChannel(channelId).getPortId
     dp.inputManager.getPort(portId).completed = true
     dp.inputManager.initBatch(channelId, Array.empty)
+    // Before the operator finishes the port, so one that received no rows on
+    // it can still say what its columns were.
+    dp.executor.inputSchemas(portId.id) = dp.inputManager.getPort(portId).schema
     try {
       // The executor finishes: a setting that refers to loop variables must have them by now.
       dp.bindStateReferences()

amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala

@@ -236,6 +236,18 @@ class EndChannelHandlerSpec extends AnyFlatSpec {
     assert(executor.finishOutputPorts.toList == List(currentPortId.id))
   }
 
+  it should "give the operator the finished port's declared schema before it finishes the port" in {
+    // A HashJoin probe worker that received no build rows names its unmatched right rows from
+    // this, so it has to be in place when the operator is asked to finish, not after.
+    val executor = new RecordingExecutor()
+    val fixture = new Fixture(executor, otherInputPortCompleted = false)
+
+    fixture.endChannel()
+
+    assert(executor.schemasAtFinish.toList == List(Map(currentPortId.id -> schema)))
+    assert(!executor.inputSchemas.contains(otherPortId.id))
+  }
+
   it should "emit the boundary state the operator produced on finish, with no loop envelope" in {
     val state = State(Map("boundary" -> "end"))
     val fixture = new Fixture(new RecordingExecutor(onFinishState = _ => Some(state)))
@@ -464,8 +476,12 @@ object EndChannelHandlerSpec {
     /** Ports handed to `onFinish`, i.e. what `onFinishMultiPort` forwarded. */
     val finishOutputPorts: ArrayBuffer[Int] = ArrayBuffer()
 
+    /** `inputSchemas` as the first finish callback saw it. */
+    val schemasAtFinish: ArrayBuffer[Map[Int, Schema]] = ArrayBuffer()
+
     override def produceStateOnFinish(port: Int): Option[State] = {
       statePorts += port
+      schemasAtFinish += inputSchemas.toMap
       onFinishState(port)
     }
 

common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala

@@ -21,9 +21,11 @@ package org.apache.texera.amber.core.executor
 
 import com.google.common.collect.MapMaker
 import org.apache.texera.amber.core.state.State
-import org.apache.texera.amber.core.tuple.{Tuple, TupleLike}
+import org.apache.texera.amber.core.tuple.{Schema, Tuple, TupleLike}
 import org.apache.texera.amber.core.workflow.PortIdentity
 
+import scala.collection.mutable
+
 /**
   * Inside a loop block a setting may refer to a loop variable (`$i`): the worker writes each
   * iteration's value into the settings object the executor holds, in place (`registerState`). For
@@ -91,6 +93,17 @@ trait OperatorExecutor {
     */
   final def bindStateReferences(): Unit =
     Option(OperatorExecutor.bindings.get(this)).foreach(_.bind())
+
+  /**
+    * What each finished input port was declared to carry, keyed by port index
+    * and written by the runtime before `onFinish` for that port.
+    *
+    * The tuples say this too, so an operator only needs to ask when there were
+    * none: a port can finish having carried no rows, and its schema is then the
+    * only record of what its columns were.
+    */
+  final def inputSchemas: mutable.Map[Int, Schema] =
+    OperatorExecutor.inputSchemas.computeIfAbsent(this, _ => mutable.Map())
 }
 
 object OperatorExecutor {
@@ -100,4 +113,6 @@ object OperatorExecutor {
   private val states = new MapMaker().weakKeys().makeMap[OperatorExecutor, State]()
   private[executor] val bindings =
     new MapMaker().weakKeys().makeMap[OperatorExecutor, StateReferenceBinding]()
+  private val inputSchemas =
+    new MapMaker().weakKeys().makeMap[OperatorExecutor, mutable.Map[Int, Schema]]()
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SamplingHelpers.scala

@@ -25,18 +25,12 @@ package org.apache.texera.amber.operator
 object SamplingHelpers {
 
   /**
-    * A Python transcription of `java.util.Random`, for operators whose executor
-    * draws from one.
-    *
-    * A sampler decides per row whether to keep it, so which rows survive is
-    * fixed by the exact sequence the generator produces. Seeding Python's
-    * `random` or numpy's with the engine's seed selects a different set, and
-    * the script would then report a different sample than the workflow it came
-    * from. Only the same generator gives the same rows.
+    * `_TexeraJavaRandom`: `java.util.Random` in Python, so a sampler keeps the rows the engine
+    * keeps. Python's or numpy's generator with the same seed picks other rows. `next_int`
+    * rejects a draw where Java's int sum overflows, which Python's integers never do.
     */
   val JavaRandom: String =
-    """# java.util.Random, transcribed so sampling matches the engine.
-      |class _TexeraJavaRandom:
+    """class _TexeraJavaRandom:
       |    _MASK = (1 << 48) - 1
       |    _MULTIPLIER = 0x5DEECE66D
       |    _ADDEND = 0xB
@@ -60,8 +54,6 @@ object SamplingHelpers {
       |        while True:
       |            bits = self._next(31)
       |            value = bits % bound
-      |            # Java rejects a draw by letting this sum overflow an int.
-      |            # Python would carry it and never reject.
       |            probe = bits - value + (bound - 1)
       |            if ((probe + (1 << 31)) % (1 << 32)) - (1 << 31) >= 0:
       |                return value""".stripMargin

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/cartesianProduct/CartesianProductOpDesc.scala

@@ -23,10 +23,11 @@ import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{Attribute, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class CartesianProductOpDesc extends LogicalOp {
+class CartesianProductOpDesc extends LogicalOp with StandaloneCodeGenerator {
+
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -103,4 +104,27 @@ class CartesianProductOpDesc extends LogicalOp {
       ),
       outputPorts = List(OutputPort())
     )
+
+  // Schema mirrors SchemaPropagationFunc: left columns kept as-is, each right
+  // column renamed by repeatedly appending "#@1" while the candidate name
+  // collides with any left column OR any other right column's ORIGINAL name.
+  // The renamed-name table is recomputed at runtime from the actual DataFrame
+  // columns. Known divergence: row order — pandas cross-merge varies right
+  // fastest (L1R1, L1R2, L2R1, L2R2); the JVM op buffers left and emits per
+  // arriving right tuple (L1R1, L2R1, L1R2, L2R2). Cartesian product is set-
+  // semantically order-agnostic, so this is acceptable.
+  override def generateStandaloneCode(): String = {
+    """_left_cols = list(in1df.columns)
+      |_right_cols = list(in2df.columns)
+      |_left_set = set(_left_cols)
+      |_right_set = set(_right_cols)
+      |_rename = {}
+      |for _col in _right_cols:
+      |    _new = _col
+      |    _others = _right_set - {_col}
+      |    while _new in _left_set or _new in _others:
+      |        _new = _new + "#@1"
+      |    _rename[_col] = _new
+      |out1df = in1df.merge(in2df.rename(columns=_rename), how="cross").reset_index(drop=True)""".stripMargin
+  }
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/difference/DifferenceOpDesc.scala

@@ -23,11 +23,10 @@ import com.google.common.base.Preconditions
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 
-class DifferenceOpDesc extends LogicalOp {
-
+class DifferenceOpDesc extends LogicalOp with StandaloneCodeGenerator {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -61,4 +60,13 @@ class DifferenceOpDesc extends LogicalOp {
       ),
       outputPorts = List(OutputPort(blocking = true))
     )
+
+  // Distinct left (port 0) rows the right (port 1) does not hold, by the
+  // engine's row equality, which `_texera_key_codes` with `signed` numbers.
+  override def generateStandaloneCode(): String =
+    """_texera_set_left, _texera_set_right = _texera_key_codes(in1df, in2df, signed=True)
+      |_texera_set_kept = ~_texera_set_left.duplicated() & ~_texera_set_left.isin(_texera_set_right)
+      |out1df = in1df[_texera_set_kept.to_numpy()].reset_index(drop=True)""".stripMargin
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala

@@ -23,20 +23,21 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.state.NoLoopVariable
-import org.apache.texera.amber.core.tuple.{Attribute, Schema}
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{
   ExecutionIdentity,
   PhysicalOpIdentity,
   WorkflowIdentity
 }
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME
 import org.apache.texera.amber.operator.metadata.annotations.{
   AutofillAttributeName,
   AutofillAttributeNameOnPort1
 }
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 object HashJoinOpDesc {
@@ -54,8 +55,7 @@ object HashJoinOpDesc {
   }
 }
 """)
-class HashJoinOpDesc[K] extends LogicalOp {
-
+class HashJoinOpDesc[K] extends LogicalOp with StandaloneCodeGenerator {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Left Input Attribute")
   @JsonPropertyDescription("attribute to be joined on the Left Input")
@@ -158,19 +158,19 @@ class HashJoinOpDesc[K] extends LogicalOp {
               buildSchema.getAttributes.filterNot(_.getName == HASH_JOIN_INTERNAL_KEY_NAME)
             val leftAttributeNames = leftAttributes.map(_.getName).toSet
 
-            // Filter and rename attributes from the probe schema to avoid conflicts
+            // Filter and rename attributes from the probe schema to avoid conflicts,
+            // by the rule the executor names its fields with. Checking the left
+            // side alone declared two columns under one name when the right side
+            // already carried the suffixed one, and the row, matched to the schema
+            // by name, lost the value the executor had put under the other name.
+            val rightNames = probeSchema.getAttributeNames.filterNot(_ == probeAttributeName)
             val rightAttributes = probeSchema.getAttributes
               .filterNot(_.getName == probeAttributeName)
               .map { attr =>
-                var newName = attr.getName
-                while (leftAttributeNames.contains(newName)) {
-                  val suffixIndex = """#@(\d+)$""".r
-                    .findFirstMatchIn(newName)
-                    .map(_.group(1).toInt + 1)
-                    .getOrElse(1)
-                  newName = s"${attr.getName}#@$suffixIndex"
-                }
-                new Attribute(newName, attr.getType)
+                new Attribute(
+                  JoinUtils.renamed(attr.getName, leftAttributeNames.toSeq, rightNames),
+                  attr.getType
+                )
               }
 
             // Combine left and right attributes into a new schema
@@ -203,4 +203,131 @@ class HashJoinOpDesc[K] extends LogicalOp {
       ),
       outputPorts = List(OutputPort())
     )
+
+  // Equi-join: the probe key is dropped and a colliding right column takes
+  // "#@1", as JoinUtils names it. Rows come out in the merge's order rather
+  // than the engine's.
+  /** Only the declared type can say which columns an outer join widened, so
+    * without a schema the widening stands.
+    */
+  override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = {
+    val integral = (a: Attribute) =>
+      a.getType == AttributeType.INTEGER || a.getType == AttributeType.LONG
+    val declaredIntegers = (port: PortIdentity) =>
+      inputSchemas.get(port).map(_.getAttributes.filter(integral).map(_.getName)).getOrElse(List())
+    // Each side keeps its own list, under the names that side's frame carries,
+    // so a column the merge renames is still the column its own schema typed.
+    if (joinType == JoinType.INNER) generateStandaloneCode()
+    else
+      generateStandaloneCode(
+        declaredIntegers(operatorInfo.inputPorts.head.id),
+        declaredIntegers(operatorInfo.inputPorts.last.id)
+      )
+  }
+
+  override def generateStandaloneCode(): String =
+    generateStandaloneCode(List(), List())
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
+
+  private def generateStandaloneCode(
+      leftIntegerColumns: List[String],
+      rightIntegerColumns: List[String]
+  ): String = {
+    val buildKeyLit = objectMapper.writeValueAsString(buildAttributeName)
+    val probeKeyLit = objectMapper.writeValueAsString(probeAttributeName)
+    val how = joinType match {
+      case JoinType.INNER       => "inner"
+      case JoinType.LEFT_OUTER  => "left"
+      case JoinType.RIGHT_OUTER => "right"
+      case JoinType.FULL_OUTER  => "outer"
+    }
+    // An unmatched row leaves a hole, and a hole costs a pandas integer column
+    // its type: int64 becomes float64, which rounds every value past 2^53
+    // before anything can put the type back. The engine writes a null and
+    // leaves the column INTEGER, so widen to the integer dtype that holds a
+    // hole before the merge digs one.
+    val widen = leftIntegerColumns.nonEmpty || rightIntegerColumns.nonEmpty
+    val namesLit = (names: List[String]) => names.map(pyStringLiteral).mkString("[", ", ", "]")
+    val widening =
+      if (!widen) ""
+      else
+        s"""_left_ints = {_c: "Int64" for _c in ${namesLit(leftIntegerColumns)} if _c in in1df.columns}
+           |_right_ints = {_c: "Int64" for _c in ${namesLit(rightIntegerColumns)} if _c in in2df.columns}
+           |""".stripMargin
+    val leftFrame = if (widen) "in1df.astype(_left_ints)" else "in1df"
+    // Cast before the rename, because the names above are the ones the right
+    // input declared.
+    val rightFrame =
+      if (widen) "in2df.astype(_right_ints).rename(columns=_rename)"
+      else "in2df.rename(columns=_rename)"
+    // HashJoinProbeOpExec's rename, written out: append "#@1" until the name is
+    // free. pandas' `suffixes` appends once and then refuses the duplicate it
+    // just made.
+    //
+    // JoinUtils sets the probe key aside before it renames anything, so a
+    // payload column never has to step around the key that is about to leave.
+    // The key itself still rides along, and it only has to land where nothing
+    // else already sits.
+    //
+    // The merge joins on codes and not on the keys. The probe looks each key up
+    // in a HashMap under Scala's ==, where a NaN finds nothing, not even another
+    // NaN, while 0.0 meets -0.0 and a null meets a null; merge pairs NaN with
+    // NaN. `_texera_key_codes` from StandaloneHelpers.AttributeCasts gives each
+    // NaN a code of its own.
+    val merge =
+      s"""_left_cols = set(in1df.columns)
+         |_right_cols = list(in2df.columns)
+         |_payload = [_c for _c in _right_cols if _c != $probeKeyLit]
+         |_payload_set = set(_payload)
+         |_rename = {}
+         |for _col in _payload:
+         |    _new = _col
+         |    _others = _payload_set - {_col}
+         |    while _new in _left_cols or _new in _others:
+         |        _new = _new + "#@1"
+         |    _rename[_col] = _new
+         |_probe_key = $probeKeyLit
+         |_taken = _left_cols | set(_rename.values())
+         |while _probe_key in _taken:
+         |    _probe_key = _probe_key + "#@1"
+         |if _probe_key != $probeKeyLit:
+         |    _rename[$probeKeyLit] = _probe_key
+         |_join_key = "_texera_join_key"
+         |while _join_key in _taken or _join_key == _probe_key:
+         |    _join_key = _join_key + "_"
+         |_left_codes, _right_codes = _texera_key_codes(in1df[[$buildKeyLit]], in2df[[$probeKeyLit]])
+         |""".stripMargin + widening +
+        s"""out1df = $leftFrame.assign(**{_join_key: _left_codes}).merge(
+           |    $rightFrame.assign(**{_join_key: _right_codes}),
+           |    how=${pyStringLiteral(how)},
+           |    on=_join_key,
+           |).drop(columns=[_join_key])""".stripMargin
+    // `_probe_key` is whatever the rename settled on, so this drops the right
+    // column rather than a left one that happened to share its name. Asked of
+    // the name the rename produced and not of the two the operator was given:
+    // when both sides name the key alike the rename still moves the right one
+    // aside, and a test on the operator's own two names read that as one shared
+    // column and kept the copy, where the engine emits the key once.
+    //
+    // Where both sides name the key alike, the engine's unmatched right row
+    // fills the left key column with its own key. The merge left that column
+    // empty and put the key in the renamed right one, which is about to go.
+    // A matched row holds the same value in both, so filling the holes changes
+    // only the rows the right side alone brought.
+    val rightRowsKept = joinType == JoinType.RIGHT_OUTER || joinType == JoinType.FULL_OUTER
+    val fillKey =
+      if (rightRowsKept && buildAttributeName == probeAttributeName)
+        s"""out1df[$buildKeyLit] = out1df[$buildKeyLit].where(
+           |    out1df[$buildKeyLit].notna(), out1df[_probe_key]
+           |)
+           |""".stripMargin
+      else ""
+    val tail =
+      fillKey +
+        s"""if _probe_key != $buildKeyLit:
+           |    out1df = out1df.drop(columns=[_probe_key])
+           |out1df = out1df.reset_index(drop=True)""".stripMargin
+    s"$merge\n$tail"
+  }
 }