feat(workflow-operator): export Aggregate and the sort 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/sort/SortCriteriaUnit.scala

@@ -21,13 +21,17 @@ package org.apache.texera.amber.operator.sort
 
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString
-import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName
+import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn}
 
 class SortCriteriaUnit {
 
+  // Verification picks a column with TIE GROUPS. Sorting on a unique key never
+  // puts two equal keys next to each other, so the stability this operator is
+  // named for goes unchecked.
   @JsonProperty(value = "attribute", required = true)
   @JsonPropertyDescription("Attribute name to sort by")
   @AutofillAttributeName
+  @SampleColumn("name")
   var attributeName: EncodableString = _
 
   @JsonProperty(value = "sortPreference", required = true)

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

@@ -21,11 +21,18 @@ package org.apache.texera.amber.operator.sort
 
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import org.apache.texera.amber.core.tuple.Schema
-import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{
+  PythonTemplateBuilderStringContext,
+  pyStringLiteral
+}
 import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
-import org.apache.texera.amber.operator.PythonOperatorDescriptor
+import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator}
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
-class SortOpDesc extends PythonOperatorDescriptor {
+class SortOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerator {
+
+  // Sorting is this operator's contract: output row order is meaningful.
+  override def orderSensitive: Boolean = true
+
   @JsonProperty(required = true)
   @JsonPropertyDescription("column to perform sorting on")
   var attributes: List[SortCriteriaUnit] = List.empty
@@ -78,4 +85,11 @@ class SortOpDesc extends PythonOperatorDescriptor {
       outputPorts = List(OutputPort(blocking = true))
     )
 
+  override def generateStandaloneCode(): String = {
+    val cols = attributes.map(c => pyStringLiteral(c.attributeName)).mkString("[", ", ", "]")
+    val ascending = attributes
+      .map(c => if (c.sortPreference == SortPreference.ASC) "True" else "False")
+      .mkString("[", ", ", "]")
+    s"out1df = in1df.sort_values(by=$cols, ascending=$ascending)"
+  }
 }

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

@@ -24,8 +24,9 @@ import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp}
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
 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 scala.collection.mutable.ListBuffer
@@ -36,7 +37,10 @@ import scala.collection.mutable.ListBuffer
   * the lexicographic order and per-key direction (ASC/DESC).
   */
 //TODO(#3922): disallowing sorting on binary type
-class StableMergeSortOpDesc extends LogicalOp {
+class StableMergeSortOpDesc extends LogicalOp with StandaloneCodeGenerator {
+
+  // Sorting is this operator's contract: output row order is meaningful.
+  override def orderSensitive: Boolean = true
 
   @JsonProperty(value = "keys", required = true)
   @JsonSchemaTitle("Sort Keys")
@@ -69,4 +73,69 @@ class StableMergeSortOpDesc extends LogicalOp {
       List(InputPort()),
       List(OutputPort(blocking = true))
     )
+
+  // The engine runs an incremental stable merge sort with nulls last whichever
+  // way a key points. pandas' mergesort is stable too, so the ordering below is
+  // the same one.
+  override def generateStandaloneCode(): String = {
+    val criteria = Option(keys).getOrElse(ListBuffer.empty)
+    if (criteria.isEmpty) return "out1df = in1df.copy()"
+    val cols = criteria
+      .map(c => pyStringLiteral(c.attributeName))
+      .mkString("[", ", ", "]")
+    val ascending = criteria
+      .map(c => if (c.sortPreference == SortPreference.ASC) "True" else "False")
+      .mkString("[", ", ", "]")
+    // Sort each key in three tiers, because the engine treats a null and a NaN
+    // differently: a null goes last whichever way the key points, while a NaN
+    // compares above every number, so it goes last ascending and first
+    // descending. A column read into a numpy dtype has one slot for both, and
+    // there both land in the null tier, which is where they were before.
+    //
+    // The tiers are built in a frame of their own and named by position, so
+    // the input keeps every column it arrived with: a helper named after the
+    // key would overwrite an input column that already answers to that name,
+    // and dropping the helper afterwards would take the payload with it.
+    //
+    // A string sorts by its UTF-16 big-endian bytes. The engine compares UTF-16
+    // code units and Python compares code points, and the two disagree once a
+    // character above U+FFFF meets one between U+E000 and U+FFFF.
+    //
+    // A double's sign is a fourth tier. Double.compare puts -0.0 below 0.0,
+    // where pandas reads the two as equal and keeps them in input order.
+    s"""import math
+       |_texera_sorted = in1df.reset_index(drop=True)
+       |_texera_keys = pd.DataFrame(index=_texera_sorted.index)
+       |_texera_by = []
+       |_texera_asc = []
+       |for _texera_i, (_texera_col, _texera_a) in enumerate(zip($cols, $ascending)):
+       |    _texera_null = "null_" + str(_texera_i)
+       |    _texera_nan = "nan_" + str(_texera_i)
+       |    _texera_val = "val_" + str(_texera_i)
+       |    _texera_keys[_texera_null] = _texera_sorted[_texera_col].isna()
+       |    _texera_keys[_texera_nan] = (
+       |        _texera_sorted[_texera_col] != _texera_sorted[_texera_col]
+       |    ).fillna(False)
+       |    _texera_v = _texera_sorted[_texera_col]
+       |    if _texera_v.dtype == object or isinstance(_texera_v.dtype, pd.StringDtype):
+       |        _texera_v = _texera_v.map(
+       |            lambda v: v.encode("utf-16-be") if isinstance(v, str) else v
+       |        )
+       |    _texera_keys[_texera_val] = _texera_v
+       |    _texera_by += [_texera_null, _texera_nan, _texera_val]
+       |    _texera_asc += [True, _texera_a, _texera_a]
+       |    if pd.api.types.is_float_dtype(_texera_v.dtype):
+       |        _texera_sign = "sign_" + str(_texera_i)
+       |        _texera_keys[_texera_sign] = [
+       |            1 if missing or v != v or math.copysign(1.0, v) > 0 else 0
+       |            for v, missing in zip(_texera_v, _texera_v.isna())
+       |        ]
+       |        _texera_by.append(_texera_sign)
+       |        _texera_asc.append(_texera_a)
+       |out1df = _texera_sorted.loc[
+       |    _texera_keys.sort_values(
+       |        by=_texera_by, ascending=_texera_asc, kind="mergesort"
+       |    ).index
+       |].reset_index(drop=True)""".stripMargin
+  }
 }

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

@@ -24,9 +24,10 @@ import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchema
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, RangePartition}
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
 import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName
 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
 
 @JsonSchemaInject(json = """
@@ -38,7 +39,10 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
   }
 }
 """)
