feat(workflow-operator): export the relational and sampling operators

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/SamplingHelpers.scala

@@ -0,0 +1,68 @@
+/*
+ * 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
+
+/** The Python a sampler's standalone code needs, emitted once per script via
+  * [[StandaloneCodeGenerator.standaloneHelpers]].
+  */
+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.
+    */
+  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 <= 0:
+      |            raise ValueError("bound must be positive")
+      |        if bound & (-bound) == bound:
+      |            return (bound * self._next(31)) >> 31
+      |        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/aggregate/AggregateOpDesc.scala

@@ -22,21 +22,24 @@ package org.apache.texera.amber.operator.aggregate
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import org.apache.texera.amber.core.executor.OpExecWithClassName
-import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.tuple.{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}
 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}
+import scala.util.Try
+
+class AggregateOpDesc extends LogicalOp with StandaloneCodeGenerator {
 
-class AggregateOpDesc extends LogicalOp {
   @JsonProperty(value = "aggregations", required = true)
   @JsonPropertyDescription("multiple aggregation functions")
   @NotNull(message = "aggregation cannot be null")
@@ -138,4 +141,198 @@ 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.
+    *
+    * SUM and AVERAGE follow the column's DECLARED type: a holed INTEGER column
+    * arrives as a float, and an INTEGER and a LONG arrive alike.
+    */
+  override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = {
+    val schema = inputSchemas.get(operatorInfo.inputPorts.head.id)
+    build(name => schema.flatMap(s => Try(s.getAttribute(name).getType).toOption))
+  }
+
+  override def generateStandaloneCode(): String = build(_ => None)
+
+  private def build(declaredType: String => Option[AttributeType]): 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):
+        |    # The accumulator starts empty and only earns a separator once it
+        |    # holds something, so a leading empty value adds neither text nor
+        |    # comma: "", "a", "" concatenates to "a," and not ",a,". A null is
+        |    # read as the empty string, which is what makes the two the same
+        |    # here. This is concatAgg's fold, written out.
+        |    partial = ""
+        |    for v in series:
+        |        if pd.isna(v):
+        |            text = ""
+        |        elif isinstance(v, bool) or (hasattr(v, "dtype") and v.dtype == bool):
+        |            # Java's toString spells a boolean in lower case.
+        |            text = "true" if v else "false"
+        |        else:
+        |            text = str(v)
+        |        partial = text if partial == "" else partial + "," + text
+        |    return partial
+        |
+        |def _texera_agg_int_sum(series):
+        |    # The engine adds an INTEGER column as Java ints, which wrap.
+        |    total = int(series.sum())
+        |    return ((total + (1 << 31)) % (1 << 32)) - (1 << 31)
+        |
+        |def _texera_agg_ts_zone():
+        |    # The engine reads and writes a timestamp in the JVM's default zone,
+        |    # so the arithmetic below has to name the same one. gettz() and not
+        |    # the current offset: the zone carries its daylight rules, and each
+        |    # instant needs the offset in force when it happened.
+        |    from dateutil.tz import gettz
+        |
+        |    return gettz()
+        |
+        |def _texera_agg_ts_epoch_ms(series):
+        |    # A timestamp reaches the engine as its epoch milliseconds whatever
+        |    # resolution the column carries, and reading the integers out of a
+        |    # microsecond column asks for a different number than a nanosecond
+        |    # one, so cast to milliseconds before reading them.
+        |    #
+        |    # A column holds a wall clock, and `Timestamp.getTime` answers for
+        |    # the instant that wall clock names locally, so localize before
+        |    # reading the integers out: left as UTC every one of them is a whole
+        |    # offset away. The two flags are java.time's own reading of an hour
+        |    # daylight saving repeats (the later one) or skips (shifted past the
+        |    # gap).
+        |    return (
+        |        series.dropna()
+        |        .astype("datetime64[ms]")
+        |        .dt.tz_localize(
+        |            _texera_agg_ts_zone(), ambiguous=False, nonexistent=pd.Timedelta("1h")
+        |        )
+        |        .astype("int64")
+        |    )
+        |
+        |def _texera_agg_ts_sum(series):
+        |    # SUM keeps the column's own type, so the engine adds the epoch
+        |    # milliseconds as Java longs, which wrap, and builds a timestamp
+        |    # from the total. Python integers carry the sum exactly, so the
+        |    # wrap is the only place a total loses anything.
+        |    total = int(_texera_agg_ts_epoch_ms(series).astype(object).sum())
+        |    total = ((total + (1 << 63)) % (1 << 64)) - (1 << 63)
+        |    return (
+        |        pd.Timestamp(total, unit="ms", tz="UTC")
+        |        .tz_convert(_texera_agg_ts_zone())
+        |        .tz_localize(None)
+        |    )
+        |
+        |def _texera_agg_ts_mean(series):
+        |    # AVERAGE is declared DOUBLE whatever column it reads, so this is
+        |    # the mean of the epoch milliseconds and not a timestamp.
+        |    kept = _texera_agg_ts_epoch_ms(series)
+        |    if len(kept) == 0:
+        |        return None
+        |    return float(kept.astype(object).sum()) / len(kept)""".stripMargin
+
+    if (keys.isEmpty) {
+      val rowEntries = aggs
+        .map(agg =>
+          s"    ${pyStringLiteral(agg.resultAttribute)}: ${aggExprScalar(agg, declaredType)},"
+        )
+        .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", declaredType)}"
+        }
+        .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,
+      declaredType: String => Option[AttributeType]
+  ): String = {
+    val attrLit =
+      if (agg.attribute == null || agg.attribute.isEmpty) "None"
+      else pyStringLiteral(agg.attribute)
+    val declared = Option(agg.attribute).filter(_.nonEmpty).flatMap(declaredType)
+    agg.aggFunction match {
+      case AggregationFunction.SUM =>
+        declared match {
+          case Some(AttributeType.INTEGER)   => s"_texera_agg_int_sum(in1df[$attrLit])"
+          case Some(AttributeType.TIMESTAMP) => s"_texera_agg_ts_sum(in1df[$attrLit])"
+          case _                             => s"in1df[$attrLit].sum()"
+        }
+      case AggregationFunction.AVERAGE =>
+        declared match {
+          case Some(AttributeType.TIMESTAMP) => s"_texera_agg_ts_mean(in1df[$attrLit])"
+          case _                             => 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,
+      declaredType: String => Option[AttributeType]
+  ): String = {
+    val attrLit =
+      if (agg.attribute == null || agg.attribute.isEmpty) "None"
+      else pyStringLiteral(agg.attribute)
+    val resultLit = pyStringLiteral(agg.resultAttribute)
+    val declared = Option(agg.attribute).filter(_.nonEmpty).flatMap(declaredType)
+    agg.aggFunction match {
+      case AggregationFunction.SUM =>
+        declared match {
+          case Some(AttributeType.INTEGER) =>
+            s"$groups[$attrLit].apply(_texera_agg_int_sum).rename($resultLit)"
+          case Some(AttributeType.TIMESTAMP) =>
+            s"$groups[$attrLit].apply(_texera_agg_ts_sum).rename($resultLit)"
+          case _ => s"$groups[$attrLit].sum().rename($resultLit)"
+        }
+      case AggregationFunction.AVERAGE =>
+        declared match {
+          case Some(AttributeType.TIMESTAMP) =>
+            s"$groups[$attrLit].apply(_texera_agg_ts_mean).rename($resultLit)"
+          case _ => 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
+  }
 }

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}
 
-class DifferenceOpDesc extends LogicalOp {
-
+class DifferenceOpDesc extends LogicalOp with StandaloneCodeGenerator {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -61,4 +60,12 @@ class DifferenceOpDesc extends LogicalOp {
       ),
       outputPorts = List(OutputPort(blocking = true))
     )
+
+  // Distinct rows in left (port 0) not in right (port 1). [left, right, right] +
+  // drop_duplicates(keep=False): right rows appear >=2x and drop, left-only
+  // survive. concat (not merge) so NaN == NaN matches the JVM HashSet.
+  override def generateStandaloneCode(): String =
+    "out1df = pd.concat([in1df.drop_duplicates(), in2df.drop_duplicates(), " +
+      "in2df.drop_duplicates()], ignore_index=True)" +
+      ".drop_duplicates(keep=False).reset_index(drop=True)"
 }