From c8f2a15ce4fce022403a308e47cd6ed813a30df6 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 14 Oct 2026 01:49:18 -0700 Subject: [PATCH 1/8] fix(workflow-operator): compare Filter predicates the way the engine does The export compared the column to the value with pandas operators, while FilterPredicate compares by type: a STRING field and the value as numbers when both parse as one, a BOOLEAN as its lower-case text, an INTEGER as a double, a DOUBLE with Double.compareTo, and a TIMESTAMP by getTime. Each comparison now runs through a transcription of FilterPredicate.evaluate. Closes #8977 Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 175 +++++++++++++--- .../filter/SpecializedFilterOpDescSpec.scala | 189 +++++++++++++++++- 2 files changed, 337 insertions(+), 27 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index e5d47ee5232..acf32835dc3 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -21,8 +21,9 @@ package org.apache.texera.amber.operator.filter import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} -import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp} +import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral @@ -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") + | + | def order(v): + | return compared(pd.Timestamp(v).floor("ms"), target) + | + | else: + | try: + | value_number = double_key(java_double(value)) + | except ValueError: + | value_number = None + | + | def order(v): + | text = str(v) + | if value_number is not None: + | try: + | return compared(double_key(java_double(text)), value_number) + | except ValueError: + | pass + | return compared(text_key(text), text_key(value)) + | + | passes = { + | "=": lambda c: c == 0, + | ">": lambda c: c > 0, + | ">=": lambda c: c >= 0, + | "<": lambda c: c < 0, + | "<=": lambda c: c <= 0, + | "!=": lambda c: c != 0, + | }[condition] + | return pd.Series( + | [ + | False if missing else passes(order(v)) + | for v, missing in zip(column, column.isna()) + | ], + | index=column.index, + | dtype=bool, + | )""".stripMargin } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 39e1ec59988..786b1f70398 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -19,16 +19,31 @@ package org.apache.texera.amber.operator.filter +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.operator.tags.IntegrationTest import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.Tag import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.nio.charset.StandardCharsets +import java.nio.file.Files +import java.sql.Timestamp +import java.time.{LocalDateTime, ZoneOffset} +import java.util.concurrent.TimeUnit +import scala.io.Source +import scala.util.Try + class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { + private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName) + private val workflowId = WorkflowIdentity(1L) private val executionId = ExecutionIdentity(1L) @@ -71,8 +86,6 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { restored.asInstanceOf[SpecializedFilterOpDesc].predicates shouldBe empty } - // A null answers false for every condition but IS_NULL / IS_NOT_NULL, which pandas - // does not do on its own for `!=`, so the emitted condition carries the guard. "SpecializedFilterOpDesc.generateStandaloneCode" should "emit one condition per predicate" in { val d = new SpecializedFilterOpDesc d.predicates = List(new FilterPredicate("age", ComparisonType.GREATER_THAN, "18")) @@ -88,4 +101,176 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { "out1df = in1df.iloc[0:0].copy()" } + /** One column of a given type, the rows the engine reads and the frame the + * script reads, and the predicates run over it. + */ + private case class Column( + attrType: AttributeType, + values: Seq[Any], + pandas: String, + predicates: Seq[(ComparisonType, String)] + ) + + // A column holds a wall clock, which the engine reads in the JVM's zone and + // pandas holds without one, so both sides are built from the same wall clock. + private def timestamp(micros: Long): Timestamp = + Timestamp.valueOf( + LocalDateTime.ofEpochSecond( + Math.floorDiv(micros, 1000000L), + (Math.floorMod(micros, 1000000L) * 1000).toInt, + ZoneOffset.UTC + ) + ) + + // 2024-01-01 00:00:00 plus a few microseconds, and a millisecond later. + private val midnight = 1704067200000000L + + private val columns = Seq( + Column( + AttributeType.DOUBLE, + Seq(Double.NaN, -0.0, 0.0, 1.0, null, Double.PositiveInfinity), + """pd.arrays.FloatingArray( + | np.array([np.nan, -0.0, 0.0, 1.0, 0.0, np.inf]), + | np.array([False, False, False, False, True, False]), + |)""".stripMargin, + Seq( + ComparisonType.EQUAL_TO -> "0", + ComparisonType.GREATER_THAN -> "0", + ComparisonType.GREATER_THAN_OR_EQUAL_TO -> "0", + ComparisonType.LESS_THAN -> "0", + ComparisonType.LESS_THAN_OR_EQUAL_TO -> "0", + ComparisonType.NOT_EQUAL_TO -> "0", + ComparisonType.GREATER_THAN -> "1", + ComparisonType.EQUAL_TO -> "NaN", + ComparisonType.LESS_THAN -> "Infinity", + ComparisonType.EQUAL_TO -> "-0.0" + ) + ), + Column( + AttributeType.STRING, + Seq("1", "2", "10", "abc", null, " 3 ", "nan", "2d"), + """pd.Series(["1", "2", "10", "abc", None, " 3 ", "nan", "2d"], dtype=object)""", + Seq( + ComparisonType.EQUAL_TO -> "1", + ComparisonType.GREATER_THAN -> "2", + ComparisonType.EQUAL_TO -> "1.0", + ComparisonType.LESS_THAN -> "b", + ComparisonType.EQUAL_TO -> "nan", + ComparisonType.EQUAL_TO -> "2", + ComparisonType.NOT_EQUAL_TO -> "abc" + ) + ), + Column( + AttributeType.BOOLEAN, + Seq(true, false, null), + """pd.array([True, False, None], dtype="boolean")""", + Seq( + ComparisonType.EQUAL_TO -> "True", + ComparisonType.EQUAL_TO -> "1", + ComparisonType.GREATER_THAN -> "false" + ) + ), + Column( + AttributeType.INTEGER, + Seq(1, 2, null, -3), + """pd.array([1, 2, None, -3], dtype="Int32")""", + Seq(ComparisonType.GREATER_THAN -> "1.5", ComparisonType.EQUAL_TO -> "2") + ), + Column( + AttributeType.LONG, + Seq(1L, 9007199254740993L, null), + """pd.array([1, 9007199254740993, None], dtype="Int64")""", + Seq( + ComparisonType.EQUAL_TO -> "9007199254740993", + ComparisonType.GREATER_THAN -> " 1 " + ) + ), + Column( + AttributeType.TIMESTAMP, + Seq(midnight, midnight + 900, midnight + 1000).map(timestamp), + s"""pd.Series(np.array([$midnight, ${midnight + 900}, ${midnight + 1000}], dtype="datetime64[us]"))""", + Seq( + ComparisonType.EQUAL_TO -> "2024-01-01 00:00:00", + ComparisonType.GREATER_THAN -> "2024-01-01 00:00:00" + ) + ) + ) + + // FilterPredicate compares by the column's type: a STRING field and the value + // as numbers when both parse as one, a BOOLEAN as its lower-case text, an + // INTEGER as a double, a DOUBLE by Double.compareTo, and a TIMESTAMP by + // getTime. Each predicate runs through SpecializedFilterOpExec and through + // the export, and the rows each keeps are named by their input position. + it should "keep the rows the engine keeps for each column type" taggedAs NeedsPythonPackages in { + val python = resolvePython().getOrElse(cancel("No runnable python executable")) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val runs = for { + column <- columns + (condition, value) <- column.predicates + } yield { + val schema = Schema().add("v", column.attrType).add("i", AttributeType.INTEGER) + val desc = new SpecializedFilterOpDesc + desc.predicates = List(new FilterPredicate("v", condition, value)) + val exec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(desc)) + exec.open() + val kept = column.values.zipWithIndex.flatMap { + case (v, i) => exec.processTuple(Tuple(schema, Array[Any](v, i)), 0) + } + exec.close() + val label = s"${column.attrType} ${condition.getName} '$value'" + val engine = s"$label: ${kept.map(_.getFields(1)).mkString(" ")}" + val code = desc.generateStandaloneCode(Map(PortIdentity() -> schema)) + val script = + s"""in1df = pd.DataFrame({"v": ${column.pandas.linesIterator.mkString("\n ")}}) + |in1df["i"] = range(len(in1df)) + |ns = {"pd": pd, "in1df": in1df} + |exec(${objectMapper.writeValueAsString( + (desc.standaloneHelpers() :+ code).mkString("\n") + )}, ns) + |print(${objectMapper.writeValueAsString( + label + ":" + )}, " ".join(str(i) for i in ns["out1df"]["i"]))""".stripMargin + (engine, script) + } + + val driver = ("import pandas as pd, numpy as np" +: runs.map(_._2)).mkString("\n") + val file = Files.createTempFile("filter-engine-parity-", ".py") + file.toFile.deleteOnExit() + Files.write(file, driver.getBytes(StandardCharsets.UTF_8)) + val process = new ProcessBuilder(python, file.toString).redirectErrorStream(true).start() + val out = Source.fromInputStream(process.getInputStream).mkString + process.waitFor(120, TimeUnit.SECONDS) + + withClue(s"python said:\n$out\nscript:\n$driver") { + process.exitValue() shouldBe 0 + out.linesIterator.map(_.trim).toList shouldBe runs.map(_._1.trim).toList + } + } + + private def resolvePython(): Option[String] = { + def fromConfig: Option[String] = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + + def runnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable) + } + + private def canImportPandas(python: String): Boolean = + Try( + new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start() + ).toOption.exists { p => + if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } } From 55ce604802b17d2793fd20a98347e6b101a7b48e Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 15 Oct 2026 22:56:52 -0700 Subject: [PATCH 2/8] fix(workflow-operator): read a Filter predicate's offset and hex overflow as the engine does An offset-bearing timestamp value is an instant, which the engine holds as the wall clock it shows in the machine's zone, so the export now converts it there before comparing instead of raising on a naive column. A hex double past the largest double is infinity in Java, so the export no longer lets float.fromhex raise OverflowError. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 19 ++++- .../filter/SpecializedFilterOpDescSpec.scala | 71 +++++++++++++++++-- 2 files changed, 83 insertions(+), 7 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index acf32835dc3..3a639d71165 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -133,7 +133,12 @@ object SpecializedFilterOpDesc { | if text[-1] in "dDfF": | text = text[:-1] | if body[:2] in ("0x", "0X"): - | return float.fromhex(text) + | # 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 @@ -190,10 +195,18 @@ object SpecializedFilterOpDesc { | # 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") + | # 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)) + | + | target = wall_clock(java_trim(value)) | | def order(v): - | return compared(pd.Timestamp(v).floor("ms"), target) + | return compared(wall_clock(v), target) | | else: | try: diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 786b1f70398..304885c8d35 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -36,6 +36,7 @@ import java.nio.charset.StandardCharsets import java.nio.file.Files import java.sql.Timestamp import java.time.{LocalDateTime, ZoneOffset} +import java.util.TimeZone import java.util.concurrent.TimeUnit import scala.io.Source import scala.util.Try @@ -143,14 +144,33 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { ComparisonType.GREATER_THAN -> "1", ComparisonType.EQUAL_TO -> "NaN", ComparisonType.LESS_THAN -> "Infinity", - ComparisonType.EQUAL_TO -> "-0.0" + ComparisonType.EQUAL_TO -> "-0.0", + // Hex past the largest double rounds to infinity, and below the + // smallest to a zero that keeps its sign. + ComparisonType.EQUAL_TO -> "0x1.fffffffffffff8p1023", + ComparisonType.EQUAL_TO -> "-0x1p-1080" + ) + ), + Column( + AttributeType.DOUBLE, + Seq(Double.PositiveInfinity, 1.0, null, Double.NegativeInfinity), + """pd.arrays.FloatingArray( + | np.array([np.inf, 1.0, 0.0, -np.inf]), + | np.array([False, False, True, False]), + |)""".stripMargin, + Seq( + ComparisonType.EQUAL_TO -> "0x1p1024", + ComparisonType.EQUAL_TO -> "-0x1p1024", + ComparisonType.LESS_THAN -> "0x1p1024d", + ComparisonType.EQUAL_TO -> "0x1p0" ) ), Column( AttributeType.STRING, - Seq("1", "2", "10", "abc", null, " 3 ", "nan", "2d"), - """pd.Series(["1", "2", "10", "abc", None, " 3 ", "nan", "2d"], dtype=object)""", + Seq("1", "2", "10", "abc", null, " 3 ", "nan", "2d", "0x1p2000"), + """pd.Series(["1", "2", "10", "abc", None, " 3 ", "nan", "2d", "0x1p2000"], dtype=object)""", Seq( + ComparisonType.EQUAL_TO -> "Infinity", ComparisonType.EQUAL_TO -> "1", ComparisonType.GREATER_THAN -> "2", ComparisonType.EQUAL_TO -> "1.0", @@ -202,6 +222,47 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { // getTime. Each predicate runs through SpecializedFilterOpExec and through // the export, and the rows each keeps are named by their input position. it should "keep the rows the engine keeps for each column type" taggedAs NeedsPythonPackages in { + expectTheRowsTheEngineKeeps(columns) + } + + // The engine holds a value with an offset as the wall clock it shows in the + // JVM's zone. A zone away from UTC, and a half hour off any whole hour, so a + // reading in UTC or a dropped offset keeps a different row. + it should "read a timestamp value with an offset in the machine's zone" taggedAs + NeedsPythonPackages in { + val zone = "Asia/Kolkata" + val previous = TimeZone.getDefault + TimeZone.setDefault(TimeZone.getTimeZone(zone)) + try { + val utcMidnight = midnight + 19800000000L + expectTheRowsTheEngineKeeps( + Seq( + Column( + AttributeType.TIMESTAMP, + Seq(timestamp(utcMidnight), timestamp(midnight), null), + s"""pd.Series(np.array([$utcMidnight, $midnight, "NaT"], dtype="datetime64[us]"))""", + Seq( + ComparisonType.EQUAL_TO -> "2024-01-01T00:00:00Z", + ComparisonType.EQUAL_TO -> "2024-01-01 00:00:00", + ComparisonType.EQUAL_TO -> "2024-01-01T00:00:00+05:30", + ComparisonType.EQUAL_TO -> "2023-12-31T16:00:00-08:00", + ComparisonType.GREATER_THAN -> "2024-01-01T00:00:00Z", + ComparisonType.LESS_THAN -> "2024-01-01T00:00:00Z" + ) + ) + ), + pythonZone = Some(zone) + ) + } finally TimeZone.setDefault(previous) + } + + /** Runs each predicate through SpecializedFilterOpExec and through the export, + * and expects the export to keep the same rows, named by input position. + */ + private def expectTheRowsTheEngineKeeps( + columns: Seq[Column], + pythonZone: Option[String] = None + ): Unit = { val python = resolvePython().getOrElse(cancel("No runnable python executable")) if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") @@ -238,7 +299,9 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { val file = Files.createTempFile("filter-engine-parity-", ".py") file.toFile.deleteOnExit() Files.write(file, driver.getBytes(StandardCharsets.UTF_8)) - val process = new ProcessBuilder(python, file.toString).redirectErrorStream(true).start() + val builder = new ProcessBuilder(python, file.toString).redirectErrorStream(true) + pythonZone.foreach(builder.environment().put("TZ", _)) + val process = builder.start() val out = Source.fromInputStream(process.getInputStream).mkString process.waitFor(120, TimeUnit.SECONDS) From e5b2ba2ca28a7b237acf514f1da2f09e1269adc0 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 16 Oct 2026 02:24:27 -0700 Subject: [PATCH 3/8] fix(workflow-operator): take a Filter offset value's historical zone offset A timestamp value with an offset now becomes the machine-zone wall clock through #8597's _texera_local_wall_clock, which uses the zone's IANA history instead of Windows localtime. Adds a Mexico City test for 2021 summer time and 1931. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 12 ++++--- .../filter/SpecializedFilterOpDescSpec.scala | 32 +++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 3a639d71165..802f90997ad 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -24,7 +24,7 @@ import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} -import org.apache.texera.amber.operator.StandaloneCodeGenerator +import org.apache.texera.amber.operator.{StandaloneCodeGenerator, StandaloneHelpers} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -90,7 +90,8 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator s"out1df = in1df[${conditions.mkString(" | ")}].reset_index(drop=True)" } - override def standaloneHelpers(): Seq[String] = Seq(SpecializedFilterOpDesc.FilterPredicates) + override def standaloneHelpers(): Seq[String] = + Seq(StandaloneHelpers.AttributeCasts, SpecializedFilterOpDesc.FilterPredicates) } object SpecializedFilterOpDesc { @@ -196,12 +197,15 @@ object SpecializedFilterOpDesc { | # 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. + | # the wall clock it shows in the machine's zone, with the offset + | # that zone had at that instant. + | zone = _texera_local_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)) + | return pd.Timestamp(_texera_local_wall_clock(ts.to_pydatetime(), zone)) | | target = wall_clock(java_trim(value)) | diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 304885c8d35..f83b1c48329 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -256,6 +256,38 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { } finally TimeZone.setDefault(previous) } + // The offset is the one the zone had at that instant, not today's. Mexico + // City kept summer time in 2021 and stood an hour further west in 1931, so + // today's standard offset keeps the other row of each pair. + it should "read a timestamp value with the offset its zone had at that instant" taggedAs + NeedsPythonPackages in { + val zone = "America/Mexico_City" + val previous = TimeZone.getDefault + TimeZone.setDefault(TimeZone.getTimeZone(zone)) + try { + val walls = Seq( + LocalDateTime.of(2021, 6, 30, 19, 0), + LocalDateTime.of(2021, 6, 30, 18, 0), + LocalDateTime.of(1930, 12, 31, 17, 0), + LocalDateTime.of(1930, 12, 31, 18, 0) + ).map(_.toEpochSecond(ZoneOffset.UTC) * 1000000L) + expectTheRowsTheEngineKeeps( + Seq( + Column( + AttributeType.TIMESTAMP, + walls.map(timestamp), + s"""pd.Series(np.array([${walls.mkString(", ")}], dtype="datetime64[us]"))""", + Seq( + ComparisonType.EQUAL_TO -> "2021-07-01T00:00:00Z", + ComparisonType.EQUAL_TO -> "1931-01-01T00:00:00Z" + ) + ) + ), + pythonZone = Some(zone) + ) + } finally TimeZone.setDefault(previous) + } + /** Runs each predicate through SpecializedFilterOpExec and through the export, * and expects the export to keep the same rows, named by input position. */ From fcff3cd64c0fd296ad5b18ad291c0a54ec3202ca Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 23 Oct 2026 14:43:47 -0700 Subject: [PATCH 4/8] fix(operator): take Filter's declared type through the schema's case-insensitive lookup The engine finds a predicate's field through Schema, which ignores case. An exact-name lookup found no type for a column named in another case, fell back to the dtype, and compared a LONG as a double. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 17 ++++--- .../filter/SpecializedFilterOpDescSpec.scala | 45 ++++++++++++++++--- 2 files changed, 51 insertions(+), 11 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 802f90997ad..56d32f27cd1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.filter import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.operator.{StandaloneCodeGenerator, StandaloneHelpers} @@ -29,6 +29,8 @@ import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, Operat import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.util.JSONUtils.objectMapper +import scala.util.Try + class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator { @JsonProperty(value = "predicates", required = true) @@ -71,17 +73,20 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator // 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 schema = inputSchemas.values.headOption 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 => - // 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") + // FilterPredicate finds the field through Schema, which ignores case, + // and compares by the type declared there. Without a schema the + // column's dtype stands in for its type. + val kind = schema + .flatMap(s => Try(s.getAttribute(p.attribute)).toOption) + .map(a => pyStringLiteral(a.getType.name)) + .getOrElse("None") val opLit = pyStringLiteral(other.getName) val valueLit = pyStringLiteral(Option(p.value).getOrElse("")) s"_texera_filter_keep(in1df[$colLit], $opLit, $valueLit, $kind)" diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index f83b1c48329..82474f0eb46 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -102,14 +102,28 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { "out1df = in1df.iloc[0:0].copy()" } + // FilterPredicate reads `KEY` from a column `Key`, since Schema ignores case, + // and compares it by that column's declared type. Read from the dtype, a LONG + // column would compare as a double and lose the digits past 2^53. + it should "take the declared type of a column named in a different case" in { + val d = new SpecializedFilterOpDesc + d.predicates = List(new FilterPredicate("KEY", ComparisonType.EQUAL_TO, "9007199254740993")) + val code = + d.generateStandaloneCode(Map(PortIdentity() -> Schema().add("Key", AttributeType.LONG))) + code should include(""""=", "9007199254740993", "LONG")""") + } + /** One column of a given type, the rows the engine reads and the frame the - * script reads, and the predicates run over it. + * script reads, and the predicates run over it. The table spells the column + * `name`, and the predicates name it `predicateName`. */ private case class Column( attrType: AttributeType, values: Seq[Any], pandas: String, - predicates: Seq[(ComparisonType, String)] + predicates: Seq[(ComparisonType, String)], + name: String = "v", + predicateName: String = "v" ) // A column holds a wall clock, which the engine reads in the JVM's zone and @@ -225,6 +239,25 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { expectTheRowsTheEngineKeeps(columns) } + // FilterPredicate reads `KEY` from a column `Key`, since Schema ignores case, + // and compares it as the LONG declared there. 2^53 + 1 then keeps only its own + // row, where a double holds both values as 2^53 and keeps both. + it should "compare a column named in a different case by its declared type" taggedAs + NeedsPythonPackages in { + expectTheRowsTheEngineKeeps( + Seq( + Column( + AttributeType.LONG, + Seq(9007199254740992L, 9007199254740993L), + """pd.array([9007199254740992, 9007199254740993], dtype="Int64")""", + Seq(ComparisonType.EQUAL_TO -> "9007199254740993"), + name = "Key", + predicateName = "KEY" + ) + ) + ) + } + // The engine holds a value with an offset as the wall clock it shows in the // JVM's zone. A zone away from UTC, and a half hour off any whole hour, so a // reading in UTC or a dropped offset keeps a different row. @@ -302,9 +335,9 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { column <- columns (condition, value) <- column.predicates } yield { - val schema = Schema().add("v", column.attrType).add("i", AttributeType.INTEGER) + val schema = Schema().add(column.name, column.attrType).add("i", AttributeType.INTEGER) val desc = new SpecializedFilterOpDesc - desc.predicates = List(new FilterPredicate("v", condition, value)) + desc.predicates = List(new FilterPredicate(column.predicateName, condition, value)) val exec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(desc)) exec.open() val kept = column.values.zipWithIndex.flatMap { @@ -315,7 +348,9 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { val engine = s"$label: ${kept.map(_.getFields(1)).mkString(" ")}" val code = desc.generateStandaloneCode(Map(PortIdentity() -> schema)) val script = - s"""in1df = pd.DataFrame({"v": ${column.pandas.linesIterator.mkString("\n ")}}) + s"""in1df = pd.DataFrame({${objectMapper.writeValueAsString( + column.name + )}: ${column.pandas.linesIterator.mkString("\n ")}}) |in1df["i"] = range(len(in1df)) |ns = {"pd": pd, "in1df": in1df} |exec(${objectMapper.writeValueAsString( From d9eee262cfc89d8cd1c31d8995bc857c4a8cf790 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 23 Oct 2026 15:20:51 -0700 Subject: [PATCH 5/8] feat(workflow-operator): keep Filter's comparison notes out of the exported script _texera_filter_keep explained each comparison in Python comments, which every exported script with a Filter printed. The rules now sit in the Scaladoc, one sentence per type. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 28 ++++--------------- 1 file changed, 5 insertions(+), 23 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 56d32f27cd1..34e92288a14 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -101,25 +101,20 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator 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. + /** `_texera_filter_keep`: `FilterPredicate.evaluate`, comparing each type as the engine does. + * STRING compares as numbers when both sides parse, else by UTF-16 code units. BOOLEAN + * compares as lower-case text. INTEGER and DOUBLE use `Double.compareTo`, so NaN sits above + * every number and -0.0 below 0.0. LONG reads the value with `Long.valueOf`. TIMESTAMP + * compares whole milliseconds in the machine's zone. A null passes nothing. */ 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+)?" @@ -139,23 +134,17 @@ object SpecializedFilterOpDesc { | 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") | @@ -182,7 +171,6 @@ object SpecializedFilterOpDesc { | 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)) @@ -198,12 +186,6 @@ object SpecializedFilterOpDesc { | 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, with the offset - | # that zone had at that instant. | zone = _texera_local_zone() | | def wall_clock(ts): From 3760bcdabcf4af427abd0b9ce95b32fa373164c2 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 23 Oct 2026 19:37:45 -0700 Subject: [PATCH 6/8] feat(workflow-operator): export only the Filter comparison a column's type needs _texera_filter_keep held every type's comparison in one function, so a Filter on a DOUBLE column carried the LONG, BOOLEAN, TIMESTAMP and STRING code too. Each type now has its own function, the generated code calls the one its column's declared type needs, and the translator drops the rest. Without a schema, _texera_filter_keep still picks one by dtype. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 259 ++++++++++-------- .../filter/SpecializedFilterOpDescSpec.scala | 3 +- 2 files changed, 152 insertions(+), 110 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 34e92288a14..451fdb7cb3d 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.filter import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName -import org.apache.texera.amber.core.tuple.Schema +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.operator.{StandaloneCodeGenerator, StandaloneHelpers} @@ -83,13 +83,13 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator // FilterPredicate finds the field through Schema, which ignores case, // and compares by the type declared there. Without a schema the // column's dtype stands in for its type. - val kind = schema + val keep = schema .flatMap(s => Try(s.getAttribute(p.attribute)).toOption) - .map(a => pyStringLiteral(a.getType.name)) - .getOrElse("None") + .map(a => SpecializedFilterOpDesc.keepFor(a.getType)) + .getOrElse("_texera_filter_keep") val opLit = pyStringLiteral(other.getName) val valueLit = pyStringLiteral(Option(p.value).getOrElse("")) - s"_texera_filter_keep(in1df[$colLit], $opLit, $valueLit, $kind)" + s"$keep(in1df[$colLit], $opLit, $valueLit)" } } s"out1df = in1df[${conditions.mkString(" | ")}].reset_index(drop=True)" @@ -101,119 +101,79 @@ class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator object SpecializedFilterOpDesc { - /** `_texera_filter_keep`: `FilterPredicate.evaluate`, comparing each type as the engine does. - * STRING compares as numbers when both sides parse, else by UTF-16 code units. BOOLEAN - * compares as lower-case text. INTEGER and DOUBLE use `Double.compareTo`, so NaN sits above - * every number and -0.0 below 0.0. LONG reads the value with `Long.valueOf`. TIMESTAMP - * compares whole milliseconds in the machine's zone. A null passes nothing. + /** The function that compares a column of `attributeType`, so a script calls only the one + * its column needs. + */ + def keepFor(attributeType: AttributeType): String = + attributeType match { + case AttributeType.INTEGER | AttributeType.DOUBLE => "_texera_filter_keep_double" + case AttributeType.LONG => "_texera_filter_keep_long" + case AttributeType.BOOLEAN => "_texera_filter_keep_boolean" + case AttributeType.TIMESTAMP => "_texera_filter_keep_timestamp" + case _ => "_texera_filter_keep_string" + } + + /** `FilterPredicate.evaluate`, one function per type the engine compares apart. A null passes + * nothing. + * + * - `_texera_filter_keep_double`: INTEGER and DOUBLE by `Double.compareTo`, NaN above all. + * - `_texera_filter_keep_long`: LONG, the value read by `Long.valueOf`. + * - `_texera_filter_keep_boolean`: BOOLEAN as lower-case text. + * - `_texera_filter_keep_timestamp`: TIMESTAMP by whole milliseconds in the machine's zone. + * - `_texera_filter_keep_string`: STRING as numbers when both parse, else by UTF-16 units. + * - `_texera_filter_keep`: the type read off the column's dtype, when no schema says it. + * - `_texera_filter_java_double`: `Double.parseDouble`, refusing what Java refuses. + * - `_texera_filter_double_key`: `Double.compareTo`'s order, -0.0 below 0.0. */ val FilterPredicates: String = - """def _texera_filter_keep(column, condition, value, kind): - | import math - | import re + """def _texera_filter_java_trim(text): + | return text.strip("".join(chr(i) for i in range(33))) + | | - | def java_trim(text): - | return text.strip("".join(chr(i) for i in range(33))) + |def _texera_filter_java_double(text): + | import re | - | java_double_text = re.compile( + | text = _texera_filter_java_trim(text) + | if not re.fullmatch( | 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]?)", + | text, | 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"): - | try: - | return float.fromhex(text) - | except OverflowError: - | return float("-inf") if text.startswith("-") else float("inf") - | return float(text) - | - | def double_key(x): - | if math.isnan(x): - | return (1, 0.0, 0.0) - | return (0, x, math.copysign(1.0, x)) - | - | 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": - | 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": - | zone = _texera_local_zone() - | - | def wall_clock(ts): - | ts = pd.Timestamp(ts).floor("ms") - | if ts.tzinfo is None: - | return ts - | return pd.Timestamp(_texera_local_wall_clock(ts.to_pydatetime(), zone)) - | - | target = wall_clock(java_trim(value)) - | - | def order(v): - | return compared(wall_clock(v), target) - | - | else: + | ): + | 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"): | try: - | value_number = double_key(java_double(value)) - | except ValueError: - | value_number = None - | - | def order(v): - | text = str(v) - | if value_number is not None: - | try: - | return compared(double_key(java_double(text)), value_number) - | except ValueError: - | pass - | return compared(text_key(text), text_key(value)) + | return float.fromhex(text) + | except OverflowError: + | return float("-inf") if text.startswith("-") else float("inf") + | return float(text) + | | + |def _texera_filter_double_key(x): + | import math + | + | if math.isnan(x): + | return (1, 0.0, 0.0) + | return (0, x, math.copysign(1.0, x)) + | + | + |def _texera_filter_text_key(s): + | return s.encode("utf-16-be", "surrogatepass") + | + | + |def _texera_filter_compared(left, right): + | return (left > right) - (left < right) + | + | + |def _texera_filter_rows(column, condition, order): | passes = { | "=": lambda c: c == 0, | ">": lambda c: c > 0, @@ -229,5 +189,86 @@ object SpecializedFilterOpDesc { | ], | index=column.index, | dtype=bool, - | )""".stripMargin + | ) + | + | + |def _texera_filter_keep_double(column, condition, value): + | target = _texera_filter_double_key(_texera_filter_java_double(value)) + | + | def order(v): + | return _texera_filter_compared(_texera_filter_double_key(float(v)), target) + | + | return _texera_filter_rows(column, condition, order) + | + | + |def _texera_filter_keep_long(column, condition, value): + | import re + | + | text = _texera_filter_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 _texera_filter_compared(int(v), target) + | + | return _texera_filter_rows(column, condition, order) + | + | + |def _texera_filter_keep_boolean(column, condition, value): + | target = _texera_filter_text_key(_texera_filter_java_trim(value).lower()) + | + | def order(v): + | return _texera_filter_compared(_texera_filter_text_key("true" if v else "false"), target) + | + | return _texera_filter_rows(column, condition, order) + | + | + |def _texera_filter_keep_timestamp(column, condition, value): + | zone = _texera_local_zone() + | + | def wall_clock(ts): + | ts = pd.Timestamp(ts).floor("ms") + | if ts.tzinfo is None: + | return ts + | return pd.Timestamp(_texera_local_wall_clock(ts.to_pydatetime(), zone)) + | + | target = wall_clock(_texera_filter_java_trim(value)) + | + | def order(v): + | return _texera_filter_compared(wall_clock(v), target) + | + | return _texera_filter_rows(column, condition, order) + | + | + |def _texera_filter_keep_string(column, condition, value): + | try: + | value_number = _texera_filter_double_key(_texera_filter_java_double(value)) + | except ValueError: + | value_number = None + | + | def order(v): + | text = str(v) + | if value_number is not None: + | try: + | number = _texera_filter_double_key(_texera_filter_java_double(text)) + | return _texera_filter_compared(number, value_number) + | except ValueError: + | pass + | return _texera_filter_compared( + | _texera_filter_text_key(text), _texera_filter_text_key(value) + | ) + | + | return _texera_filter_rows(column, condition, order) + | + | + |def _texera_filter_keep(column, condition, value): + | dtype = column.dtype + | if pd.api.types.is_bool_dtype(dtype): + | return _texera_filter_keep_boolean(column, condition, value) + | if pd.api.types.is_integer_dtype(dtype) or pd.api.types.is_float_dtype(dtype): + | return _texera_filter_keep_double(column, condition, value) + | if pd.api.types.is_datetime64_any_dtype(dtype): + | return _texera_filter_keep_timestamp(column, condition, value) + | return _texera_filter_keep_string(column, condition, value)""".stripMargin } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 82474f0eb46..fb0dfdcbe49 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -110,7 +110,8 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { d.predicates = List(new FilterPredicate("KEY", ComparisonType.EQUAL_TO, "9007199254740993")) val code = d.generateStandaloneCode(Map(PortIdentity() -> Schema().add("Key", AttributeType.LONG))) - code should include(""""=", "9007199254740993", "LONG")""") + code should include("_texera_filter_keep_long(") + code should include(""""=", "9007199254740993")""") } /** One column of a given type, the rows the engine reads and the frame the From 21955685ab4bb110f37785cd13c5e3b635e531e7 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 24 Oct 2026 00:53:51 -0700 Subject: [PATCH 7/8] feat(workflow-operator): compare Filter values with the shared Java number parsers Filter kept its own String.trim, Double.parseDouble and a LONG check that matched Long.parseLong. It now calls _texera_java_trim, _texera_java_double and _texera_parse_int from AttributeCasts. Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 54 ++++--------------- 1 file changed, 10 insertions(+), 44 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 451fdb7cb3d..669428bb84d 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -122,42 +122,13 @@ object SpecializedFilterOpDesc { * - `_texera_filter_keep_timestamp`: TIMESTAMP by whole milliseconds in the machine's zone. * - `_texera_filter_keep_string`: STRING as numbers when both parse, else by UTF-16 units. * - `_texera_filter_keep`: the type read off the column's dtype, when no schema says it. - * - `_texera_filter_java_double`: `Double.parseDouble`, refusing what Java refuses. * - `_texera_filter_double_key`: `Double.compareTo`'s order, -0.0 below 0.0. + * + * The value is read with `_texera_java_trim`, `_texera_java_double` and `_texera_parse_int` + * from [[StandaloneHelpers.AttributeCasts]]. */ val FilterPredicates: String = - """def _texera_filter_java_trim(text): - | return text.strip("".join(chr(i) for i in range(33))) - | - | - |def _texera_filter_java_double(text): - | import re - | - | text = _texera_filter_java_trim(text) - | if not re.fullmatch( - | 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]?)", - | text, - | re.ASCII, - | ): - | 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"): - | try: - | return float.fromhex(text) - | except OverflowError: - | return float("-inf") if text.startswith("-") else float("inf") - | return float(text) - | - | - |def _texera_filter_double_key(x): + """def _texera_filter_double_key(x): | import math | | if math.isnan(x): @@ -193,7 +164,7 @@ object SpecializedFilterOpDesc { | | |def _texera_filter_keep_double(column, condition, value): - | target = _texera_filter_double_key(_texera_filter_java_double(value)) + | target = _texera_filter_double_key(_texera_java_double(value)) | | def order(v): | return _texera_filter_compared(_texera_filter_double_key(float(v)), target) @@ -202,12 +173,7 @@ object SpecializedFilterOpDesc { | | |def _texera_filter_keep_long(column, condition, value): - | import re - | - | text = _texera_filter_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) + | target = _texera_parse_int(value, 64) | | def order(v): | return _texera_filter_compared(int(v), target) @@ -216,7 +182,7 @@ object SpecializedFilterOpDesc { | | |def _texera_filter_keep_boolean(column, condition, value): - | target = _texera_filter_text_key(_texera_filter_java_trim(value).lower()) + | target = _texera_filter_text_key(_texera_java_trim(value).lower()) | | def order(v): | return _texera_filter_compared(_texera_filter_text_key("true" if v else "false"), target) @@ -233,7 +199,7 @@ object SpecializedFilterOpDesc { | return ts | return pd.Timestamp(_texera_local_wall_clock(ts.to_pydatetime(), zone)) | - | target = wall_clock(_texera_filter_java_trim(value)) + | target = wall_clock(_texera_java_trim(value)) | | def order(v): | return _texera_filter_compared(wall_clock(v), target) @@ -243,7 +209,7 @@ object SpecializedFilterOpDesc { | |def _texera_filter_keep_string(column, condition, value): | try: - | value_number = _texera_filter_double_key(_texera_filter_java_double(value)) + | value_number = _texera_filter_double_key(_texera_java_double(value)) | except ValueError: | value_number = None | @@ -251,7 +217,7 @@ object SpecializedFilterOpDesc { | text = str(v) | if value_number is not None: | try: - | number = _texera_filter_double_key(_texera_filter_java_double(text)) + | number = _texera_filter_double_key(_texera_java_double(text)) | return _texera_filter_compared(number, value_number) | except ValueError: | pass From c21c9a2018b8f8fd6b5fad9041c6879b20a41998 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 24 Oct 2026 14:54:01 -0700 Subject: [PATCH 8/8] feat(workflow-operator): order Filter's doubles with the shared key Co-Authored-By: Claude Opus 5.5 --- .../filter/SpecializedFilterOpDesc.scala | 24 +++++++------------ 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 669428bb84d..c0a298d8393 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -122,21 +122,13 @@ object SpecializedFilterOpDesc { * - `_texera_filter_keep_timestamp`: TIMESTAMP by whole milliseconds in the machine's zone. * - `_texera_filter_keep_string`: STRING as numbers when both parse, else by UTF-16 units. * - `_texera_filter_keep`: the type read off the column's dtype, when no schema says it. - * - `_texera_filter_double_key`: `Double.compareTo`'s order, -0.0 below 0.0. * - * The value is read with `_texera_java_trim`, `_texera_java_double` and `_texera_parse_int` - * from [[StandaloneHelpers.AttributeCasts]]. + * The value is read with `_texera_java_trim`, `_texera_java_double` and `_texera_parse_int`, + * and a double is ordered by `_texera_double_key`, all from + * [[StandaloneHelpers.AttributeCasts]]. */ val FilterPredicates: String = - """def _texera_filter_double_key(x): - | import math - | - | if math.isnan(x): - | return (1, 0.0, 0.0) - | return (0, x, math.copysign(1.0, x)) - | - | - |def _texera_filter_text_key(s): + """def _texera_filter_text_key(s): | return s.encode("utf-16-be", "surrogatepass") | | @@ -164,10 +156,10 @@ object SpecializedFilterOpDesc { | | |def _texera_filter_keep_double(column, condition, value): - | target = _texera_filter_double_key(_texera_java_double(value)) + | target = _texera_double_key(_texera_java_double(value)) | | def order(v): - | return _texera_filter_compared(_texera_filter_double_key(float(v)), target) + | return _texera_filter_compared(_texera_double_key(float(v)), target) | | return _texera_filter_rows(column, condition, order) | @@ -209,7 +201,7 @@ object SpecializedFilterOpDesc { | |def _texera_filter_keep_string(column, condition, value): | try: - | value_number = _texera_filter_double_key(_texera_java_double(value)) + | value_number = _texera_double_key(_texera_java_double(value)) | except ValueError: | value_number = None | @@ -217,7 +209,7 @@ object SpecializedFilterOpDesc { | text = str(v) | if value_number is not None: | try: - | number = _texera_filter_double_key(_texera_java_double(text)) + | number = _texera_double_key(_texera_java_double(text)) | return _texera_filter_compared(number, value_number) | except ValueError: | pass