From 45d52d424467af69a90fd7fa282582a5402f10bf Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 12:54:24 -0700 Subject: [PATCH 1/6] feat(operator): read a timestamp cast over the years the engine reaches pandas parses into nanoseconds, which reach only 1677 to 2262, so the exported script emptied every moment outside that window where the engine holds a java.sql.Timestamp and reads it like any other. It also inferred one format for the whole column, so a row written differently from the first one was emptied too, where the engine hands DateParserUtils a field at a time. The text branch now reads the column cell by cell at microsecond resolution. Text neither side can read still leaves an empty cell rather than ending the run, which is what the cast has always promised. Closes #8595 Co-Authored-By: Claude Opus 5 (1M context) --- .../amber/operator/StandaloneHelpers.scala | 29 +++++ .../typecasting/TypeCastingOpDesc.scala | 8 +- .../typecasting/TypeCastingOpDescSpec.scala | 119 ++++++++++++++++++ 3 files changed, 153 insertions(+), 3 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index 1414f6b043c..e70477c1ff1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -122,6 +122,35 @@ object StandaloneHelpers { | return max(-2147483648, min(2147483647, int(value))) | | + |def _texera_text_to_timestamp(s): + | # Read cell by cell, the way the engine reads a column: DateParserUtils + | # is handed one field at a time, so a row states its own format and the + | # rest of the column has no say in it. + | # + | # Held at microsecond resolution and not the nanoseconds pandas parses + | # into by default, which reach 1677 to 2262: the engine holds a + | # java.sql.Timestamp, where the year 2500 is an ordinary moment and + | # emptying it would answer for a row the run itself had no trouble with. + | # + | # Still coerced, which the strict cast is not: the engine accepts a set + | # of formats no single pandas call states, so text neither can read is + | # answered with an empty cell rather than by ending the run. + | from dateutil.parser import parse as _parse_date + | + | if pd.api.types.is_datetime64_any_dtype(s): + | return s.astype("datetime64[us]") + | + | def _one(x): + | if pd.isna(x): + | return None + | try: + | return _parse_date(str(x).strip()) + | except (ValueError, OverflowError): + | return None + | + | return s.map(_one).astype("datetime64[us]") + | + | |def _texera_epoch_millis_to_timestamp(s): | # `new Timestamp(long)` reads MILLISECONDS where pd.to_datetime defaults | # to nanoseconds, and renders in the JVM's default zone, so leaving the diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala index c23c3dd66f0..7a1422c7154 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala @@ -129,11 +129,13 @@ class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { // 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 => - // A number is an instant in milliseconds and needs its own reading; - // see the helper. Text keeps the parser it already had. + // A number is an instant in milliseconds and needs its own reading. + // Text is read a cell at a time, and over the years the engine + // reaches rather than the ones pandas parses into by default; see + // the helpers. if (declared.get(unit.attribute).contains(AttributeType.LONG)) s"""_texera_epoch_millis_to_timestamp(out1df[$colLit])""" - else s"""pd.to_datetime(out1df[$colLit], errors="coerce")""" + else s"""_texera_text_to_timestamp(out1df[$colLit])""" case _ => s"""out1df[$colLit]""" } lines += s"""out1df[$colLit] = $expr""" diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 2aa9341e7c9..608c58a42d2 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala @@ -239,6 +239,125 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { fromEngine shouldBe Seq("6.0", "7.25", "6", "7", "true", "false") } + // The moments where the two sides used to part. pandas parses into nanoseconds + // by default, which reach only 1677 to 2262, so everything past that edge was + // emptied where the engine holds a java.sql.Timestamp and reads it like any + // other moment. The first three rows are ordinary ones, and they are also three + // different formats in one column: the engine hands DateParserUtils a field at + // a time, so a row states its own format rather than the column's first one. + private val timestampCases = Seq( + "2024-03-05 14:09:07", + "2024-03-05T14:09:07", + "March 5, 2024", + "1677-09-22 00:12:44", + "2262-04-11 23:47:16", + "2500-01-01 00:00:00", + "1500-06-15 08:30:00", + "9999-12-31 23:59:59" + ) + + /** `java.sql.Timestamp.toString` always writes a fraction where Python writes + * one only when there is something to write, and every case here lands on a + * whole second. + */ + private def withoutFraction(text: String): String = text.stripSuffix(".0") + + /** The cells the driver printed, told apart from anything pandas wrote to + * stderr, which this process merges into the same stream. + */ + private def cellsOf(out: String): Seq[String] = + out.linesIterator.filter(_.startsWith("cell ")).map(_.drop("cell ".length)).toSeq + + it should "cast text to a timestamp the way AttributeTypeUtils does" in { + val python = resolvePython().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val op = new TypeCastingOpDesc + op.typeCastingUnits = List(castUnit("v", AttributeType.TIMESTAMP)) + + // One frame rather than a row at a time, unlike the casts that refuse a + // value: this one coerces, and the column is where a format that is not the + // first row's would be lost. + val values = timestampCases.map(v => "\"" + v + "\"").mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": $values}) + |${op.generateStandaloneCode()} + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-timestamp-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val process = + new ProcessBuilder(python, script.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 } + + // Only the printed cells: pandas writes a parsing warning to stderr, which + // this process merges into the same stream. + val fromScript = cellsOf(out) + val fromEngine = + timestampCases.map(v => withoutFraction(engineAnswer(v, AttributeType.TIMESTAMP))) + withClue(s"cases=${timestampCases.mkString(", ")}\nscript said $fromScript\n") { + fromScript shouldBe fromEngine + } + // The rows that made the issue: a moment either side of the nanosecond edge. + fromEngine.takeRight(3) shouldBe + Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59") + } + + // The one place the script is meant to differ, and the reason it cannot simply + // parse strictly: the engine accepts a set of formats no single pandas call + // states, so text neither can read leaves an empty cell instead of ending an + // exported run halfway. + it should "leave a cell it cannot read empty rather than refusing it" in { + val python = resolvePython().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val op = new TypeCastingOpDesc + op.typeCastingUnits = List(castUnit("v", AttributeType.TIMESTAMP)) + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": ["not a date", None, "2024-03-05 14:09:07"]}) + |${op.generateStandaloneCode()} + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-timestamp-unreadable-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val process = + new ProcessBuilder(python, script.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 + cellsOf(out) shouldBe Seq("null", "null", "2024-03-05 14:09:07") + } + // The engine refuses the same text, which is the difference this coercion is. + engineAnswer("not a date", AttributeType.TIMESTAMP) shouldBe "error" + } + // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path // (UDF_PYTHON_PATH), then python3 / python / py. private def resolvePython(): Option[String] = { From c074933878adc8d88c0abd0b060b594fc44f5b6e Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 17:36:54 -0700 Subject: [PATCH 2/6] fix(operator): read a timestamp that states an offset The cast stopped on one. A reading that carries a zone cannot be converted to a column that holds none, so `2024-03-05T14:09:07Z` ended the cast where the previous generator had read it, and an explicit offset did the same. The engine reads the offset and keeps no zone for it: DateParserUtils parses the reading and java.sql.Timestamp holds the wall clock of the machine's own zone, so `...T14:09:07Z` is 06:09:07 where the machine is eight hours behind UTC. The helper now does that, which is also what the epoch-milliseconds branch beside it already did. The two spellings join the cases the timestamp test compares against parseField cell by cell, so what they should read is taken from the engine rather than written down and the pair says the same thing in any zone. Co-Authored-By: Claude Opus 5 (1M context) --- .../amber/operator/StandaloneHelpers.scala | 16 +++++++++++++++- .../typecasting/TypeCastingOpDescSpec.scala | 15 +++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index e70477c1ff1..4ad2c01a3da 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -135,18 +135,32 @@ object StandaloneHelpers { | # Still coerced, which the strict cast is not: the engine accepts a set | # of formats no single pandas call states, so text neither can read is | # answered with an empty cell rather than by ending the run. + | # + | # A reading that states an offset is moved to the zone the machine is + | # set to and then holds that wall clock, which is what the engine does + | # with one: DateParserUtils reads the offset and java.sql.Timestamp + | # keeps no zone of its own. Left alone, the offset travels as far as + | # the conversion below and stops the cast on a value the run reads. + | # tzlocal() and not a fixed offset, so each instant gets the one in + | # force when it happened. | from dateutil.parser import parse as _parse_date + | from dateutil.tz import tzlocal | | if pd.api.types.is_datetime64_any_dtype(s): + | if getattr(s.dtype, "tz", None) is not None: + | s = s.dt.tz_convert(tzlocal()).dt.tz_localize(None) | return s.astype("datetime64[us]") | | def _one(x): | if pd.isna(x): | return None | try: - | return _parse_date(str(x).strip()) + | parsed = _parse_date(str(x).strip()) | except (ValueError, OverflowError): | return None + | if parsed.tzinfo is not None: + | parsed = parsed.astimezone(tzlocal()).replace(tzinfo=None) + | return parsed | | return s.map(_one).astype("datetime64[us]") | diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 608c58a42d2..6896dd25b5b 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala @@ -245,10 +245,16 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { // other moment. The first three rows are ordinary ones, and they are also three // different formats in one column: the engine hands DateParserUtils a field at // a time, so a row states its own format rather than the column's first one. + // Two of them state an offset, which DateParserUtils reads and java.sql.Timestamp + // then keeps no zone for: the moment is held as the wall clock of the machine's + // own zone. The expectation is taken from the engine rather than written down, + // so the pair says the same thing wherever the suite runs. private val timestampCases = Seq( "2024-03-05 14:09:07", "2024-03-05T14:09:07", "March 5, 2024", + "2024-03-05T14:09:07Z", + "2024-03-05T14:09:07+05:30", "1677-09-22 00:12:44", "2262-04-11 23:47:16", "2500-01-01 00:00:00", @@ -315,6 +321,15 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { // The rows that made the issue: a moment either side of the nanosecond edge. fromEngine.takeRight(3) shouldBe Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59") + // And the pair that states an offset, which reaches the same moment by two + // spellings: five and a half hours apart in the text, and so in the reading. + val zoned = fromScript.slice(3, 5) + java.time.Duration + .between( + java.time.LocalDateTime.parse(zoned(1).replace(' ', 'T')), + java.time.LocalDateTime.parse(zoned(0).replace(' ', 'T')) + ) + .toMinutes shouldBe 330 } // The one place the script is meant to differ, and the reason it cannot simply From ca67ce12b759b0b7093fc499c5839e9c88c149a1 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Mon, 21 Sep 2026 01:00:03 -0700 Subject: [PATCH 3/6] fix(operator): keep the resolution a timestamp column arrived in The cast narrowed one. A column the engine already holds as a moment went through the same astype as freshly parsed text, so 14:09:07.123456789 and 14:09:07.123456001 both came back as .123456 and two rows the run tells apart became one. The previous generator left such a column alone. parseField hands a java.sql.Timestamp back untouched and that class counts nanoseconds, so there is nothing for this branch to decide: the column is returned at the resolution it arrived in, zoned or not. The text branch below is unchanged and still reads into microseconds, which is what the years past the nanosecond edge need. A test casts a timestamp column to timestamp and compares the two values cell by cell against parseField, which is how the rest of the spec compares a cast. Co-Authored-By: Claude Opus 5 (1M context) --- .../amber/operator/StandaloneHelpers.scala | 15 +++--- .../typecasting/TypeCastingOpDescSpec.scala | 54 +++++++++++++++++++ 2 files changed, 63 insertions(+), 6 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index 4ad2c01a3da..d6ee4e9a913 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -127,10 +127,13 @@ object StandaloneHelpers { | # is handed one field at a time, so a row states its own format and the | # rest of the column has no say in it. | # - | # Held at microsecond resolution and not the nanoseconds pandas parses - | # into by default, which reach 1677 to 2262: the engine holds a - | # java.sql.Timestamp, where the year 2500 is an ordinary moment and - | # emptying it would answer for a row the run itself had no trouble with. + | # What is read here is held at microsecond resolution and not the + | # nanoseconds pandas parses into by default, which reach 1677 to 2262: + | # the engine holds a java.sql.Timestamp, where the year 2500 is an + | # ordinary moment and emptying it would answer for a row the run itself + | # had no trouble with. A column that is already a moment is handed back + | # at the resolution it arrived in instead, because parseField returns a + | # java.sql.Timestamp untouched and that class counts nanoseconds. | # | # Still coerced, which the strict cast is not: the engine accepts a set | # of formats no single pandas call states, so text neither can read is @@ -148,8 +151,8 @@ object StandaloneHelpers { | | if pd.api.types.is_datetime64_any_dtype(s): | if getattr(s.dtype, "tz", None) is not None: - | s = s.dt.tz_convert(tzlocal()).dt.tz_localize(None) - | return s.astype("datetime64[us]") + | return s.dt.tz_convert(tzlocal()).dt.tz_localize(None) + | return s | | def _one(x): | if pd.isna(x): diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 6896dd25b5b..155c73dd65f 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala @@ -332,6 +332,60 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { .toMinutes shouldBe 330 } + // A column that is already a moment is not text and is not re-read: parseField + // hands a java.sql.Timestamp back untouched, and that class counts nanoseconds, + // so narrowing the column here would fold two moments the run tells apart into + // one. The pair below differs only past the microsecond. + private val nanosecondCases = Seq( + "2024-03-05 14:09:07.123456789", + "2024-03-05 14:09:07.123456001" + ) + + it should "keep the resolution a timestamp column arrived in" in { + val python = resolvePython().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val op = new TypeCastingOpDesc + op.typeCastingUnits = List(castUnit("v", AttributeType.TIMESTAMP)) + // The declared type is what sends this down the branch under test: a column + // the engine already holds as a moment, cast to the type it already has. + val input = Schema().add(new Attribute("v", AttributeType.TIMESTAMP)) + val generated = op.generateStandaloneCode(Map(op.operatorInfo.inputPorts.head.id -> input)) + + val values = nanosecondCases.map(v => "\"" + v + "\"").mkString("[", ", ", "]") + val driver = + s"""import pandas as pd + | + |${op.standaloneHelpers().mkString("\n\n")} + | + | + |in1df = pd.DataFrame({"v": pd.to_datetime($values)}) + |$generated + | + |for answer in out1df["v"]: + | print("cell", "null" if pd.isna(answer) else str(answer)) + |""".stripMargin + + val script = Files.createTempFile("typecast-timestamp-nanos-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + + val process = + new ProcessBuilder(python, script.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 } + + val fromScript = cellsOf(out) + val fromEngine = + nanosecondCases.map(v => engineAnswer(java.sql.Timestamp.valueOf(v), AttributeType.TIMESTAMP)) + withClue(s"script said $fromScript\n") { fromScript shouldBe fromEngine } + // What the two answers have to carry: the rows stay apart. + fromEngine shouldBe nanosecondCases + } + // The one place the script is meant to differ, and the reason it cannot simply // parse strictly: the engine accepts a set of formats no single pandas call // states, so text neither can read leaves an empty cell instead of ending an From 189bf79a36bfd091a5a6091e736bf22d40f7874e Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 26 Sep 2026 20:09:35 -0700 Subject: [PATCH 4/6] fix(operator): cut a timestamp read from text to the millisecond, and write text as Java's toString does Co-Authored-By: Claude Opus 5.5 (1M context) --- .../amber/operator/StandaloneHelpers.scala | 55 +++++--- .../typecasting/TypeCastingOpDescSpec.scala | 120 +++++------------- 2 files changed, 66 insertions(+), 109 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index d6ee4e9a913..90f8123d36d 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -131,7 +131,9 @@ object StandaloneHelpers { | # nanoseconds pandas parses into by default, which reach 1677 to 2262: | # the engine holds a java.sql.Timestamp, where the year 2500 is an | # ordinary moment and emptying it would answer for a row the run itself - | # had no trouble with. A column that is already a moment is handed back + | # had no trouble with. Text is cut to the millisecond, because + | # DateParserUtils reads it into a java.util.Date, which counts no finer. + | # A column that is already a moment is handed back | # at the resolution it arrived in instead, because parseField returns a | # java.sql.Timestamp untouched and that class counts nanoseconds. | # @@ -163,7 +165,7 @@ object StandaloneHelpers { | return None | if parsed.tzinfo is not None: | parsed = parsed.astimezone(tzlocal()).replace(tzinfo=None) - | return parsed + | return parsed.replace(microsecond=parsed.microsecond // 1000 * 1000) | | return s.map(_one).astype("datetime64[us]") | @@ -190,6 +192,38 @@ object StandaloneHelpers { | return float(x) | | + |def _texera_java_text(x): + | # The text Java's toString gives the value Texera holds. A double past + | # 1e7 or under 1e-3 is written in E notation, where Python's str() + | # waits until 1e16 and 1e-4. A java.sql.Timestamp writes every digit of + | # its nanoseconds, trailing zeros dropped but never all of them, where + | # Python writes no fraction on a whole second and six digits otherwise. + | # A boolean is written in lower case. + | import datetime + | import decimal + | + | if pd.api.types.is_bool(x): + | return "true" if x else "false" + | if isinstance(x, datetime.datetime): + | nanos = x.microsecond * 1000 + getattr(x, "nanosecond", 0) + | return x.strftime("%Y-%m-%d %H:%M:%S") + "." + (f"{nanos:09d}".rstrip("0") or "0") + | if pd.api.types.is_integer(x): + | return str(int(x)) + | if pd.api.types.is_float(x): + | value = float(x) + | if value != value: + | return "NaN" + | if value in (float("inf"), float("-inf")): + | return "Infinity" if value > 0 else "-Infinity" + | if value == 0 or 1e-3 <= abs(value) < 1e7: + | return repr(value) + | sign, digits, exponent = decimal.Decimal(repr(value)).as_tuple() + | power = len(digits) + exponent - 1 + | kept = "".join(map(str, digits)).rstrip("0") or "0" + | return ("-" if sign else "") + kept[0] + "." + (kept[1:] or "0") + "E" + str(power) + | return str(x) + | + | |def _texera_cast_string(s): | # `toString` on the field, so the COLUMN's type decides the text and | # not the shape of the value: a double keeps its point, whether or not @@ -199,20 +233,5 @@ object StandaloneHelpers { | return s.map(lambda x: None if pd.isna(x) else ("true" if x else "false")) | if pd.api.types.is_integer_dtype(s): | return s.map(lambda x: None if pd.isna(x) else str(int(x))) - | if pd.api.types.is_datetime64_any_dtype(s): - | # java.sql.Timestamp.toString: trailing zeros dropped from the - | # fraction, but never all of them. Python's own str() writes no - | # fraction at all on a whole second and six digits otherwise. - | def _ts(x): - | if pd.isna(x): - | return None - | text = x.strftime("%Y-%m-%d %H:%M:%S.%f").rstrip("0") - | return text + "0" if text.endswith(".") else text - | - | return s.map(_ts) - | return s.map( - | lambda x: None - | if pd.isna(x) - | else ("true" if x else "false") if pd.api.types.is_bool(x) else str(x) - | )""".stripMargin + | return s.map(lambda x: None if pd.isna(x) else _texera_java_text(x))""".stripMargin } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala index 155c73dd65f..60e27ac2856 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDescSpec.scala @@ -25,7 +25,9 @@ import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, AttributeTy import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} 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 @@ -37,6 +39,8 @@ import scala.util.Try class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { + private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName) + private val workflowId = WorkflowIdentity(1L) private val executionId = ExecutionIdentity(1L) @@ -98,11 +102,10 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { tc.typeCastingUnits.head.resultType shouldBe AttributeType.STRING } - // The values a cast reads differently on the two sides. Python's own `bool` - // answers true for every non-empty string, so "false" and "0" are where the - // script used to disagree with the run it came from; text that is neither a - // boolean nor a number, and an empty cell, are the two ends of the range. - private val boolCases = Seq("true", "false", "0", "1", "not a boolean", null) + // Text that is neither a boolean nor a number, which the engine refuses. The + // verification only compares runs that both produced something, so it cannot + // ask this; the values both sides read are its `str_to_bool` column. + private val boolCases = Seq("not a boolean") /** What the engine answers, as the string the Python side prints back: the * literal, or `error` for a value `parseField` refuses. @@ -112,7 +115,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { .map(v => if (v == null) "null" else v.toString) .getOrElse("error") - it should "cast to boolean the way AttributeTypeUtils does" in { + it should "refuse a boolean the way AttributeTypeUtils does" taggedAs NeedsPythonPackages in { val python = resolvePython().getOrElse( cancel("No runnable python executable (udf.conf python.path, python3, python, py)") ) @@ -167,76 +170,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { withClue(s"cases=${boolCases.mkString(", ")}\nscript said $fromScript\n") { fromScript shouldBe fromEngine } - // The pairs that made the review: "false" is not true, and "0" is not true. - fromEngine shouldBe Seq("true", "false", "false", "true", "error", "null") - } - - // One column per source type, since what the text looks like follows the - // column and not the value: a whole double keeps its point, an integer never - // grows one, and a boolean is lower case. - private val stringColumns: Seq[(String, String, Seq[AnyRef])] = Seq( - ("dbl", "float64", Seq(Double.box(6.0), Double.box(7.25))), - ("int", "int64", Seq(Int.box(6), Int.box(7))), - ("flag", "bool", Seq(Boolean.box(true), Boolean.box(false))) - ) - - it should "cast to string the way AttributeTypeUtils does" in { - val python = resolvePython().getOrElse( - cancel("No runnable python executable (udf.conf python.path, python3, python, py)") - ) - if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") - - val op = new TypeCastingOpDesc - op.typeCastingUnits = stringColumns.map { - case (name, _, _) => castUnit(name, AttributeType.STRING) - }.toList - - val frame = stringColumns - .map { - case (name, dtype, values) => - val cells = values - .map { - case b: java.lang.Boolean => if (b) "True" else "False" - case other => other.toString - } - .mkString("[", ", ", "]") - s""" "$name": pd.Series($cells, dtype="$dtype"),""" - } - .mkString("\n") - val driver = - s"""import pandas as pd - | - |${op.standaloneHelpers().mkString("\n\n")} - | - | - |in1df = pd.DataFrame({ - |$frame - |}) - |${op.generateStandaloneCode()} - |for column in ${stringColumns.map(c => "\"" + c._1 + "\"").mkString("[", ", ", "]")}: - | for cell in out1df[column]: - | print("null" if pd.isna(cell) else cell) - |""".stripMargin - - val script = Files.createTempFile("typecast-string-", ".py") - script.toFile.deleteOnExit() - Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) - - val process = - new ProcessBuilder(python, script.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 - } - - val fromEngine = stringColumns.flatMap(_._3).map(v => engineAnswer(v, AttributeType.STRING)) - withClue(s"script said ${out.trim.linesIterator.toSeq}\n") { - out.trim.linesIterator.toSeq shouldBe fromEngine - } - // The pair that made the review: a double keeps its point at 6.0, where the - // integer 6 has none. - fromEngine shouldBe Seq("6.0", "7.25", "6", "7", "true", "false") + fromEngine shouldBe Seq("error") } // The moments where the two sides used to part. pandas parses into nanoseconds @@ -248,13 +182,17 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { // Two of them state an offset, which DateParserUtils reads and java.sql.Timestamp // then keeps no zone for: the moment is held as the wall clock of the machine's // own zone. The expectation is taken from the engine rather than written down, - // so the pair says the same thing wherever the suite runs. + // so the pair says the same thing wherever the suite runs. Two carry digits + // past the millisecond, which DateParserUtils reads into a java.util.Date and + // so drops. private val timestampCases = Seq( "2024-03-05 14:09:07", "2024-03-05T14:09:07", "March 5, 2024", "2024-03-05T14:09:07Z", "2024-03-05T14:09:07+05:30", + "2024-03-05 14:09:07.123456", + "2024-03-05 14:09:07.9999", "1677-09-22 00:12:44", "2262-04-11 23:47:16", "2500-01-01 00:00:00", @@ -262,11 +200,12 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { "9999-12-31 23:59:59" ) - /** `java.sql.Timestamp.toString` always writes a fraction where Python writes - * one only when there is something to write, and every case here lands on a - * whole second. + /** A printed moment as the wall clock it names. `java.sql.Timestamp.toString` + * trims its fraction and writes ".0" for none, where Python pads it to six + * digits or leaves it out. */ - private def withoutFraction(text: String): String = text.stripSuffix(".0") + private def moment(text: String): Any = + Try(java.time.LocalDateTime.parse(text.replace(' ', 'T'))).getOrElse(text) /** The cells the driver printed, told apart from anything pandas wrote to * stderr, which this process merges into the same stream. @@ -274,7 +213,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { private def cellsOf(out: String): Seq[String] = out.linesIterator.filter(_.startsWith("cell ")).map(_.drop("cell ".length)).toSeq - it should "cast text to a timestamp the way AttributeTypeUtils does" in { + it should "cast text to a timestamp the way AttributeTypeUtils does" taggedAs NeedsPythonPackages in { val python = resolvePython().getOrElse( cancel("No runnable python executable (udf.conf python.path, python3, python, py)") ) @@ -310,17 +249,16 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { process.waitFor(120, TimeUnit.SECONDS) withClue(s"python said:\n$out\nscript:\n$driver") { process.exitValue() shouldBe 0 } - // Only the printed cells: pandas writes a parsing warning to stderr, which - // this process merges into the same stream. + // Only the printed cells, so nothing else written to the merged stderr is + // read as one. val fromScript = cellsOf(out) - val fromEngine = - timestampCases.map(v => withoutFraction(engineAnswer(v, AttributeType.TIMESTAMP))) + val fromEngine = timestampCases.map(v => engineAnswer(v, AttributeType.TIMESTAMP)) withClue(s"cases=${timestampCases.mkString(", ")}\nscript said $fromScript\n") { - fromScript shouldBe fromEngine + fromScript.map(moment) shouldBe fromEngine.map(moment) } // The rows that made the issue: a moment either side of the nanosecond edge. - fromEngine.takeRight(3) shouldBe - Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59") + fromEngine.takeRight(3).map(moment) shouldBe + Seq("2500-01-01 00:00:00", "1500-06-15 08:30:00", "9999-12-31 23:59:59").map(moment) // And the pair that states an offset, which reaches the same moment by two // spellings: five and a half hours apart in the text, and so in the reading. val zoned = fromScript.slice(3, 5) @@ -341,7 +279,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { "2024-03-05 14:09:07.123456001" ) - it should "keep the resolution a timestamp column arrived in" in { + it should "keep the resolution a timestamp column arrived in" taggedAs NeedsPythonPackages in { val python = resolvePython().getOrElse( cancel("No runnable python executable (udf.conf python.path, python3, python, py)") ) @@ -390,7 +328,7 @@ class TypeCastingOpDescSpec extends AnyFlatSpec with Matchers { // parse strictly: the engine accepts a set of formats no single pandas call // states, so text neither can read leaves an empty cell instead of ending an // exported run halfway. - it should "leave a cell it cannot read empty rather than refusing it" in { + it should "leave a cell it cannot read empty rather than refusing it" taggedAs NeedsPythonPackages in { val python = resolvePython().getOrElse( cancel("No runnable python executable (udf.conf python.path, python3, python, py)") ) From 65f662052d8fe56fabf02bcf016fc8a119116997 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Mon, 12 Oct 2026 15:10:05 -0700 Subject: [PATCH 5/6] docs(operator): note the JDK version a cast double's text follows The helper writes a double's shortest digits, as Double.toString does from JDK 19. On JDK 17 the engine sometimes writes more digits for the same value, so say so where the text is built. Co-Authored-By: Claude Opus 5.5 --- .../org/apache/texera/amber/operator/StandaloneHelpers.scala | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index 90f8123d36d..fa862042468 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -199,6 +199,11 @@ object StandaloneHelpers { | # its nanoseconds, trailing zeros dropped but never all of them, where | # Python writes no fraction on a whole second and six digits otherwise. | # A boolean is written in lower case. + | # + | # A double's digits are the shortest that read back as the same value, + | # as Double.toString writes them from JDK 19. An earlier JDK sometimes + | # writes more: 9.999999999999999E22 for 1e23. The value is the same but + | # the text is not. | import datetime | import decimal | From 2b1dcf5c984f4e62dd8ad889bcb41c32be416d0c Mon Sep 17 00:00:00 2001 From: kary zheng Date: Mon, 12 Oct 2026 15:12:32 -0700 Subject: [PATCH 6/6] docs(operator): fold the double-digits note into the toString comment Co-Authored-By: Claude Opus 5.5 --- .../texera/amber/operator/StandaloneHelpers.scala | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala index fa862042468..b8f7036bd1a 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala @@ -195,15 +195,12 @@ object StandaloneHelpers { |def _texera_java_text(x): | # The text Java's toString gives the value Texera holds. A double past | # 1e7 or under 1e-3 is written in E notation, where Python's str() - | # waits until 1e16 and 1e-4. A java.sql.Timestamp writes every digit of - | # its nanoseconds, trailing zeros dropped but never all of them, where - | # Python writes no fraction on a whole second and six digits otherwise. - | # A boolean is written in lower case. - | # - | # A double's digits are the shortest that read back as the same value, - | # as Double.toString writes them from JDK 19. An earlier JDK sometimes - | # writes more: 9.999999999999999E22 for 1e23. The value is the same but - | # the text is not. + | # waits until 1e16 and 1e-4. Its digits are the shortest that read back + | # as the same value, as from JDK 19; an earlier JDK sometimes writes + | # more, 9.999999999999999E22 for 1e23. A java.sql.Timestamp writes every + | # digit of its nanoseconds, trailing zeros dropped but never all of them, + | # where Python writes no fraction on a whole second and six digits + | # otherwise. A boolean is written in lower case. | import datetime | import decimal |