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