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-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 825b7bbe [Fix] Adapt Arrow timestamp conversion for DATETIME and
TIMESTAMPTZ (#676)
825b7bbe is described below
commit 825b7bbe9647d54398eed91870219f322c617967
Author: wudi <[email protected]>
AuthorDate: Wed Jul 22 12:44:13 2026 +0800
[Fix] Adapt Arrow timestamp conversion for DATETIME and TIMESTAMPTZ (#676)
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/flink/serialization/RowBatch.java | 59 +++++++----
.../doris/flink/serialization/TestRowBatch.java | 111 ++++++++++++++++++---
2 files changed, 140 insertions(+), 30 deletions(-)
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
index 21a42614..65c6b6b1 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
@@ -42,12 +42,13 @@ import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.complex.MapVector;
import org.apache.arrow.vector.complex.StructVector;
import org.apache.arrow.vector.complex.impl.DateDayReaderImpl;
-import org.apache.arrow.vector.complex.impl.TimeStampMicroReaderImpl;
import org.apache.arrow.vector.complex.impl.UnionMapReader;
import org.apache.arrow.vector.complex.reader.FieldReader;
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.doris.flink.exception.DorisException;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.models.Schema;
@@ -424,6 +425,7 @@ public class RowBatch {
}
break;
case "DATETIMEV2":
+ case "TIMESTAMPTZ":
if (minorType.equals(MinorType.VARCHAR)) {
VarCharVector varCharVector = (VarCharVector) fieldVector;
if (varCharVector.isNull(rowIndex)) {
@@ -441,7 +443,8 @@ public class RowBatch {
addValueToRow(rowIndex, dateTime);
} else {
logger.error(
- "Unsupported type for DATETIMEV2, minorType {},
class is {}",
+ "Unsupported type for {}, minorType {}, class is
{}",
+ currentType,
minorType.name(),
fieldVector == null ? null :
fieldVector.getClass());
return false;
@@ -572,8 +575,13 @@ public class RowBatch {
}
private Object handleMapFieldReader(FieldReader reader) {
- if (reader instanceof TimeStampMicroReaderImpl) {
- return longToLocalDateTime(reader.readLong());
+ ArrowType fieldType = reader.getField().getType();
+ if (fieldType instanceof ArrowType.Timestamp) {
+ ArrowType.Timestamp timestampType = (ArrowType.Timestamp)
fieldType;
+ if (timestampType.getTimezone() == null) {
+ return reader.readObject();
+ }
+ return longToLocalDateTime(reader.readLong(),
timestampType.getUnit(), DEFAULT_ZONE_ID);
}
if (reader instanceof DateDayReaderImpl) {
return LocalDate.ofEpochDay(((Integer)
reader.readObject()).longValue());
@@ -587,24 +595,41 @@ public class RowBatch {
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();
+ String timezone = timestampType.getTimezone();
+ if (timezone == null) {
+ // datetime type does not carry timezone, it represents UTC time.
+ return (LocalDateTime) vector.getObject(rowIndex);
+ }
+ return longToLocalDateTime(vector.get(rowIndex),
timestampType.getUnit(), DEFAULT_ZONE_ID);
}
@VisibleForTesting
- 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);
}
/**
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/TestRowBatch.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/TestRowBatch.java
index a740b2ad..6adc9b9f 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/TestRowBatch.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/TestRowBatch.java
@@ -34,8 +34,11 @@ 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.TimeStampMilliVector;
+import org.apache.arrow.vector.TimeStampSecTZVector;
import org.apache.arrow.vector.TimeStampSecVector;
import org.apache.arrow.vector.TinyIntVector;
import org.apache.arrow.vector.UInt4Vector;
@@ -80,6 +83,7 @@ import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
@@ -841,7 +845,7 @@ public class TestRowBatch {
vector.setValueCount(3);
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");
@@ -849,11 +853,11 @@ public class TestRowBatch {
datetimeV2Vector.setInitialCapacity(3);
datetimeV2Vector.allocateNew();
datetimeV2Vector.setIndexDefined(0);
- datetimeV2Vector.setSafe(0, second);
+ datetimeV2Vector.setSafe(0, second * 1_000_000L);
datetimeV2Vector.setIndexDefined(1);
- datetimeV2Vector.setSafe(1, second * 1000 + nano / 1000000);
+ datetimeV2Vector.setSafe(1, second * 1_000_000L + nano / 1_000_000 *
1_000L);
datetimeV2Vector.setIndexDefined(2);
- datetimeV2Vector.setSafe(2, second * 1000000 + nano / 1000);
+ datetimeV2Vector.setSafe(2, second * 1_000_000L + nano / 1_000L);
vector.setValueCount(3);
arrowStreamWriter.writeBatch();
@@ -1721,9 +1725,12 @@ public class TestRowBatch {
now.toInstant(defaultZoneId.getRules().getOffset(now)).getEpochSecond() *
1_000_000
+ now.getNano() / 1_000;
- LocalDateTime dateTime1 =
RowBatch.longToLocalDateTime(secondTimestamp);
- LocalDateTime dateTime2 = RowBatch.longToLocalDateTime(milliTimestamp);
- LocalDateTime dateTime3 = RowBatch.longToLocalDateTime(microTimestamp);
+ LocalDateTime dateTime1 =
+ RowBatch.longToLocalDateTime(secondTimestamp, TimeUnit.SECOND,
defaultZoneId);
+ LocalDateTime dateTime2 =
+ RowBatch.longToLocalDateTime(milliTimestamp,
TimeUnit.MILLISECOND, defaultZoneId);
+ LocalDateTime dateTime3 =
+ RowBatch.longToLocalDateTime(microTimestamp,
TimeUnit.MICROSECOND, defaultZoneId);
long result1 =
dateTime1.atZone(defaultZoneId).toInstant().getEpochSecond();
long result2 =
dateTime2.atZone(defaultZoneId).toInstant().toEpochMilli();
@@ -1769,6 +1776,7 @@ public class TestRowBatch {
arrowStreamWriter.start();
root.setRowCount(1);
+ // no tz vector is utc timestamp
FieldVector vector = root.getVector("k0");
TimeStampMicroVector mircoVec = (TimeStampMicroVector) vector;
mircoVec.allocateNew(1);
@@ -1811,23 +1819,100 @@ public class TestRowBatch {
RowBatch rowBatch = new RowBatch(scanBatchResult, schema).readArrow();
List<Object> next = rowBatch.next();
Assert.assertEquals(next.size(), 3);
+ Assert.assertEquals(LocalDateTime.of(2024, 7, 25, 7, 22, 23,
586123000), next.get(0));
+ Assert.assertEquals(LocalDateTime.of(2024, 7, 25, 7, 22, 23,
586000000), next.get(1));
+ Assert.assertEquals(LocalDateTime.of(2024, 7, 25, 7, 22, 23, 0),
next.get(2));
+ }
+
+ @Test
+ public void timestampTZVector() throws IOException, DorisException {
+ List<Field> childrenBuilder = new ArrayList<>();
+ childrenBuilder.add(
+ new Field(
+ "k0",
+ FieldType.nullable(new
ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC+8")),
+ null));
+ childrenBuilder.add(
+ new Field(
+ "k1",
+ FieldType.nullable(new
ArrowType.Timestamp(TimeUnit.MILLISECOND, "UTC+8")),
+ null));
+ childrenBuilder.add(
+ 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(childrenBuilder, null),
+ new RootAllocator(Integer.MAX_VALUE));
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ ArrowStreamWriter arrowStreamWriter =
+ new ArrowStreamWriter(
+ root, new DictionaryProvider.MapDictionaryProvider(),
outputStream);
+
+ arrowStreamWriter.start();
+ root.setRowCount(1);
+
+ FieldVector vector = root.getVector("k0");
+ TimeStampMicroTZVector microVector = (TimeStampMicroTZVector) vector;
+ microVector.allocateNew(1);
+ microVector.setIndexDefined(0);
+ microVector.setSafe(0, 1721892143586123L);
+ vector.setValueCount(1);
+
+ vector = root.getVector("k1");
+ TimeStampMilliTZVector milliVector = (TimeStampMilliTZVector) vector;
+ milliVector.allocateNew(1);
+ milliVector.setIndexDefined(0);
+ milliVector.setSafe(0, 1721892143586L);
+ vector.setValueCount(1);
+
+ vector = root.getVector("k2");
+ TimeStampSecTZVector secVector = (TimeStampSecTZVector) vector;
+ secVector.allocateNew(1);
+ secVector.setIndexDefined(0);
+ secVector.setSafe(0, 1721892143L);
+ vector.setValueCount(1);
+
+ arrowStreamWriter.writeBatch();
+ arrowStreamWriter.end();
+ arrowStreamWriter.close();
+
+ TStatus status = new TStatus();
+ status.setStatusCode(TStatusCode.OK);
+ TScanBatchResult scanBatchResult = new TScanBatchResult();
+ scanBatchResult.setStatus(status);
+ scanBatchResult.setEos(false);
+ scanBatchResult.setRows(outputStream.toByteArray());
+
+ String schemaStr =
+
"{\"properties\":[{\"type\":\"TIMESTAMPTZ\",\"name\":\"k0\",\"comment\":\"\"},
{\"type\":\"TIMESTAMPTZ\",\"name\":\"k1\",\"comment\":\"\"},
{\"type\":\"TIMESTAMPTZ\",\"name\":\"k2\",\"comment\":\"\"}],"
+ + "\"status\":200}";
+
+ Schema schema = RestService.parseSchema(schemaStr, logger);
+
+ RowBatch rowBatch = new RowBatch(scanBatchResult, schema).readArrow();
+ List<Object> next = rowBatch.next();
+ Assert.assertEquals(3, next.size());
Assert.assertEquals(
- next.get(0),
LocalDateTime.of(2024, 7, 25, 15, 22, 23, 586123000)
.atZone(ZoneId.of("UTC+8"))
.withZoneSameInstant(ZoneId.systemDefault())
- .toLocalDateTime());
+ .toLocalDateTime(),
+ next.get(0));
Assert.assertEquals(
- next.get(1),
LocalDateTime.of(2024, 7, 25, 15, 22, 23, 586000000)
.atZone(ZoneId.of("UTC+8"))
.withZoneSameInstant(ZoneId.systemDefault())
- .toLocalDateTime());
+ .toLocalDateTime(),
+ next.get(1));
Assert.assertEquals(
- next.get(2),
LocalDateTime.of(2024, 7, 25, 15, 22, 23, 0)
.atZone(ZoneId.of("UTC+8"))
.withZoneSameInstant(ZoneId.systemDefault())
- .toLocalDateTime());
+ .toLocalDateTime(),
+ next.get(2));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
