kz930 commented on code in PR #8709:
URL: https://github.com/apache/texera/pull/8709#discussion_r4254093565
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala:
##########
@@ -138,4 +141,278 @@ 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)
+
+ override def standaloneHelpers(): Seq[String] =
Seq(StandaloneHelpers.AttributeCasts)
+
+ 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.
+ | # Each value is written as Java's toString writes it.
+ | partial = ""
+ | for v in series:
+ | text = "" if pd.isna(v) else _texera_java_text(v)
Review Comment:
Fixed: the CONCAT fold now reads nulls from the column's mask, so a stored
NaN is written as NaN. The case is added to the NaN parity test for both
grouped and ungrouped input, and it fails on the old code with 1.0,,3.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
