From 45d52d424467af69a90fd7fa282582a5402f10bf Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 12:54:24 -0700 Subject: [PATCH 1/3] 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/3] 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/3] 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