You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/09/02 00:50:25 UTC
(texera) branch gh-readonly-queue/main/pr-7672-fc7cfcd9e14a68b43484a4d9ba923715076798f5 created (now a26c12c8da)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7672-fc7cfcd9e14a68b43484a4d9ba923715076798f5
in repository https://gitbox.apache.org/repos/asf/texera.git
at a26c12c8da feat(arrow): store and read a timestamp as the wall clock it is (#7672)
This branch includes the following new commits:
new a26c12c8da feat(arrow): store and read a timestamp as the wall clock it is (#7672)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: feat(arrow): store and read a timestamp as the wall clock it is (#7672)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7672-fc7cfcd9e14a68b43484a4d9ba923715076798f5
in repository https://gitbox.apache.org/repos/asf/texera.git
commit a26c12c8da8c7f795f67f7bfb2e5a1f947fd15b8
Author: Kary Zheng <15...@users.noreply.github.com>
AuthorDate: Wed Sep 2 00:50:03 2026 +0000
feat(arrow): store and read a timestamp as the wall clock it is (#7672)
### What changes were proposed in this PR?
Every other source hands the engine a wall clock. CSV and JSONL carry no
zone at all, so what the file states is what the operator gets, and
`TypeCasting` and the time-series plot both parse with a plain
`pd.to_datetime`. A Texera TIMESTAMP has no zone to record one either.
Arrow was the exception: its timestamp fields state how their own
numbers are to be read, and `ArrowUtils` ignored that and went through
the value's own epoch, which reads the wall clock in the JVM's zone and
puts the machine's setting where the file cannot record it.
So one file read out different values on servers in different places,
with nothing to account for the difference:
| Server timezone | Value produced, before | After |
| --- | --- | --- |
| America/Los_Angeles | 2024-01-07 00:00:00 | 2024-01-07 00:00:00 |
| UTC | 2024-01-07 08:00:00 | 2024-01-07 00:00:00 |
| Asia/Tokyo | 2024-01-07 17:00:00 | 2024-01-07 00:00:00 |
| Europe/Berlin | 2024-01-07 09:00:00 | 2024-01-07 00:00:00 |
Writing was the same mismatch from the other end: the number stored was
the local instant of the wall clock while the label beside it said UTC,
so every reader other than a Texera in that same zone saw the value
moved. Reading that file with `pandas.read_feather` reported 08:00
before and reports 00:00 now, which is what Texera itself shows.
Both directions now go through what the field itself states, which puts
Arrow on the same footing as the rest: what the file states is what the
engine gets, wherever it runs. A field states two things and both
matter. One is the zone. Every field this writes is labelled UTC, but an
`.arrow` file handed to the source operator may name any zone, and the
wall clock to hand over is the one that zone gives. The other is the
unit. Arrow hands a zoned vector its number unscaled, so the field's
unit is what says how far from the epoch that number reaches. A tz-aware
pandas column arrives as nanoseconds, and taken for milliseconds it
landed in 1970. Unlabelled columns, which is what an ordinary `.arrow`
file written by pandas holds, still hand back a `LocalDateTime`, already
the wall clock itself.
One thing for anyone holding existing files: bytes written before this
are read by the new rule, so a timestamp in them shifts by the offset of
the zone that wrote it.
### Any related issues, documentation, discussions?
Closes #7666
### How was this PR tested?
Two `ArrowUtilsSpec`s cover it. In `workflow-operator`, one assertion
moved with the change: it stated the identity between the stored long
and the value's own epoch, which is machine-dependent, and now states
the wall clock explicitly. The round-trip case needed no change,
preserving the wall clock being the property both before and after.
In `workflow-core`, six cases cover the timestamp fields
`fromTexeraSchema` never produces, since every root built from it is UTC
milliseconds. Three read: a zoned millisecond field, whose number is one
reading in the zone it names and another five hours off in UTC; a zoned
nanosecond field, which taken for milliseconds would land in 1970; and
an unlabelled field, which hands back the wall clock already. Two write,
one per axis, pinning the number that reaches the vector. The last sends
a wall clock out and back through five combinations of unit and zone, an
unlabelled one among them.
Both specs were also run in four JVMs started with different
`-Duser.timezone` (America/Los_Angeles, UTC, Asia/Tokyo, Europe/Berlin)
and pass identically in each, which is the property the table above
states.
`workflow-core` (765 tests) and `workflow-operator` (2214) pass, as do
the Python-channel specs that share this conversion,
`PythonProxyServerSpec` and `PythonProxyClientSpec`.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
---
.../org/apache/texera/amber/util/ArrowUtils.scala | 82 ++++++++++++++--
.../apache/texera/amber/util/ArrowUtilsSpec.scala | 108 ++++++++++++++++++++-
.../apache/texera/amber/util/ArrowUtilsSpec.scala | 10 +-
3 files changed, 190 insertions(+), 10 deletions(-)
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala
index af14ae9acd..434519b3de 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala
@@ -24,7 +24,7 @@ import org.apache.texera.amber.core.tuple.AttributeTypeUtils.AttributeTypeExcept
import org.apache.texera.amber.core.tuple._
import org.apache.arrow.memory.{BufferAllocator, RootAllocator}
import org.apache.arrow.vector.types.FloatingPointPrecision
-import org.apache.arrow.vector.types.TimeUnit.MILLISECOND
+import org.apache.arrow.vector.types.TimeUnit
import org.apache.arrow.vector.types.pojo.ArrowType.PrimitiveType
import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType}
import org.apache.arrow.vector.{
@@ -40,6 +40,9 @@ import org.apache.arrow.vector.{
}
import java.nio.charset.StandardCharsets
+import java.sql.Timestamp
+import java.time.temporal.ChronoUnit
+import java.time.{Instant, LocalDateTime, ZoneId, ZoneOffset}
import java.util
import scala.jdk.CollectionConverters.CollectionHasAsScala
import scala.language.implicitConversions
@@ -81,7 +84,12 @@ object ArrowUtils extends LazyLogging {
// Use the attribute type from the schema (which includes metadata)
// instead of deriving it from the Arrow type
val attributeType = schema.getAttributes(index).getType
- AttributeTypeUtils.parseField(value, attributeType)
+ // A timestamp is the one type whose field says more than the
+ // schema does, so it is the only one that reads the field.
+ attributeType match {
+ case AttributeType.TIMESTAMP => wallClockOf(value, fieldVector.getField.getType)
+ case _ => AttributeTypeUtils.parseField(value, attributeType)
+ }
} catch {
case e: Exception =>
logger.warn("Caught error during parsing Arrow value back to Texera value", e)
@@ -93,6 +101,54 @@ object ArrowUtils extends LazyLogging {
.build()
}
+ /** The wall clock a timestamp column holds, read the way its own field states.
+ *
+ * A Texera TIMESTAMP carries no zone, so the wall clock is the whole of what
+ * crosses over, and the field says how to arrive at one: the zone its numbers
+ * are counted in, and the unit they are counted in. A zoned vector hands back
+ * that number bare, and letting `new Timestamp(number)` make the wall clock
+ * would take it for milliseconds in the JVM's zone: one file would then say
+ * different things on servers in different places, with nothing in the file to
+ * account for the difference. A zoneless vector hands back a LocalDateTime
+ * already, which is the wall clock itself.
+ */
+ private def wallClockOf(value: AnyRef, arrowType: ArrowType): Timestamp =
+ (value, arrowType) match {
+ case (null, _) => null
+ case (ldt: LocalDateTime, _) => Timestamp.valueOf(ldt)
+ case (number: java.lang.Long, field: ArrowType.Timestamp) =>
+ Timestamp.valueOf(
+ LocalDateTime.ofInstant(instantOf(number, field.getUnit), zoneOf(field))
+ )
+ case (other, _) => AttributeTypeUtils.parseTimestamp(other)
+ }
+
+ /** The zone a timestamp field counts its numbers in. An unlabelled field counts
+ * from the epoch with no zone in the picture, which is what UTC arithmetic is.
+ */
+ private def zoneOf(field: ArrowType.Timestamp): ZoneId =
+ Option(field.getTimezone).map(ZoneId.of).getOrElse(ZoneOffset.UTC)
+
+ /** The instant a zoned vector's bare number stands for. Arrow leaves those
+ * unscaled, so the field's unit is what says how far from the epoch it reaches.
+ */
+ private def instantOf(number: Long, unit: TimeUnit): Instant =
+ unit match {
+ case TimeUnit.SECOND => Instant.ofEpochSecond(number)
+ case TimeUnit.MILLISECOND => Instant.ofEpochMilli(number)
+ case TimeUnit.MICROSECOND => Instant.EPOCH.plus(number, ChronoUnit.MICROS)
+ case TimeUnit.NANOSECOND => Instant.EPOCH.plusNanos(number)
+ }
+
+ /** The number a field of this unit records an instant as, inverting [[instantOf]]. */
+ private def numberOf(instant: Instant, unit: TimeUnit): Long =
+ unit match {
+ case TimeUnit.SECOND => instant.getEpochSecond
+ case TimeUnit.MILLISECOND => instant.toEpochMilli
+ case TimeUnit.MICROSECOND => ChronoUnit.MICROS.between(Instant.EPOCH, instant)
+ case TimeUnit.NANOSECOND => ChronoUnit.NANOS.between(Instant.EPOCH, instant)
+ }
+
/**
* Converts an Arrow Schema into Texera Schema.
* Checks field metadata to recover types that share an Arrow representation
@@ -213,7 +269,13 @@ object ArrowUtils extends LazyLogging {
.asInstanceOf[Float8Vector]
.setSafe(index, !isNull, if (isNull) 0 else value.asInstanceOf[Double])
- case _: ArrowType.Timestamp =>
+ // The wall clock written as the field's own zone and unit, so the number
+ // and the label beside it agree. Going through the value's own epoch
+ // would have read the wall clock in the JVM's zone instead, putting a
+ // machine's setting into the file: the same table written in two places
+ // would hold two different instants under one label. Inverts
+ // [[wallClockOf]].
+ case timestamp: ArrowType.Timestamp =>
vector
.asInstanceOf[TimeStampVector]
.setSafe(
@@ -221,9 +283,15 @@ object ArrowUtils extends LazyLogging {
!isNull,
if (isNull) 0L
else
- AttributeTypeUtils
- .parseField(value, AttributeType.LONG)
- .asInstanceOf[Long]
+ numberOf(
+ AttributeTypeUtils
+ .parseField(value, AttributeType.TIMESTAMP)
+ .asInstanceOf[Timestamp]
+ .toLocalDateTime
+ .atZone(zoneOf(timestamp))
+ .toInstant,
+ timestamp.getUnit
+ )
)
case _: ArrowType.Utf8 =>
@@ -299,7 +367,7 @@ object ArrowUtils extends LazyLogging {
ArrowType.Bool.INSTANCE
case AttributeType.TIMESTAMP =>
- new ArrowType.Timestamp(MILLISECOND, "UTC")
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, "UTC")
case AttributeType.BINARY =>
new ArrowType.Binary
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
index c91f09540f..e3f8014462 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
@@ -20,7 +20,7 @@
package org.apache.texera.amber.util
import org.apache.arrow.memory.RootAllocator
-import org.apache.arrow.vector.{VarCharVector, VectorSchemaRoot}
+import org.apache.arrow.vector.{TimeStampVector, VarCharVector, VectorSchemaRoot}
import org.apache.arrow.vector.types.{FloatingPointPrecision, TimeUnit}
import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType}
import org.apache.texera.amber.core.state.State
@@ -436,6 +436,112 @@ class ArrowUtilsSpec extends AnyFlatSpec with Matchers {
}
}
+ // ----- Timestamp fields that are not the UTC millisecond ones we write -----
+
+ // fromTexeraSchema only ever writes Timestamp(MILLISECOND, "UTC"), so the
+ // roots built above never exercise another zone or unit. An .arrow file handed
+ // to ArrowSourceOpDesc can carry either: pandas writes a tz-aware column as
+ // Timestamp(NANOSECOND, <its zone>). These build the field directly.
+ private def withFieldRoot(field: Field)(body: VectorSchemaRoot => Unit): Unit = {
+ val allocator = new RootAllocator()
+ val root = VectorSchemaRoot.create(
+ new org.apache.arrow.vector.types.pojo.Schema(util.Arrays.asList(field)),
+ allocator
+ )
+ try {
+ root.allocateNew()
+ body(root)
+ } finally {
+ root.close()
+ allocator.close()
+ }
+ }
+
+ private val timestampSchema = Schema(List(new Attribute("t", AttributeType.TIMESTAMP)))
+
+ private def timestampTuple(wallClock: String): Tuple =
+ Tuple
+ .builder(timestampSchema)
+ .addSequentially(Array[Any](Timestamp.valueOf(wallClock)))
+ .build()
+
+ "getTexeraTuple" should "read a zoned timestamp as the wall clock its field's zone gives" in {
+ // 1704603600000 is 2024-01-07 00:00 in New York and 05:00 in UTC. The field
+ // says New York, so New York is the reading that comes back.
+ val field =
+ arrowField("t", new ArrowType.Timestamp(TimeUnit.MILLISECOND, "America/New_York"))
+ withFieldRoot(field) { root =>
+ root.getVector(0).asInstanceOf[TimeStampVector].setSafe(0, 1704603600000L)
+ root.setRowCount(1)
+ ArrowUtils.getTexeraTuple(0, root).getField[Timestamp]("t") shouldBe
+ Timestamp.valueOf("2024-01-07 00:00:00")
+ }
+ }
+
+ it should "count a zoned timestamp in the unit its field declares" in {
+ // Arrow hands a zoned vector its number unscaled, so the same reading is a
+ // million times the number in a nanosecond field that it is in a millisecond
+ // one. Taken for milliseconds, this one would land in 1970.
+ val field =
+ arrowField("t", new ArrowType.Timestamp(TimeUnit.NANOSECOND, "America/New_York"))
+ withFieldRoot(field) { root =>
+ root.getVector(0).asInstanceOf[TimeStampVector].setSafe(0, 1704603600000000000L)
+ root.setRowCount(1)
+ ArrowUtils.getTexeraTuple(0, root).getField[Timestamp]("t") shouldBe
+ Timestamp.valueOf("2024-01-07 00:00:00")
+ }
+ }
+
+ it should "read an unlabelled timestamp as the wall clock it already is" in {
+ // No zone to reconcile, and Arrow scales the number itself: the vector hands
+ // back a LocalDateTime, which is the wall clock, whatever the server's zone.
+ val field = arrowField("t", new ArrowType.Timestamp(TimeUnit.MILLISECOND, null))
+ withFieldRoot(field) { root =>
+ root.getVector(0).asInstanceOf[TimeStampVector].setSafe(0, 1704603600000L)
+ root.setRowCount(1)
+ ArrowUtils.getTexeraTuple(0, root).getField[Timestamp]("t") shouldBe
+ Timestamp.valueOf("2024-01-07 05:00:00")
+ }
+ }
+
+ "setTexeraTuple" should "store a wall clock as the number its field's zone calls for" in {
+ val field =
+ arrowField("t", new ArrowType.Timestamp(TimeUnit.MILLISECOND, "America/New_York"))
+ withFieldRoot(field) { root =>
+ ArrowUtils.setTexeraTuple(timestampTuple("2024-01-07 00:00:00"), 0, root)
+ // The New York instant of that reading. Stored as the UTC one it would be
+ // 1704585600000, five hours off what the field's own label promises.
+ root.getVector(0).asInstanceOf[TimeStampVector].get(0) shouldBe 1704603600000L
+ }
+ }
+
+ it should "store a wall clock in the unit its field declares" in {
+ val field = arrowField("t", new ArrowType.Timestamp(TimeUnit.NANOSECOND, "America/New_York"))
+ withFieldRoot(field) { root =>
+ ArrowUtils.setTexeraTuple(timestampTuple("2024-01-07 00:00:00"), 0, root)
+ root.getVector(0).asInstanceOf[TimeStampVector].get(0) shouldBe 1704603600000000000L
+ }
+ }
+
+ "a timestamp round-trip" should "preserve the wall clock through a field of any zone and unit" in {
+ val fields = List(
+ new ArrowType.Timestamp(TimeUnit.SECOND, "Asia/Tokyo"),
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, "America/New_York"),
+ new ArrowType.Timestamp(TimeUnit.MICROSECOND, "Europe/Berlin"),
+ new ArrowType.Timestamp(TimeUnit.NANOSECOND, "UTC"),
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, null)
+ )
+ fields.foreach { arrowType =>
+ withFieldRoot(arrowField("t", arrowType)) { root =>
+ ArrowUtils.setTexeraTuple(timestampTuple("2024-01-07 09:30:00"), 0, root)
+ withClue(s"$arrowType: ") {
+ ArrowUtils.getTexeraTuple(0, root).getField[Timestamp]("t") shouldBe
+ Timestamp.valueOf("2024-01-07 09:30:00")
+ }
+ }
+ }
+ }
+
// ----- fromAttributeType (null input) -----
"fromAttributeType" should "reject a null attribute type" in {
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
index 6a97df8ca3..94abd91b8a 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala
@@ -149,7 +149,11 @@ class ArrowUtilsSpec extends AnyFlatSpec {
Long.box(1L),
Boolean.box(true),
Double.box(1.1),
- new Timestamp(10000L),
+ // Stated as a wall clock, which is what a Texera TIMESTAMP holds. Built
+ // from an epoch instead, the wall clock would be whichever one the
+ // machine's zone gives that instant, and what gets stored below would
+ // move with it.
+ Timestamp.valueOf("1970-01-01 00:00:10"),
"hello world"
)
)
@@ -169,7 +173,9 @@ class ArrowUtilsSpec extends AnyFlatSpec {
assert(vectorSchemaRoot.getVector(2).getObject(index).asInstanceOf[Boolean] == true)
assert(vectorSchemaRoot.getVector(3).getObject(index).asInstanceOf[Double] == 1.1)
- // the arrow storage type of timestamp is Long
+ // The arrow storage type of timestamp is Long, and the field is labelled
+ // UTC, so the wall clock above is stored as the UTC instant of the same
+ // reading: ten seconds past the epoch, on a server anywhere.
assert(vectorSchemaRoot.getVector(4).getObject(index).asInstanceOf[Long] == 10000L)
// the arrow storage type of string is Text