carloea2 commented on code in PR #8980:
URL: https://github.com/apache/texera/pull/8980#discussion_r4291837138
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala:
##########
@@ -63,42 +64,179 @@ 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"):
+ | # Java rounds a hex value past the largest double to
infinity
+ | # where float.fromhex raises.
+ | try:
+ | return float.fromhex(text)
+ | except OverflowError:
+ | return float("-inf") if text.startswith("-") else
float("inf")
+ | 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.
+ | # A reading with an offset is an instant, which the engine
holds as
+ | # the wall clock it shows in the machine's zone.
+ | def wall_clock(ts):
+ | ts = pd.Timestamp(ts).floor("ms")
+ | if ts.tzinfo is None:
+ | return ts
+ | return
pd.Timestamp(ts.to_pydatetime().astimezone().replace(tzinfo=None))
Review Comment:
[P2] Preserve historical timezone offsets on Windows. In
America/Mexico_City, equality against 1931-01-01T00:00:00Z raises OSError here.
For 2021-07-01T00:00:00Z, this uses 18:00 instead of the native 19:00 and
silently drops a matching row. Modern, unzoned and null controls pass. Please
use historical IANA zone conversion, as in #8597, instead of implicit Windows
localtime.
--
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]
