From c46b3c35e9d3b24aa62cecc12f47c13d1e25d649 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 14 Aug 2026 16:52:49 -0700 Subject: [PATCH] feat(arrow): store and read a timestamp as the wall clock it is 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 a Texera TIMESTAMP has no zone to record one either. Arrow was the exception: its fields carry a zone, and both directions reconciled that 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. Reading was the visible half. A zoned vector hands back epoch milliseconds, and `new Timestamp(millis)` renders those in the local zone, so one file read out 00:00 in Los Angeles, 08:00 in UTC and 17:00 in Tokyo. Nothing in the file accounted for the difference and nothing reported it. 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. Both now go through the label the field already carries, which puts Arrow on the same footing as the rest: what the file states is what the engine gets, wherever it runs. Zoneless columns are untouched, those handing back a LocalDateTime that is already the wall clock itself. The spec asserted the old identity, that the stored long equals the value's own epoch. It states the wall clock explicitly now, so what it expects no longer depends on where it runs. Note for anyone with 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. Closes #7666 Co-Authored-By: Claude Opus 5 (1M context) --- .../apache/texera/amber/util/ArrowUtils.scala | 36 +++++++++++++++++-- .../texera/amber/util/ArrowUtilsSpec.scala | 10 ++++-- 2 files changed, 41 insertions(+), 5 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 af14ae9acd0..ebcef128a88 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 @@ -40,6 +40,8 @@ import org.apache.arrow.vector.{ } import java.nio.charset.StandardCharsets +import java.sql.Timestamp +import java.time.{Instant, LocalDateTime, ZoneOffset} import java.util import scala.jdk.CollectionConverters.CollectionHasAsScala import scala.language.implicitConversions @@ -81,7 +83,8 @@ 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) + if (attributeType == AttributeType.TIMESTAMP) wallClockOf(value) + else AttributeTypeUtils.parseField(value, attributeType) } catch { case e: Exception => logger.warn("Caught error during parsing Arrow value back to Texera value", e) @@ -93,6 +96,25 @@ object ArrowUtils extends LazyLogging { .build() } + /** The wall clock a timestamp column holds, read as UTC. + * + * A Texera TIMESTAMP has no zone of its own, and the fields this writes are + * labelled UTC, so UTC is what the number beside the label means. A zoned + * vector hands back epoch milliseconds, and letting `new Timestamp(millis)` + * turn those into a wall clock would read them 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): Timestamp = + value match { + case null => null + case ldt: LocalDateTime => Timestamp.valueOf(ldt) + case millis: java.lang.Long => + Timestamp.valueOf(LocalDateTime.ofInstant(Instant.ofEpochMilli(millis), ZoneOffset.UTC)) + case other => AttributeTypeUtils.parseTimestamp(other) + } + /** * Converts an Arrow Schema into Texera Schema. * Checks field metadata to recover types that share an Arrow representation @@ -213,6 +235,11 @@ object ArrowUtils extends LazyLogging { .asInstanceOf[Float8Vector] .setSafe(index, !isNull, if (isNull) 0 else value.asInstanceOf[Double]) + // The wall clock written AS UTC, the label the field carries, so the + // number and the label 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 UTC label. Mirrors [[wallClockOf]]. case _: ArrowType.Timestamp => vector .asInstanceOf[TimeStampVector] @@ -222,8 +249,11 @@ object ArrowUtils extends LazyLogging { if (isNull) 0L else AttributeTypeUtils - .parseField(value, AttributeType.LONG) - .asInstanceOf[Long] + .parseField(value, AttributeType.TIMESTAMP) + .asInstanceOf[Timestamp] + .toLocalDateTime + .toInstant(ZoneOffset.UTC) + .toEpochMilli ) case _: ArrowType.Utf8 => 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 6a97df8ca3e..94abd91b8ae 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