carloea2 commented on code in PR #8506:
URL: https://github.com/apache/texera/pull/8506#discussion_r4067015920
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpDesc.scala:
##########
@@ -69,4 +73,45 @@ 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.
+ //
+ // A string column parts more narrowly: the engine reads UTF-16 code units
and
+ // pandas reads code points, which agree below U+FFFF and can differ above
it.
+ 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.
+ s"""_texera_sorted = in1df.copy()
+ |_texera_by = []
+ |_texera_asc = []
+ |_texera_helpers = []
+ |for _texera_col, _texera_a in zip($cols, $ascending):
+ | _texera_null = "_texera_null_" + _texera_col
+ | _texera_nan = "_texera_nan_" + _texera_col
+ | _texera_sorted[_texera_null] = _texera_sorted[_texera_col].isna()
Review Comment:
Sorting by `x` deletes an existing `_texera_null_x` or `_texera_nan_x`
column. I tested `x=[2,1]` with payload values in each name: the helper
overwrites the payload, then the final drop removes it. The ordinary-name
control passes. Please allocate collision-free helpers or sort by separate
temporary keys so all input columns survive.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala:
##########
@@ -200,4 +200,95 @@ class HashJoinOpDesc[K] extends LogicalOp {
),
outputPorts = List(OutputPort())
)
+
+ // Equi-join: drop the probe key (kept only when its name differs from the
+ // build key), suffix colliding right columns "#@1" — matches JoinUtils.
Known
+ // Texera divergences: row order, null keys (NaN != NaN in merge), outer
+ // anti-row column placement.
+ /** Only the declared type can say which columns an outer join widened, so
+ * without a schema the widening stands.
+ */
+ override def generateStandaloneCode(inputSchemas: Map[PortIdentity,
Schema]): String = {
+ val integral = (a: Attribute) =>
+ a.getType == AttributeType.INTEGER || a.getType == AttributeType.LONG
+ val declaredIntegers = (port: PortIdentity) =>
+
inputSchemas.get(port).map(_.getAttributes.filter(integral).map(_.getName)).getOrElse(List())
+ // Each side keeps its own list, under the names that side's frame carries,
+ // so a column the merge renames is still the column its own schema typed.
+ if (joinType == JoinType.INNER) generateStandaloneCode()
+ else
+ generateStandaloneCode(
+ declaredIntegers(operatorInfo.inputPorts.head.id),
+ declaredIntegers(operatorInfo.inputPorts.last.id)
+ )
+ }
+
+ override def generateStandaloneCode(): String =
+ generateStandaloneCode(List(), List())
+
+ private def generateStandaloneCode(
+ leftIntegerColumns: List[String],
+ rightIntegerColumns: List[String]
+ ): String = {
+ val buildKeyLit = objectMapper.writeValueAsString(buildAttributeName)
+ val probeKeyLit = objectMapper.writeValueAsString(probeAttributeName)
+ val how = joinType match {
+ case JoinType.INNER => "inner"
+ case JoinType.LEFT_OUTER => "left"
+ case JoinType.RIGHT_OUTER => "right"
+ case JoinType.FULL_OUTER => "outer"
+ }
+ // An unmatched row leaves a hole, and a hole costs a pandas integer column
+ // its type: int64 becomes float64, which rounds every value past 2^53
+ // before anything can put the type back. The engine writes a null and
+ // leaves the column INTEGER, so widen to the integer dtype that holds a
+ // hole before the merge digs one.
+ val widen = leftIntegerColumns.nonEmpty || rightIntegerColumns.nonEmpty
+ val namesLit = (names: List[String]) =>
names.map(pyStringLiteral).mkString("[", ", ", "]")
+ val widening =
+ if (!widen) ""
+ else
+ s"""_left_ints = {_c: "Int64" for _c in
${namesLit(leftIntegerColumns)} if _c in in1df.columns}
+ |_right_ints = {_c: "Int64" for _c in
${namesLit(rightIntegerColumns)} if _c in in2df.columns}
+ |""".stripMargin
+ val leftFrame = if (widen) "in1df.astype(_left_ints)" else "in1df"
+ // Cast before the rename, because the names above are the ones the right
+ // input declared.
+ val rightFrame =
+ if (widen) "in2df.astype(_right_ints).rename(columns=_rename)"
+ else "in2df.rename(columns=_rename)"
+ // HashJoinProbeOpExec's rename, written out: append "#@1" until the name
is
+ // free. pandas' `suffixes` appends once and then refuses the duplicate it
+ // just made.
+ val merge =
+ s"""_left_cols = set(in1df.columns)
+ |_right_cols = list(in2df.columns)
+ |_right_set = set(_right_cols)
+ |_rename = {}
+ |for _col in _right_cols:
+ | _new = _col
+ | _others = _right_set - {_col}
Review Comment:
The probe key affects payload renaming even though it is dropped later.
Joining left `{k:1, x:'left'}` with right `{'x#@1':1, x:'right'}` on `k` and
`x#@1` emits `x#@1#@1`. Native schema propagation and JoinUtils exclude the
probe key first and name the payload `x#@1`, so a downstream projection fails.
Please exclude the discarded key from payload collision checks and add this
chained-join case.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala:
##########
@@ -138,4 +141,170 @@ 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)
+
+ 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.
+ | partial = ""
+ | for v in series:
+ | if pd.isna(v):
+ | text = ""
+ | elif isinstance(v, bool) or (hasattr(v, "dtype") and v.dtype
== bool):
+ | # Java's toString spells a boolean in lower case.
+ | text = "true" if v else "false"
+ | else:
+ | text = str(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_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.
+ | return series.dropna().astype("datetime64[ms]").astype("int64")
Review Comment:
This treats a naive wall-clock timestamp as UTC. With the JVM in
America/Mexico_City, summing midnight on 2024-01-01 and 2024-01-02 using the
engine's Timestamp arithmetic gives 2078-01-01 06:00:00; the generated helper
returns midnight. The average also differs by 21600000 milliseconds. UTC
passes. Please match the engine's timezone when converting to epoch
milliseconds and back, and test a non-UTC zone.
--
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]