-class SortPartitionsOpDesc extends LogicalOp {
+class SortPartitionsOpDesc extends LogicalOp with StandaloneCodeGenerator {
+
+  // Sorting is this operator's contract: output row order is meaningful.
+  override def orderSensitive: Boolean = true
 
   @JsonProperty(required = true)
   @JsonSchemaTitle("Attribute")
@@ -84,4 +88,48 @@ class SortPartitionsOpDesc extends LogicalOp {
       inputPorts = List(InputPort()),
       outputPorts = List(OutputPort(blocking = true))
     )
+
+  // Row order is this operator's answer, so the sort matches the engine's:
+  // stable, ascending, nulls first. The domain bounds are partitioning hints
+  // that the sort itself never reads.
+  //
+  // The engine orders a NaN after positive infinity. In a nullable column,
+  // which is what an Arrow file is read into, pandas keeps a NaN apart from a
+  // null and sorts it last too. A numpy float column has one slot for both, so
+  // there a NaN goes first with the nulls.
+  //
+  // The rest of AttributeTypeUtils.compare is written out as tiers of a key
+  // frame, as in StableMergeSortOpDesc. Double.compare puts -0.0 below 0.0,
+  // where pandas reads the two as equal. A timestamp is compared by getTime,
+  // so two inside one millisecond keep their input order. A string compares
+  // UTF-16 code units, where pandas compares code points, and the two disagree
+  // once a character above U+FFFF meets one between U+E000 and U+FFFF.
+  override def generateStandaloneCode(): String = {
+    val col = pyStringLiteral(Option(sortAttributeName).getOrElse(""))
+    s"""import math
+       |out1df = in1df.reset_index(drop=True)
+       |_texera_v = out1df[$col]
+       |_texera_keys = pd.DataFrame(index=out1df.index)
+       |_texera_keys["held"] = _texera_v.notna()
+       |_texera_keys["nan"] = (_texera_v != _texera_v).fillna(False)
+       |if pd.api.types.is_datetime64_any_dtype(_texera_v.dtype):
+       |    _texera_keys["val"] = _texera_v.dt.floor("ms")
+       |elif _texera_v.dtype == object or isinstance(_texera_v.dtype, pd.StringDtype):
+       |    _texera_keys["val"] = _texera_v.map(
+       |        lambda v: v.encode("utf-16-be") if isinstance(v, str) else v
+       |    )
+       |else:
+       |    _texera_keys["val"] = _texera_v
+       |_texera_by = ["held", "nan", "val"]
+       |if pd.api.types.is_float_dtype(_texera_v.dtype):
+       |    _texera_keys["sign"] = [
+       |        1 if missing or v != v or math.copysign(1.0, v) > 0 else 0
+       |        for v, missing in zip(_texera_v, _texera_v.isna())
+       |    ]
+       |    _texera_by.append("sign")
+       |out1df = out1df.loc[
+       |    _texera_keys.sort_values(by=_texera_by, kind="mergesort").index
+       |].reset_index(drop=True)""".stripMargin
+  }
+
 }