feat(workflow-operator): export the base transform operators as Python

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

@@ -459,6 +459,18 @@ abstract class LogicalOp extends PortDescriptor with Serializable {
 
   def operatorInfo: OperatorInfo
 
+  /**
+    * Whether the row ORDER of this operator's output is part of its contract.
+    * Defaults to false: the engine runs operators across parallel workers, so
+    * for almost every operator the output row order is an implementation-defined
+    * interleaving. Only operators whose very purpose is to establish an order
+    * override this, which here is the sort family: Sort, Stable Merge Sort and
+    * Sort Partitions. Anything comparing two runs of an operator reads it to
+    * decide whether the rows have to arrive in the same order or only be the
+    * same rows.
+    */
+  def orderSensitive: Boolean = false
+
   private def getOperatorVersion: String = {
     val path = "amber/src/main/scala/"
     val operatorPath = path + this.getClass.getPackage.getName.replace(".", "/")

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

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator
+
+/** Python definitions shared by several operators' standalone code, emitted
+  * once per script via [[StandaloneCodeGenerator.standaloneHelpers]].
+  */
+object StandaloneHelpers {
+
+  /**
+    * 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.
+    */
+  val JavaRandom: String =
+    """# java.util.Random, transcribed so sampling matches the engine.
+      |class _TexeraJavaRandom:
+      |    _MASK = (1 << 48) - 1
+      |    _MULTIPLIER = 0x5DEECE66D
+      |    _ADDEND = 0xB
+      |
+      |    def __init__(self, seed):
+      |        self._seed = (seed ^ self._MULTIPLIER) & self._MASK
+      |
+      |    def _next(self, bits):
+      |        self._seed = (self._seed * self._MULTIPLIER + self._ADDEND) & self._MASK
+      |        value = self._seed >> (48 - bits)
+      |        return value - (1 << 32) if value >= (1 << 31) else value
+      |
+      |    def next_double(self):
+      |        return ((self._next(26) << 27) + self._next(27)) * (2.0 ** -53)
+      |
+      |    def next_int(self, bound):
+      |        if bound & (-bound) == bound:
+      |            return (bound * self._next(31)) >> 31
+      |        while True:
+      |            bits = self._next(31)
+      |            value = bits % bound
+      |            if bits - value + (bound - 1) >= 0:
+      |                return value""".stripMargin
+}

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

@@ -29,14 +29,16 @@ import org.apache.texera.amber.core.virtualidentity.{
   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.annotations.AutofillAttributeNameList
 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
 
 import javax.validation.constraints.{NotNull, Size}
 
-class AggregateOpDesc extends LogicalOp {
+class AggregateOpDesc extends LogicalOp with StandaloneCodeGenerator {
+
   @JsonProperty(value = "aggregations", required = true)
   @JsonPropertyDescription("multiple aggregation functions")
   @NotNull(message = "aggregation cannot be null")
@@ -138,4 +140,93 @@ class AggregateOpDesc extends LogicalOp {
       inputPorts = List(InputPort()),
       outputPorts = List(OutputPort())
     )
+
+  // The engine aggregates in two phases across partitions; one process needs
+  // only the one groupby, or a single-row reduction when no key is grouped on.
+  //
+  // Must run before `getPhysicalPlan`, which rewrites `aggregations` in place:
+  // it turns COUNT into SUM for the final phase, and this reads them as written.
+  override def generateStandaloneCode(): String = {
+    val keys = Option(groupByKeys).getOrElse(List())
+    val aggs = Option(aggregations).getOrElse(List())
+
+    // Identical helper definition each call — keeps the standalone module
+    // self-contained without relying on a shared prelude.
+    val concatHelper =
+      """def _texera_agg_concat(series):
+        |    parts = []
+        |    started = False
+        |    for v in series:
+        |        if not started:
+        |            if pd.isna(v):
+        |                continue
+        |            parts.append(str(v))
+        |            started = True
+        |        else:
+        |            parts.append("" if pd.isna(v) else str(v))
+        |    return ",".join(parts)""".stripMargin
+
+    if (keys.isEmpty) {
+      val rowEntries = aggs
+        .map(agg => s"    ${pyStringLiteral(agg.resultAttribute)}: ${aggExprScalar(agg)},")
+        .mkString("\n")
+      s"""$concatHelper
+         |out1df = pd.DataFrame([{
+         |$rowEntries
+         |}])""".stripMargin
+    } else {
+      val keysLit = keys.map(pyStringLiteral).mkString("[", ", ", "]")
+      val aggLines = aggs.zipWithIndex
+        .map {
+          case (agg, i) =>
+            s"_texera_agg_s$i = ${aggExprGroupby(agg, "_texera_agg_groups")}"
+        }
+        .mkString("\n")
+      val mergeLines = aggs.indices
+        .map(i =>
+          s"""out1df = out1df.merge(_texera_agg_s$i.reset_index(), on=$keysLit, how="left")"""
+        )
+        .mkString("\n")
+      s"""$concatHelper
+         |_texera_agg_groups = in1df.groupby($keysLit, dropna=False, sort=False)
+         |out1df = in1df[$keysLit].drop_duplicates().reset_index(drop=True)
+         |$aggLines
+         |$mergeLines""".stripMargin
+    }
+  }
+
+  private def aggExprScalar(agg: AggregationOperation): String = {
+    val attrLit =
+      if (agg.attribute == null || agg.attribute.isEmpty) "None"
+      else pyStringLiteral(agg.attribute)
+    agg.aggFunction match {
+      case AggregationFunction.SUM     => s"in1df[$attrLit].sum()"
+      case AggregationFunction.AVERAGE => s"in1df[$attrLit].mean()"
+      case AggregationFunction.MIN     => s"in1df[$attrLit].min()"
+      case AggregationFunction.MAX     => s"in1df[$attrLit].max()"
+      case AggregationFunction.COUNT =>
+        if (agg.attribute == null || agg.attribute.isEmpty) "int(len(in1df))"
+        else s"int(in1df[$attrLit].count())"
+      case AggregationFunction.CONCAT => s"_texera_agg_concat(in1df[$attrLit])"
+    }
+  }
+
+  private def aggExprGroupby(agg: AggregationOperation, groups: String): String = {
+    val attrLit =
+      if (agg.attribute == null || agg.attribute.isEmpty) "None"
+      else pyStringLiteral(agg.attribute)
+    val resultLit = pyStringLiteral(agg.resultAttribute)
+    agg.aggFunction match {
+      case AggregationFunction.SUM     => s"$groups[$attrLit].sum().rename($resultLit)"
+      case AggregationFunction.AVERAGE => s"$groups[$attrLit].mean().rename($resultLit)"
+      case AggregationFunction.MIN     => s"$groups[$attrLit].min().rename($resultLit)"
+      case AggregationFunction.MAX     => s"$groups[$attrLit].max().rename($resultLit)"
+      case AggregationFunction.COUNT =>
+        if (agg.attribute == null || agg.attribute.isEmpty)
+          s"$groups.size().rename($resultLit)"
+        else s"$groups[$attrLit].count().rename($resultLit)"
+      case AggregationFunction.CONCAT =>
+        s"$groups[$attrLit].apply(_texera_agg_concat).rename($resultLit)"
+    }
+  }
 }

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
+  }
 }