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 3a7c98b [Fix] Recover DATETIME wall clock from Arrow timezone (#377)
3a7c98b is described below
commit 3a7c98b416c5ca7982421405c42f49a602d9b662
Author: Zack <[email protected]>
AuthorDate: Wed Sep 2 22:24:17 2026 -0400
[Fix] Recover DATETIME wall clock from Arrow timezone (#377)
Closes #376.
Doris DATETIME and DATETIMEV2 store wall-clock values without timezone
semantics. At times, Doris BE nodes can return them through timezone-aware
Arrow vectors.
RowBatch treated these values like TIMESTAMPTZ instants. When the Arrow
timezone differed from the Spark JVM timezone, the displayed clock shifted.
For example, Doris 2026-08-24 10:00:09 could become 2026-08-24 02:00:09
when Spark ran in UTC but the BE configured tz was Shanghai/UTC+8.
#366 supports Arrow timestamps without timezone metadata. It does not
recover wall-clock values when older backends include this metadata.
---
.../apache/doris/spark/client/read/RowBatch.java | 93 ++++++++++-----
.../doris/spark/client/read/RowBatchTest.java | 128 +++++++++++++++++++++
.../doris/spark/sql/Doris2DorisE2ECase.scala | 8 +-
.../apache/doris/spark/sql/DorisReaderITCase.scala | 18 +--
4 files changed, 200 insertions(+), 47 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 a9d30a5..3d56054 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
@@ -71,11 +71,13 @@ import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
+import java.util.TreeMap;
/**
* row batch data container.
@@ -83,6 +85,9 @@ import java.util.Objects;
public class RowBatch implements Serializable {
private static final Logger logger =
LoggerFactory.getLogger(RowBatch.class);
private static final ZoneId DEFAULT_ZONE_ID = ZoneId.systemDefault();
+ private static final String DORIS_DEFAULT_TIME_ZONE = "Asia/Shanghai";
+ private static final String UTC_TIME_ZONE = "UTC";
+ private static final Map<String, String> DORIS_TIME_ZONE_ALIASES =
buildDorisTimeZoneAliases();
private static final DateTimeFormatter DATE_TIME_FORMATTER = new
DateTimeFormatterBuilder()
.appendPattern("yyyy-MM-dd HH:mm:ss")
@@ -424,42 +429,30 @@ public class RowBatch implements Serializable {
break;
case "DATETIME":
case "DATETIMEV2":
- case "TIMESTAMPTZ":
-
if (mt.equals(MinorType.VARCHAR)) {
- VarCharVector varCharVector = (VarCharVector)
curFieldVector;
+ convertVarcharDateTime((VarCharVector)
curFieldVector);
+ } else if (curFieldVector instanceof TimeStampVector) {
+ TimeStampVector timeStampVector =
(TimeStampVector) curFieldVector;
for (int rowIndex = 0; rowIndex <
rowCountInOneBatch; rowIndex++) {
- if (varCharVector.isNull(rowIndex)) {
- addValueToRow(rowIndex, null);
- continue;
- }
- String stringValue = completeMilliseconds(new
String(varCharVector.get(rowIndex),
- StandardCharsets.UTF_8));
- LocalDateTime dateTime =
LocalDateTime.parse(stringValue, dateTimeV2Formatter);
- if (datetimeJava8ApiEnabled) {
- Instant instant =
dateTime.atZone(DEFAULT_ZONE_ID).toInstant();
- addValueToRow(rowIndex, instant);
- } else {
- addValueToRow(rowIndex,
Timestamp.valueOf(dateTime));
- }
+ addDateTimeValue(rowIndex,
getDateTime(rowIndex, timeStampVector));
}
+ } else {
+ String errMsg = String.format("Unsupported type
for DATETIMEV2, minorType %s, class is %s",
+ mt.name(), curFieldVector.getClass());
+ throw new
java.lang.IllegalArgumentException(errMsg);
+ }
+ break;
+ case "TIMESTAMPTZ":
+ if (mt.equals(MinorType.VARCHAR)) {
+ convertVarcharDateTime((VarCharVector)
curFieldVector);
} else if (curFieldVector instanceof TimeStampVector) {
TimeStampVector timeStampVector =
(TimeStampVector) curFieldVector;
for (int rowIndex = 0; rowIndex <
rowCountInOneBatch; rowIndex++) {
- if (timeStampVector.isNull(rowIndex)) {
- addValueToRow(rowIndex, null);
- continue;
- }
- LocalDateTime dateTime = getDateTime(rowIndex,
timeStampVector);
- if (datetimeJava8ApiEnabled) {
- Instant instant =
dateTime.atZone(DEFAULT_ZONE_ID).toInstant();
- addValueToRow(rowIndex, instant);
- } else {
- addValueToRow(rowIndex,
Timestamp.valueOf(dateTime));
- }
+ addDateTimeValue(rowIndex,
+ getDateTime(rowIndex, timeStampVector,
DEFAULT_ZONE_ID));
}
} else {
- String errMsg = String.format("Unsupported type
for DATETIMEV2, minorType %s, class is %s",
+ String errMsg = String.format("Unsupported type
for TIMESTAMPTZ, minorType %s, class is %s",
mt.name(), curFieldVector.getClass());
throw new
java.lang.IllegalArgumentException(errMsg);
}
@@ -597,7 +590,35 @@ public class RowBatch implements Serializable {
}
}
+ private void convertVarcharDateTime(VarCharVector varCharVector) {
+ for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
+ if (varCharVector.isNull(rowIndex)) {
+ addDateTimeValue(rowIndex, null);
+ continue;
+ }
+ String stringValue = completeMilliseconds(new
String(varCharVector.get(rowIndex),
+ StandardCharsets.UTF_8));
+ addDateTimeValue(rowIndex, LocalDateTime.parse(stringValue,
dateTimeV2Formatter));
+ }
+ }
+
+ private void addDateTimeValue(int rowIndex, LocalDateTime dateTime) {
+ if (dateTime == null) {
+ addValueToRow(rowIndex, null);
+ return;
+ }
+ if (datetimeJava8ApiEnabled) {
+ addValueToRow(rowIndex,
dateTime.atZone(DEFAULT_ZONE_ID).toInstant());
+ } else {
+ addValueToRow(rowIndex, Timestamp.valueOf(dateTime));
+ }
+ }
+
public LocalDateTime getDateTime(int rowIndex, FieldVector fieldVector) {
+ return getDateTime(rowIndex, fieldVector, null);
+ }
+
+ public LocalDateTime getDateTime(int rowIndex, FieldVector fieldVector,
ZoneId zoneIdOverride) {
TimeStampVector vector = (TimeStampVector) fieldVector;
if (vector.isNull(rowIndex)) {
return null;
@@ -606,7 +627,21 @@ public class RowBatch implements Serializable {
if (timestampType.getTimezone() == null) {
return (LocalDateTime) vector.getObject(rowIndex);
}
- return longToLocalDateTime(vector.get(rowIndex),
timestampType.getUnit(), DEFAULT_ZONE_ID);
+ ZoneId zoneId = zoneIdOverride == null
+ ? ZoneId.of(timestampType.getTimezone(),
DORIS_TIME_ZONE_ALIASES)
+ : zoneIdOverride;
+ return longToLocalDateTime(vector.get(rowIndex),
timestampType.getUnit(), zoneId);
+ }
+
+ private static Map<String, String> buildDorisTimeZoneAliases() {
+ Map<String, String> aliases = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ aliases.putAll(ZoneId.SHORT_IDS);
+ // Keep aliases consistent with Doris TimeUtils.timeZoneAliasMap.
+ aliases.put("CST", DORIS_DEFAULT_TIME_ZONE);
+ aliases.put("PRC", DORIS_DEFAULT_TIME_ZONE);
+ aliases.put(UTC_TIME_ZONE, UTC_TIME_ZONE);
+ aliases.put("GMT", UTC_TIME_ZONE);
+ return Collections.unmodifiableMap(aliases);
}
public static String completeMilliseconds(String stringValue) {
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 1d2b5bb..b115206 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
@@ -1328,6 +1328,134 @@ public class RowBatchTest {
Assert.assertEquals(secInstant, instantRow.get(2));
}
+ @Test
+ public void testDatetimeTzVector() throws IOException, DorisException {
+ ImmutableList<Field> fields = ImmutableList.of(
+ new Field("k0", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MICROSECOND,
"+08:00")), null),
+ new Field("k1", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, "CST")),
null),
+ new Field("k2", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.SECOND, "EST")),
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\":\"DATETIME\",\"name\":\"k0\",\"comment\":\"\"},"
+ +
"{\"type\":\"DATETIMEV2\",\"name\":\"k1\",\"comment\":\"\"},"
+ +
"{\"type\":\"DATETIMEV2\",\"name\":\"k2\",\"comment\":\"\"}],\"status\":200}",
+ Schema.class);
+ Instant microInstant = Instant.ofEpochSecond(1721892143L, 586123000L);
+ Instant milliInstant = Instant.ofEpochSecond(1721892143L, 586000000L);
+ Instant secInstant = Instant.ofEpochSecond(1721892143L);
+ ZoneId arrowTz = ZoneId.of("+08:00");
+ Timestamp microTimestamp =
Timestamp.valueOf(LocalDateTime.ofInstant(microInstant, arrowTz));
+ Timestamp milliTimestamp = Timestamp.valueOf(
+ LocalDateTime.ofInstant(milliInstant,
ZoneId.of("Asia/Shanghai")));
+ Timestamp secTimestamp = Timestamp.valueOf(
+ LocalDateTime.ofInstant(secInstant, ZoneId.of("EST",
ZoneId.SHORT_IDS)));
+
+ List<Object> timestampRow = new RowBatch(result, schema, false).next();
+ Assert.assertEquals(microTimestamp, timestampRow.get(0));
+ Assert.assertEquals(milliTimestamp, timestampRow.get(1));
+ Assert.assertEquals(secTimestamp, timestampRow.get(2));
+
+ List<Object> instantRow = new RowBatch(result, schema, true).next();
+
Assert.assertEquals(microTimestamp.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant(),
+ instantRow.get(0));
+
Assert.assertEquals(milliTimestamp.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant(),
+ instantRow.get(1));
+
Assert.assertEquals(secTimestamp.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant(),
+ instantRow.get(2));
+ }
+
+ @Test
+ public void testTimestampTzAndDatetimeSameArrowBytes() throws IOException,
DorisException {
+ Instant milliInstant = Instant.ofEpochMilli(1721892143586L);
+ ZoneOffset systemOffset =
ZoneId.systemDefault().getRules().getOffset(milliInstant);
+ String arrowTimezone = ZoneOffset.ofHours(8).equals(systemOffset) ?
"+00:00" : "+08:00";
+ ZoneId arrowTz = ZoneId.of(arrowTimezone);
+ ImmutableList<Field> fields = ImmutableList.of(
+ new Field("k0", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND,
arrowTimezone)), null),
+ new Field("k1", FieldType.nullable(
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND,
arrowTimezone)), 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);
+ TimeStampMilliTZVector timestamptzVector = (TimeStampMilliTZVector)
root.getVector("k0");
+ timestamptzVector.allocateNew(1);
+ timestamptzVector.setSafe(0, 1721892143586L);
+ timestamptzVector.setValueCount(1);
+ TimeStampMilliTZVector datetimeVector = (TimeStampMilliTZVector)
root.getVector("k1");
+ datetimeVector.allocateNew(1);
+ datetimeVector.setSafe(0, 1721892143586L);
+ datetimeVector.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\":\"DATETIME\",\"name\":\"k1\",\"comment\":\"\"}],\"status\":200}",
+ Schema.class);
+ Timestamp datetimeTimestamp = Timestamp.valueOf(
+ LocalDateTime.ofInstant(milliInstant, arrowTz));
+
+ List<Object> timestampRow = new RowBatch(result, schema, false).next();
+ Assert.assertNotEquals(systemOffset,
arrowTz.getRules().getOffset(milliInstant));
+ Assert.assertEquals(Timestamp.from(milliInstant), timestampRow.get(0));
+ Assert.assertEquals(datetimeTimestamp, timestampRow.get(1));
+ Assert.assertNotEquals(timestampRow.get(0), timestampRow.get(1));
+
+ List<Object> instantRow = new RowBatch(result, schema, true).next();
+ Assert.assertEquals(milliInstant, instantRow.get(0));
+
Assert.assertEquals(datetimeTimestamp.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant(),
+ instantRow.get(1));
+ }
+
@Test
public void testLongToLocalDateTimeUsesDeclaredUnit() {
ZoneId utc = ZoneId.of("UTC");
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
index 76ecccf..26c9941 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
@@ -97,11 +97,9 @@ class Doris2DorisE2ECase(readMode: String, flightSqlPort:
Int) extends AbstractC
|""".stripMargin)
session.stop()
- // TODO: Remove the legacy Scanner expectations after Doris returns
DATETIME as a
- // timezone-naive Arrow timestamp.
- val datetime1 = if (readMode == "arrow") "2025-03-11T12:34:56" else
"2025-03-11T04:34:56"
- val datetime2 = if (readMode == "arrow") "2024-12-25T23:59:59" else
"2024-12-25T15:59:59"
- val datetime3 = if (readMode == "arrow") "2023-06-15T08:00" else
"2023-06-15T00:00"
+ val datetime1 = "2025-03-11T12:34:56"
+ val datetime2 = "2024-12-25T23:59:59"
+ val datetime3 = "2023-06-15T08:00"
val excepted =
util.Arrays.asList(
"1,true,127,32767,2147483647,9223372036854775807,170141183460469231731687303715884105727,3.14,2.71828,12345.6789,2025-03-11,"
+ datetime1 + ",A,Hello, Doris!,This is a string,[\"Alice\",
\"Bob\"],{\"key1\":\"value1\", \"key2\":\"value2\"},{\"name\":\"Tom\",
\"age\":30},{\"key\":\"value\"},{\"data\":123,\"type\":\"variant\"}",
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
index 5653dfe..86bceae 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
@@ -235,9 +235,9 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select * from test_source order by id
|""".stripMargin).collect()
- val expectedTimestamp1 = if (readMode == "thrift") "2025-03-11 04:34:56"
else "2025-03-11 12:34:56"
- val expectedTimestamp2 = if (readMode == "thrift") "2024-12-25 15:59:59"
else "2024-12-25 23:59:59"
- val expectedTimestamp3 = if (readMode == "thrift") "2023-06-15 00:00:00"
else "2023-06-15 08:00:00"
+ val expectedTimestamp1 = "2025-03-11 12:34:56"
+ val expectedTimestamp2 = "2024-12-25 23:59:59"
+ val expectedTimestamp3 = "2023-06-15 08:00:00"
val expectedData = Array(
Row(1, true, 127, 32767, 2147483647, 9223372036854775807L,
"170141183460469231731687303715884105727",
@@ -416,11 +416,7 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select id,c10,c11 from test_source where c10 = '2025-03-11' and c13
like 'Hello%'
|""".stripMargin).collect()
- val expectedDateFilter = if (readMode == "thrift") {
- "List([1,2025-03-11,2025-03-11 04:34:56.0])"
- } else {
- "List([1,2025-03-11,2025-03-11 12:34:56.0])"
- }
+ val expectedDateFilter = "List([1,2025-03-11,2025-03-11 12:34:56.0])"
assert(expectedDateFilter.equals(dateFilter.toList.toString()))
val datetimeFilter = session.sql(
@@ -428,11 +424,7 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select id,c11,c12 from test_source where c10 < '2025-03-11' and c11
= '2024-12-25 23:59:59'
|""".stripMargin).collect()
- val expectedDatetimeFilter = if (readMode == "thrift") {
- "List([2,2024-12-25 15:59:59.0,B])"
- } else {
- "List([2,2024-12-25 23:59:59.0,B])"
- }
+ val expectedDatetimeFilter = "List([2,2024-12-25 23:59:59.0,B])"
assert(expectedDatetimeFilter.equals(datetimeFilter.toList.toString()))
val stringFilter = session.sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
