carloea2 commented on code in PR #8709:
URL: https://github.com/apache/texera/pull/8709#discussion_r4253803425
##########
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:
CONCAT still treats a stored NaN as null here. With nullable Arrow DOUBLE
values NaN, null and 1.0, AggregateOpExec returns NaN,,1.0 but the generated
scalar aggregate returns 1.0. I ran the actual generated block with the
declared #8597 helper; finite values with nulls agree. Please use the column's
validity mask rather than pd.isna on each scalar, and add this case alongside
the numeric reductions.
--
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]
