kz930 commented on code in PR #8980:
URL: https://github.com/apache/texera/pull/8980#discussion_r4291517937
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala:
##########
@@ -63,42 +64,166 @@ class SpecializedFilterOpDesc extends FilterOpDesc with
StandaloneCodeGenerator
)
}
- override def generateStandaloneCode(): String = {
+ override def generateStandaloneCode(): String =
generateStandaloneCode(Map.empty)
+
+ override def generateStandaloneCode(inputSchemas: Map[PortIdentity,
Schema]): String = {
// No predicate keeps no row: the executor's filter is `predicates.exists`,
// which answers false on an empty list. Passing the frame through would be
// the opposite answer.
if (predicates.isEmpty) return "out1df = in1df.iloc[0:0].copy()"
+ val declared: Map[String, AttributeType] = inputSchemas.values.headOption
+ .map(_.getAttributes.map(a => a.getName -> a.getType).toMap)
+ .getOrElse(Map.empty)
val conditions = predicates.map { p =>
val colLit = pyStringLiteral(p.attribute)
p.condition match {
case ComparisonType.IS_NULL => s"""(in1df[$colLit].isna())"""
case ComparisonType.IS_NOT_NULL => s"""(in1df[$colLit].notna())"""
- case other =>
- val op = other.getName // returns "=", ">=", "<", etc. (see
ComparisonType.java)
- val pyOp = if (op == "=") "==" else op
- // notna mirrors FilterPredicate, which answers false for every
condition
- // but IS_NULL / IS_NOT_NULL once the field is null. Only `!=` needs
it —
- // pandas answers True there, where every other operator answers
False —
- // but guarding all of them keeps the one rule visible in one place.
- s"""(in1df[$colLit].notna() & (in1df[$colLit] $pyOp
${coerceValue(p.value)}))"""
+ case other =>
+ // Without a schema the column's dtype stands in for its type.
+ val kind = declared.get(p.attribute).map(t =>
pyStringLiteral(t.name)).getOrElse("None")
+ val opLit = pyStringLiteral(other.getName)
+ val valueLit = pyStringLiteral(Option(p.value).getOrElse(""))
+ s"_texera_filter_keep(in1df[$colLit], $opLit, $valueLit, $kind)"
}
}
s"out1df = in1df[${conditions.mkString(" | ")}].reset_index(drop=True)"
}
- // Try numeric coercion so generated code compares column values against the
right type.
- // Strings that don't parse fall through to a quoted string literal.
- private def coerceValue(raw: String): String = {
- try {
- raw.toInt.toString
- } catch {
- case _: NumberFormatException =>
- try {
- raw.toDouble.toString
- } catch {
- case _: NumberFormatException =>
- pyStringLiteral(raw)
- }
- }
- }
+ override def standaloneHelpers(): Seq[String] =
Seq(SpecializedFilterOpDesc.FilterPredicates)
+}
+
+object SpecializedFilterOpDesc {
+
+ /** `FilterPredicate.evaluate`, transcribed. pandas' operators answer
+ * differently on every type it handles apart: a STRING field and the value
+ * are compared as numbers when both parse as one, a BOOLEAN as its
+ * lower-case text, an INTEGER as a double, and a DOUBLE with
+ * `Double.compareTo`, under which NaN equals itself and sits above every
+ * number, and -0.0 sits below 0.0. A null field passes no comparison.
+ */
+ val FilterPredicates: String =
+ """def _texera_filter_keep(column, condition, value, kind):
+ | import math
+ | import re
+ |
+ | # String.trim drops every character up to U+0020, not only
whitespace.
+ | def java_trim(text):
+ | return text.strip("".join(chr(i) for i in range(33)))
+ |
+ | # Double.parseDouble: a trailing d or f is a type suffix and
0x...p...
+ | # is hex, where float() refuses both, and "nan", "inf" and "1_0",
+ | # which float() reads, are refused.
+ | java_double_text = re.compile(
+ | r"[+-]?(NaN|Infinity|"
+ | r"((\d+\.?\d*|\.\d+)([eE][+-]?\d+)?"
+ |
r"|0[xX]([0-9a-fA-F]+\.?|[0-9a-fA-F]*\.[0-9a-fA-F]+)[pP][+-]?\d+)[fFdD]?)",
+ | re.ASCII,
+ | )
+ |
+ | def java_double(text):
+ | text = java_trim(text)
+ | if not java_double_text.fullmatch(text):
+ | raise ValueError("For input string: " + repr(text))
+ | body = text.lstrip("+-")
+ | if body == "NaN":
+ | return float("nan")
+ | if body == "Infinity":
+ | return float("-inf") if text.startswith("-") else
float("inf")
+ | if text[-1] in "dDfF":
+ | text = text[:-1]
+ | if body[:2] in ("0x", "0X"):
+ | return float.fromhex(text)
+ | return float(text)
+ |
+ | # Double.compareTo: NaN equals itself and sits above every number,
and
+ | # -0.0 sits below 0.0.
+ | def double_key(x):
+ | if math.isnan(x):
+ | return (1, 0.0, 0.0)
+ | return (0, x, math.copysign(1.0, x))
+ |
+ | # String.compareTo reads UTF-16 code units, which the big-endian
bytes
+ | # order the same way.
+ | def text_key(s):
+ | return s.encode("utf-16-be", "surrogatepass")
+ |
+ | def compared(left, right):
+ | return (left > right) - (left < right)
+ |
+ | if kind is None:
+ | dtype = column.dtype
+ | if pd.api.types.is_bool_dtype(dtype):
+ | kind = "BOOLEAN"
+ | elif pd.api.types.is_integer_dtype(dtype):
+ | kind = "INTEGER"
+ | elif pd.api.types.is_float_dtype(dtype):
+ | kind = "DOUBLE"
+ | elif pd.api.types.is_datetime64_any_dtype(dtype):
+ | kind = "TIMESTAMP"
+ | else:
+ | kind = "STRING"
+ |
+ | if kind in ("DOUBLE", "INTEGER"):
+ | target = double_key(java_double(value))
+ |
+ | def order(v):
+ | return compared(double_key(float(v)), target)
+ |
+ | elif kind == "LONG":
+ | # Long.valueOf takes a sign and digits, and nothing else.
+ | text = java_trim(value)
+ | if not re.fullmatch(r"[+-]?\d+", text) or not -(2**63) <=
int(text) < 2**63:
+ | raise ValueError("For input string: " + repr(text))
+ | target = int(text)
+ |
+ | def order(v):
+ | return compared(int(v), target)
+ |
+ | elif kind == "BOOLEAN":
+ | target = text_key(java_trim(value).lower())
+ |
+ | def order(v):
+ | return compared(text_key("true" if v else "false"), target)
+ |
+ | elif kind == "TIMESTAMP":
+ | # Both sides by getTime, so only whole milliseconds count. The
+ | # engine reads the value with DateParserUtils, which takes more
+ | # formats than pandas; a value pandas cannot read raises here.
+ | target = pd.Timestamp(java_trim(value)).floor("ms")
Review Comment:
Fixed in 55ce60480. A value with an offset is now read as the wall clock it
shows in the machine's zone before comparing, which is how the engine holds it.
A new test runs your `Z` equality case with the unzoned and null controls, with
both the JVM and Python set to Asia/Kolkata, so it means something under UTC CI
too.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala:
##########
@@ -63,42 +64,166 @@ class SpecializedFilterOpDesc extends FilterOpDesc with
StandaloneCodeGenerator
)
}
- override def generateStandaloneCode(): String = {
+ override def generateStandaloneCode(): String =
generateStandaloneCode(Map.empty)
+
+ override def generateStandaloneCode(inputSchemas: Map[PortIdentity,
Schema]): String = {
// No predicate keeps no row: the executor's filter is `predicates.exists`,
// which answers false on an empty list. Passing the frame through would be
// the opposite answer.
if (predicates.isEmpty) return "out1df = in1df.iloc[0:0].copy()"
+ val declared: Map[String, AttributeType] = inputSchemas.values.headOption
+ .map(_.getAttributes.map(a => a.getName -> a.getType).toMap)
+ .getOrElse(Map.empty)
val conditions = predicates.map { p =>
val colLit = pyStringLiteral(p.attribute)
p.condition match {
case ComparisonType.IS_NULL => s"""(in1df[$colLit].isna())"""
case ComparisonType.IS_NOT_NULL => s"""(in1df[$colLit].notna())"""
- case other =>
- val op = other.getName // returns "=", ">=", "<", etc. (see
ComparisonType.java)
- val pyOp = if (op == "=") "==" else op
- // notna mirrors FilterPredicate, which answers false for every
condition
- // but IS_NULL / IS_NOT_NULL once the field is null. Only `!=` needs
it —
- // pandas answers True there, where every other operator answers
False —
- // but guarding all of them keeps the one rule visible in one place.
- s"""(in1df[$colLit].notna() & (in1df[$colLit] $pyOp
${coerceValue(p.value)}))"""
+ case other =>
+ // Without a schema the column's dtype stands in for its type.
+ val kind = declared.get(p.attribute).map(t =>
pyStringLiteral(t.name)).getOrElse("None")
+ val opLit = pyStringLiteral(other.getName)
+ val valueLit = pyStringLiteral(Option(p.value).getOrElse(""))
+ s"_texera_filter_keep(in1df[$colLit], $opLit, $valueLit, $kind)"
}
}
s"out1df = in1df[${conditions.mkString(" | ")}].reset_index(drop=True)"
}
- // Try numeric coercion so generated code compares column values against the
right type.
- // Strings that don't parse fall through to a quoted string literal.
- private def coerceValue(raw: String): String = {
- try {
- raw.toInt.toString
- } catch {
- case _: NumberFormatException =>
- try {
- raw.toDouble.toString
- } catch {
- case _: NumberFormatException =>
- pyStringLiteral(raw)
- }
- }
- }
+ override def standaloneHelpers(): Seq[String] =
Seq(SpecializedFilterOpDesc.FilterPredicates)
+}
+
+object SpecializedFilterOpDesc {
+
+ /** `FilterPredicate.evaluate`, transcribed. pandas' operators answer
+ * differently on every type it handles apart: a STRING field and the value
+ * are compared as numbers when both parse as one, a BOOLEAN as its
+ * lower-case text, an INTEGER as a double, and a DOUBLE with
+ * `Double.compareTo`, under which NaN equals itself and sits above every
+ * number, and -0.0 sits below 0.0. A null field passes no comparison.
+ */
+ val FilterPredicates: String =
+ """def _texera_filter_keep(column, condition, value, kind):
+ | import math
+ | import re
+ |
+ | # String.trim drops every character up to U+0020, not only
whitespace.
+ | def java_trim(text):
+ | return text.strip("".join(chr(i) for i in range(33)))
+ |
+ | # Double.parseDouble: a trailing d or f is a type suffix and
0x...p...
+ | # is hex, where float() refuses both, and "nan", "inf" and "1_0",
+ | # which float() reads, are refused.
+ | java_double_text = re.compile(
+ | r"[+-]?(NaN|Infinity|"
+ | r"((\d+\.?\d*|\.\d+)([eE][+-]?\d+)?"
+ |
r"|0[xX]([0-9a-fA-F]+\.?|[0-9a-fA-F]*\.[0-9a-fA-F]+)[pP][+-]?\d+)[fFdD]?)",
+ | re.ASCII,
+ | )
+ |
+ | def java_double(text):
+ | text = java_trim(text)
+ | if not java_double_text.fullmatch(text):
+ | raise ValueError("For input string: " + repr(text))
+ | body = text.lstrip("+-")
+ | if body == "NaN":
+ | return float("nan")
+ | if body == "Infinity":
+ | return float("-inf") if text.startswith("-") else
float("inf")
+ | if text[-1] in "dDfF":
+ | text = text[:-1]
+ | if body[:2] in ("0x", "0X"):
+ | return float.fromhex(text)
Review Comment:
Fixed in 55ce60480. A hex value past the largest double now reads as
infinity with its sign, as `Double.parseDouble` does. The regressions cover
your Infinity, 1.0, null column against 0x1p1024 and -0x1p1024, a value that
rounds up past the largest double, a signed-zero underflow, and a hex value in
a STRING column.
--
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]
