carloea2 commented on code in PR #8709:
URL: https://github.com/apache/texera/pull/8709#discussion_r4252110113
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala:
##########
@@ -138,4 +141,202 @@ 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)
+ | 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")
+ val resultsLit =
+ aggs.map(agg => pyStringLiteral(agg.resultAttribute)).mkString("[", ",
", "]")
+ // The engine opens a group when a row arrives, so with no rows there is
no
+ // group to finish and it emits nothing, not one row of COUNT 0.
+ s"""$concatHelper
+ |if len(in1df):
+ | out1df = pd.DataFrame([{
+ |${rowEntries.linesIterator.map(" " + _).mkString("\n")}
+ | }])
+ |else:
+ | out1df = pd.DataFrame(columns=$resultsLit)""".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()"
Review Comment:
With a nullable DOUBLE column containing 1, a stored NaN and 3, the engine
returns NaN for SUM and AVERAGE. These pandas reductions return pd.NA instead,
turning NaN into null. Reproduced against AggregateOpExec with pandas 2.2.3;
numeric and null-only controls pass. Please preserve the engine result and add
the parity test.
--
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]
