kz930 commented on code in PR #8340:
URL: https://github.com/apache/texera/pull/8340#discussion_r3983302000
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala:
##########
@@ -138,4 +140,93 @@ 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.
+ override def generateStandaloneCode(): 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):
+ | parts = []
+ | started = False
+ | for v in series:
+ | if not started:
+ | if pd.isna(v):
+ | continue
+ | parts.append(str(v))
+ | started = True
Review Comment:
Fixed: the accumulator earns its separator only once it holds something, so
`"", "a", ""` is `a,` now. A leading empty string is the only value that tells
the two spellings apart, so the fixture gained one.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala:
##########
@@ -72,4 +74,51 @@ class TypeCastingOpDesc extends MapOpDesc {
List(OutputPort())
)
}
+
+ override def generateStandaloneCode(): String = {
+ val units = Option(typeCastingUnits).getOrElse(List.empty)
+ if (units.isEmpty) return "out1df = in1df.copy()"
+
+ val lines = scala.collection.mutable.ArrayBuffer[String]("out1df =
in1df.copy()")
+ units.foreach { unit =>
+ val colLit = pyStringLiteral(unit.attribute)
+ // Every cast goes through the transcription of AttributeTypeUtils rather
+ // than through Python's own conversions, which answer differently: a
+ // non-empty string is always a true boolean, and a coercing numeric cast
+ // reads "6.7" as an integer the engine refuses.
+ //
+ // A timestamp is the one that stays approximate. The engine reads it
with
+ // DateParserUtils, which accepts a set of formats no single pandas call
+ // states, so this coerces what it cannot read rather than claiming a
+ // match it does not have.
+ val expr = unit.resultType match {
+ case AttributeType.STRING =>
+ // `astype(str)` gets two things wrong against `toString`: an empty
+ // cell renders as the text "nan", and a boolean capitalises. The
+ // helper handles both, and reads the point of a double off the
+ // column's type rather than off the value, which cannot tell a whole
+ // double from an integer.
+ s"""_texera_cast_string(out1df[$colLit])"""
+ case AttributeType.INTEGER | AttributeType.LONG =>
+ // A hole survives the cast, because parseField returns a null field
+ // untouched; pandas' nullable "Int64" holds one where numpy's int
+ // cannot.
+ s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else
_texera_cast_integral(x)).astype("Int64")"""
+ case AttributeType.DOUBLE =>
+ // NaN rather than pd.NA: float64 is how a double column is held
here,
+ // and it carries its hole as NaN. pd.NA would not survive the
astype.
+ s"""out1df[$colLit].apply(lambda x: float("nan") if pd.isna(x) else
_texera_cast_double(x)).astype("float64")"""
+ case AttributeType.BOOLEAN =>
+ // Nullable "boolean" for the same reason, and because
`.astype(bool)`
+ // reads NaN as True: NaN is a non-zero float.
+ s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else
_texera_cast_boolean(x)).astype("boolean")"""
+ case AttributeType.TIMESTAMP => s"""pd.to_datetime(out1df[$colLit],
errors="coerce")"""
Review Comment:
Fixed: a LONG source is read as milliseconds and converted to the local
zone, since `Timestamp.toString` renders there rather than in UTC.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala:
##########
@@ -72,4 +74,51 @@ class TypeCastingOpDesc extends MapOpDesc {
List(OutputPort())
)
}
+
+ override def generateStandaloneCode(): String = {
+ val units = Option(typeCastingUnits).getOrElse(List.empty)
+ if (units.isEmpty) return "out1df = in1df.copy()"
+
+ val lines = scala.collection.mutable.ArrayBuffer[String]("out1df =
in1df.copy()")
+ units.foreach { unit =>
+ val colLit = pyStringLiteral(unit.attribute)
+ // Every cast goes through the transcription of AttributeTypeUtils rather
+ // than through Python's own conversions, which answer differently: a
+ // non-empty string is always a true boolean, and a coercing numeric cast
+ // reads "6.7" as an integer the engine refuses.
+ //
+ // A timestamp is the one that stays approximate. The engine reads it
with
+ // DateParserUtils, which accepts a set of formats no single pandas call
+ // states, so this coerces what it cannot read rather than claiming a
+ // match it does not have.
+ val expr = unit.resultType match {
+ case AttributeType.STRING =>
+ // `astype(str)` gets two things wrong against `toString`: an empty
+ // cell renders as the text "nan", and a boolean capitalises. The
+ // helper handles both, and reads the point of a double off the
+ // column's type rather than off the value, which cannot tell a whole
+ // double from an integer.
+ s"""_texera_cast_string(out1df[$colLit])"""
+ case AttributeType.INTEGER | AttributeType.LONG =>
+ // A hole survives the cast, because parseField returns a null field
+ // untouched; pandas' nullable "Int64" holds one where numpy's int
+ // cannot.
+ s"""out1df[$colLit].apply(lambda x: pd.NA if pd.isna(x) else
_texera_cast_integral(x)).astype("Int64")"""
Review Comment:
Fixed: INTEGER wraps the low 32 bits for an integral source and saturates
for a double, as `Long.toInt` and `Double.toInt` do. The source type is passed
in, since `Series.apply` hands a nullable integer column over as floats.
--
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]
