This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-spark-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 278f1ff [Fix] Adapt Arrow timestamp conversion for DATETIME and
TIMESTAMPTZ (#366)
278f1ff is described below
commit 278f1ff51e9a95ba0c9cfdf813dfddfe3933f561
Author: wudi <[email protected]>
AuthorDate: Mon Jul 27 16:18:15 2026 +0800
[Fix] Adapt Arrow timestamp conversion for DATETIME and TIMESTAMPTZ (#366)
Background:
PR apache/doris#38215 added timezone support to datetime. Versions prior to
this PR hardcoded the use of TimeStampMicroVector, whereas subsequent versions
return TimeStampTZVector. Consequently, to ensure compatibility, the connector
inferred the time unit based on the timestamp itself.
Changes:
Since datetime is inherently timezone-agnostic, PR apache/doris#65823
removed the timezone from datetime, causing it to return a native timestamp;
timestamptz continues to return TimeStampTZVector.
Connector Adaptation:
No conversion is performed if there is no timezone; the existing default
conversion logic is applied if a timezone is present.
Impact:
Reading datetime data may result in errors on versions prior to PR
apache/doris#38215.
---
.../apache/doris/spark/client/read/RowBatch.java | 42 +++++++---
.../apache/doris/spark/util/SchemaConvertors.scala | 1 +
.../doris/spark/client/read/RowBatchTest.java | 96 ++++++++++++++++++++--
.../doris/spark/util/SchemaConvertorsTest.scala | 6 ++
4 files changed, 126 insertions(+), 19 deletions(-)
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/RowBatch.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/RowBatch.java
index fe5c59f..a9d30a5 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/RowBatch.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/RowBatch.java
@@ -47,7 +47,9 @@ import org.apache.arrow.vector.complex.StructVector;
import org.apache.arrow.vector.complex.impl.UnionMapReader;
import org.apache.arrow.vector.ipc.ArrowReader;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.spark.sql.types.Decimal;
import org.slf4j.Logger;
@@ -175,17 +177,29 @@ public class RowBatch implements Serializable {
readRowCount += root.getRowCount();
}
- public static LocalDateTime longToLocalDateTime(long time) {
+ public static LocalDateTime longToLocalDateTime(long time, TimeUnit
timeUnit, ZoneId zoneId) {
Instant instant;
- // Determine the timestamp accuracy and process it
- if (time < 10_000_000_000L) { // Second timestamp
- instant = Instant.ofEpochSecond(time);
- } else if (time < 10_000_000_000_000L) { // milli second
- instant = Instant.ofEpochMilli(time);
- } else { // micro second
- instant = Instant.ofEpochSecond(time / 1_000_000, (time %
1_000_000) * 1_000);
+ switch (timeUnit) {
+ case SECOND:
+ instant = Instant.ofEpochSecond(time);
+ break;
+ case MILLISECOND:
+ instant = Instant.ofEpochMilli(time);
+ break;
+ case MICROSECOND:
+ instant = Instant.ofEpochSecond(
+ Math.floorDiv(time, 1_000_000L),
+ Math.floorMod(time, 1_000_000L) * 1_000L);
+ break;
+ case NANOSECOND:
+ instant = Instant.ofEpochSecond(
+ Math.floorDiv(time, 1_000_000_000L),
+ Math.floorMod(time, 1_000_000_000L));
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported timestamp
unit: " + timeUnit);
}
- return LocalDateTime.ofInstant(instant, DEFAULT_ZONE_ID);
+ return LocalDateTime.ofInstant(instant, zoneId);
}
public boolean hasNext() {
@@ -410,6 +424,7 @@ public class RowBatch implements Serializable {
break;
case "DATETIME":
case "DATETIMEV2":
+ case "TIMESTAMPTZ":
if (mt.equals(MinorType.VARCHAR)) {
VarCharVector varCharVector = (VarCharVector)
curFieldVector;
@@ -587,10 +602,11 @@ public class RowBatch implements Serializable {
if (vector.isNull(rowIndex)) {
return null;
}
- // todo: Currently, the scale of doris's arrow datetimev2 is hardcoded
to 6,
- // and there is also a time zone problem in arrow, so use timestamp to
convert first
- long time = vector.get(rowIndex);
- return longToLocalDateTime(time);
+ ArrowType.Timestamp timestampType = (ArrowType.Timestamp)
vector.getField().getType();
+ if (timestampType.getTimezone() == null) {
+ return (LocalDateTime) vector.getObject(rowIndex);
+ }
+ return longToLocalDateTime(vector.get(rowIndex),
timestampType.getUnit(), DEFAULT_ZONE_ID);
}
public static String completeMilliseconds(String stringValue) {
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/SchemaConvertors.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/SchemaConvertors.scala
index 8285064..8ccb003 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/SchemaConvertors.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/SchemaConvertors.scala
@@ -39,6 +39,7 @@ object SchemaConvertors {
case "DATEV2" => DataTypes.DateType
case "DATETIME" => DataTypes.TimestampType
case "DATETIMEV2" => DataTypes.TimestampType
+ case "TIMESTAMPTZ" => DataTypes.TimestampType
case "BINARY" => DataTypes.BinaryType
case "DECIMAL" => DecimalType(precision, scale)
case "CHAR" => DataTypes.StringType
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/RowBatchTest.java
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/RowBatchTest.java
index 09f2e07..1d2b5bb 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/RowBatchTest.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/RowBatchTest.java
@@ -38,7 +38,10 @@ import org.apache.arrow.vector.Float4Vector;
import org.apache.arrow.vector.Float8Vector;
import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampMilliTZVector;
+import org.apache.arrow.vector.TimeStampSecTZVector;
import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.UInt4Vector;
import org.apache.arrow.vector.VarBinaryVector;
@@ -74,9 +77,11 @@ import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.sql.Date;
import java.sql.Timestamp;
+import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.List;
import java.util.NoSuchElementException;
@@ -793,7 +798,7 @@ public class RowBatchTest {
LocalDateTime localDateTime = LocalDateTime.of(2024, 3, 20,
0, 0, 0, 123456000);
- long second =
localDateTime.atZone(ZoneId.systemDefault()).toEpochSecond();
+ long second = localDateTime.toEpochSecond(ZoneOffset.UTC);
int nano = localDateTime.getNano();
vector = root.getVector("k2");
@@ -801,9 +806,9 @@ public class RowBatchTest {
datetimeV2Vector.setInitialCapacity(3);
datetimeV2Vector.allocateNew();
datetimeV2Vector.setIndexDefined(0);
- datetimeV2Vector.setSafe(0, second);
+ datetimeV2Vector.setSafe(0, second * 1000000);
datetimeV2Vector.setIndexDefined(1);
- datetimeV2Vector.setSafe(1, second * 1000 + nano / 1000000);
+ datetimeV2Vector.setSafe(1, second * 1000000 + nano / 1000000 * 1000);
datetimeV2Vector.setIndexDefined(2);
datetimeV2Vector.setSafe(2, second * 1000000 + nano / 1000);
vector.setValueCount(3);
@@ -1192,7 +1197,7 @@ public class RowBatchTest {
LocalDateTime localDateTime = LocalDateTime.of(2025, 2, 24,
0, 0, 0, 123000000);
- long second =
localDateTime.atZone(ZoneId.systemDefault()).toEpochSecond();
+ long second = localDateTime.toEpochSecond(ZoneOffset.UTC);
int nano = localDateTime.getNano();
vector = root.getVector("k2");
@@ -1205,7 +1210,7 @@ public class RowBatchTest {
LocalDateTime localDateTime1 = LocalDateTime.of(2025, 2, 24,
1, 2, 3, 123456000);
- long second1 =
localDateTime1.atZone(ZoneId.systemDefault()).toEpochSecond();
+ long second1 = localDateTime1.toEpochSecond(ZoneOffset.UTC);
int nano1 = localDateTime1.getNano();
vector = root.getVector("k3");
@@ -1262,4 +1267,83 @@ public class RowBatchTest {
}
-}
\ No newline at end of file
+ @Test
+ public void testTimestampTzVector() throws IOException, DorisException {
+ ImmutableList<Field> fields = ImmutableList.of(
+ new Field("k0", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MICROSECOND,
"UTC+8")), null),
+ new Field("k1", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND,
"UTC+8")), null),
+ new Field("k2", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.SECOND, "UTC+8")),
null));
+ VectorSchemaRoot root = VectorSchemaRoot.create(
+ new org.apache.arrow.vector.types.pojo.Schema(fields, null),
+ new RootAllocator(Integer.MAX_VALUE));
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ ArrowStreamWriter writer = new ArrowStreamWriter(
+ root, new DictionaryProvider.MapDictionaryProvider(),
outputStream);
+
+ writer.start();
+ root.setRowCount(1);
+ TimeStampMicroTZVector microVector = (TimeStampMicroTZVector)
root.getVector("k0");
+ microVector.allocateNew(1);
+ microVector.setSafe(0, 1721892143586123L);
+ microVector.setValueCount(1);
+ TimeStampMilliTZVector milliVector = (TimeStampMilliTZVector)
root.getVector("k1");
+ milliVector.allocateNew(1);
+ milliVector.setSafe(0, 1721892143586L);
+ milliVector.setValueCount(1);
+ TimeStampSecTZVector secVector = (TimeStampSecTZVector)
root.getVector("k2");
+ secVector.allocateNew(1);
+ secVector.setSafe(0, 1721892143L);
+ secVector.setValueCount(1);
+ writer.writeBatch();
+ writer.end();
+ writer.close();
+
+ TStatus status = new TStatus();
+ status.setStatusCode(TStatusCode.OK);
+ TScanBatchResult result = new TScanBatchResult();
+ result.setStatus(status);
+ result.setEos(false);
+ result.setRows(outputStream.toByteArray());
+ Schema schema = MAPPER.readValue(
+ "{\"properties\":["
+ +
"{\"type\":\"TIMESTAMPTZ\",\"name\":\"k0\",\"comment\":\"\"},"
+ +
"{\"type\":\"TIMESTAMPTZ\",\"name\":\"k1\",\"comment\":\"\"},"
+ +
"{\"type\":\"TIMESTAMPTZ\",\"name\":\"k2\",\"comment\":\"\"}],\"status\":200}",
+ Schema.class);
+ Instant microInstant = Instant.ofEpochSecond(1721892143L, 586123000L);
+ Instant milliInstant = Instant.ofEpochSecond(1721892143L, 586000000L);
+ Instant secInstant = Instant.ofEpochSecond(1721892143L);
+
+ List<Object> timestampRow = new RowBatch(result, schema, false).next();
+ Assert.assertEquals(Timestamp.from(microInstant), timestampRow.get(0));
+ Assert.assertEquals(Timestamp.from(milliInstant), timestampRow.get(1));
+ Assert.assertEquals(Timestamp.from(secInstant), timestampRow.get(2));
+
+ List<Object> instantRow = new RowBatch(result, schema, true).next();
+ Assert.assertEquals(microInstant, instantRow.get(0));
+ Assert.assertEquals(milliInstant, instantRow.get(1));
+ Assert.assertEquals(secInstant, instantRow.get(2));
+ }
+
+ @Test
+ public void testLongToLocalDateTimeUsesDeclaredUnit() {
+ ZoneId utc = ZoneId.of("UTC");
+
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59),
+ RowBatch.longToLocalDateTime(-1L, TimeUnit.SECOND, utc));
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999000000),
+ RowBatch.longToLocalDateTime(-1L, TimeUnit.MILLISECOND, utc));
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999000),
+ RowBatch.longToLocalDateTime(-1L, TimeUnit.MICROSECOND, utc));
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999999),
+ RowBatch.longToLocalDateTime(-1L, TimeUnit.NANOSECOND, utc));
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/SchemaConvertorsTest.scala
b/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/SchemaConvertorsTest.scala
index f014b5f..6f7947c 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/SchemaConvertorsTest.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/SchemaConvertorsTest.scala
@@ -20,6 +20,7 @@ package org.apache.doris.spark.util
import org.apache.spark.sql.types.{ArrayType, DataTypes, DecimalType, MapType}
import org.junit.Assert
+import org.junit.{Test => JUnit4Test}
import org.junit.jupiter.api.Test
class SchemaConvertorsTest {
@@ -69,4 +70,9 @@ class SchemaConvertorsTest {
Assert.assertEquals(SchemaConvertors.toCatalystType("ARRAY", -1, -1,
false), DataTypes.StringType)
}
+ @JUnit4Test
+ def timestampTzToCatalystTypeTest(): Unit = {
+ Assert.assertEquals(DataTypes.TimestampType,
SchemaConvertors.toCatalystType("TIMESTAMPTZ", -1, -1))
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
