This is an automated email from the ASF dual-hosted git repository.
nzw921rx pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 0145209931 [Fix][Connector-V2] Fix Arrow microsecond timestamp with
timezone conversion (#12616)
0145209931 is described below
commit 0145209931a4fe6795bbcf64a827f0f75302a624
Author: nankeChen75 <[email protected]>
AuthorDate: Mon Oct 26 21:15:29 2026 +0800
[Fix][Connector-V2] Fix Arrow microsecond timestamp with timezone
conversion (#12616)
Co-authored-by: Xiaonan Chen <[email protected]>
---
.../arrow/converter/TimeStampMicroTZConverter.java | 61 ++++++++++++++++++
.../arrow/converter/TimeStampNanoTZConverter.java | 59 ++++++++++++++++++
.../arrow/ArrowToSeatunnelRowReaderTest.java | 72 +++++++++++++++++++++-
.../source/arrow/SeaTunnelDataTypeHolder.java | 2 +
4 files changed, 193 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampMicroTZConverter.java
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampMicroTZConverter.java
new file mode 100644
index 0000000000..6f4342e6ba
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampMicroTZConverter.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package
org.apache.seatunnel.connectors.seatunnel.common.source.arrow.converter;
+
+import
org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.seatunnel.shade.org.apache.arrow.vector.types.Types;
+
+import com.google.auto.service.AutoService;
+
+import java.time.Instant;
+import java.time.ZoneId;
+
+/**
+ * Converts Arrow {@code TIMESTAMPMICROTZ} values from epoch microseconds to
{@link
+ * java.time.LocalDateTime} using the time zone declared by the Arrow field.
+ *
+ * <p>{@link Math#floorDiv(long, long)} and {@link Math#floorMod(long, long)}
ensure that timestamps
+ * before the Unix epoch are converted correctly.
+ */
+@AutoService(Converter.class)
+public class TimeStampMicroTZConverter implements
Converter<TimeStampMicroTZVector> {
+ @Override
+ public Object convert(int rowIndex, TimeStampMicroTZVector fieldVector) {
+ if (fieldVector == null || fieldVector.isNull(rowIndex)) {
+ return null;
+ }
+
+ long epochMicro = fieldVector.getObject(rowIndex);
+ long epochSecond = Math.floorDiv(epochMicro, 1_000_000L);
+ long microOfSecond = Math.floorMod(epochMicro, 1_000_000L);
+
+ String timeZone = fieldVector.getTimeZone();
+ ZoneId zoneId =
+ timeZone == null || timeZone.isEmpty()
+ ? ZoneId.systemDefault()
+ : ZoneId.of(timeZone);
+
+ return Instant.ofEpochSecond(epochSecond, microOfSecond * 1_000L)
+ .atZone(zoneId)
+ .toLocalDateTime();
+ }
+
+ @Override
+ public boolean support(Types.MinorType type) {
+ return Types.MinorType.TIMESTAMPMICROTZ == type;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampNanoTZConverter.java
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampNanoTZConverter.java
new file mode 100644
index 0000000000..2d4a925e4a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/converter/TimeStampNanoTZConverter.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package
org.apache.seatunnel.connectors.seatunnel.common.source.arrow.converter;
+
+import
org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.seatunnel.shade.org.apache.arrow.vector.types.Types;
+
+import com.google.auto.service.AutoService;
+
+import java.time.Instant;
+import java.time.ZoneId;
+
+/**
+ * Converts Arrow {@code TIMESTAMPNANOTZ} values from epoch nanoseconds to
{@link
+ * java.time.LocalDateTime} using the time zone declared by the Arrow field.
+ *
+ * <p>{@link Math#floorDiv(long, long)} and {@link Math#floorMod(long, long)}
ensure that timestamps
+ * before the Unix epoch are converted correctly.
+ */
+@AutoService(Converter.class)
+public class TimeStampNanoTZConverter implements
Converter<TimeStampNanoTZVector> {
+ @Override
+ public Object convert(int rowIndex, TimeStampNanoTZVector fieldVector) {
+ if (fieldVector == null || fieldVector.isNull(rowIndex)) {
+ return null;
+ }
+
+ long epochNano = fieldVector.getObject(rowIndex);
+ long epochSecond = Math.floorDiv(epochNano, 1_000_000_000L);
+ long nanoOfSecond = Math.floorMod(epochNano, 1_000_000_000L);
+
+ String timeZone = fieldVector.getTimeZone();
+ ZoneId zoneId =
+ timeZone == null || timeZone.isEmpty()
+ ? ZoneId.systemDefault()
+ : ZoneId.of(timeZone);
+
+ return Instant.ofEpochSecond(epochSecond,
nanoOfSecond).atZone(zoneId).toLocalDateTime();
+ }
+
+ @Override
+ public boolean support(Types.MinorType type) {
+ return Types.MinorType.TIMESTAMPNANOTZ == type;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java
index 604c10110e..87c7b62db7 100644
---
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java
+++
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java
@@ -34,8 +34,10 @@ import
org.apache.seatunnel.shade.org.apache.arrow.vector.IntVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.LargeVarCharVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.SmallIntVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.TimeMicroVector;
+import
org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampMicroTZVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampMicroVector;
import
org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampMilliTZVector;
+import
org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampNanoTZVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampSecTZVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.TimeStampSecVector;
import org.apache.seatunnel.shade.org.apache.arrow.vector.TinyIntVector;
@@ -69,6 +71,7 @@ import java.io.ByteArrayOutputStream;
import java.math.BigDecimal;
import java.nio.channels.Channels;
import java.nio.charset.StandardCharsets;
+import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@@ -89,6 +92,19 @@ public class ArrowToSeatunnelRowReaderTest {
private static final LocalDateTime localDateTime =
LocalDateTime.parse(
"2025-02-15 02:21:23",
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+ private static final LocalDateTime localDateTimeMicro =
+ LocalDateTime.parse(
+ "2025-02-15 02:21:23.123456",
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS"));
+ private static final LocalDateTime localDateTimeNano =
+ LocalDateTime.parse(
+ "2025-02-15 02:21:23.123456789",
+ DateTimeFormatter.ofPattern("yyyy-MM-dd
HH:mm:ss.SSSSSSSSS"));
+
+ private static final ZoneId timestampTzZoneId =
+ ZoneId.systemDefault().equals(ZoneId.of("Asia/Shanghai"))
+ ? ZoneId.of("America/New_York")
+ : ZoneId.of("Asia/Shanghai");
private static final List<String> stringData = new ArrayList<>();
private static final List<Byte> byteData = new ArrayList<>();
@@ -127,6 +143,8 @@ public class ArrowToSeatunnelRowReaderTest {
seaTunnelDataTypeHolder.add(new
SeaTunnelDataTypeHolder("timestampSec", 0));
seaTunnelDataTypeHolder.add(new
SeaTunnelDataTypeHolder("timestampSecTz", 0));
seaTunnelDataTypeHolder.add(new SeaTunnelDataTypeHolder("map", 0));
+ seaTunnelDataTypeHolder.add(new
SeaTunnelDataTypeHolder("timestampMicroTz", 0));
+ seaTunnelDataTypeHolder.add(new
SeaTunnelDataTypeHolder("timestampNanoTz", 0));
}
private static VectorSchemaRoot buildVectorSchemaRoot(
@@ -161,6 +179,22 @@ public class ArrowToSeatunnelRowReaderTest {
new ArrowType.Timestamp(
TimeUnit.MILLISECOND,
ZoneId.systemDefault().getId())),
rootAllocator));
+ // Microsecond timestamp with timezone
+ vectors.add(
+ new TimeStampMicroTZVector(
+ Field.nullable(
+ "timestampMicroTz",
+ new ArrowType.Timestamp(
+ TimeUnit.MICROSECOND,
timestampTzZoneId.getId())),
+ rootAllocator));
+ // Nanosecond timestamp with timezone
+ vectors.add(
+ new TimeStampNanoTZVector(
+ Field.nullable(
+ "timestampNanoTz",
+ new ArrowType.Timestamp(
+ TimeUnit.NANOSECOND,
timestampTzZoneId.getId())),
+ rootAllocator));
vectors.add(new TimeMicroVector("time", rootAllocator));
vectors.add(new DateMilliVector("date1", rootAllocator));
vectors.add(new DateDayVector("date2", rootAllocator));
@@ -186,6 +220,12 @@ public class ArrowToSeatunnelRowReaderTest {
long epochMilli =
localDateTime.atZone(zoneId).toInstant().toEpochMilli();
long epochSecond =
localDateTime.atZone(zoneId).toInstant().getEpochSecond();
+ Instant microInstant =
localDateTimeMicro.atZone(timestampTzZoneId).toInstant();
+ long epochMicro =
+ microInstant.getEpochSecond() * 1_000_000L +
microInstant.getNano() / 1_000;
+ Instant nanoInstant =
localDateTimeNano.atZone(timestampTzZoneId).toInstant();
+ long epochNano = nanoInstant.getEpochSecond() * 1_000_000_000L +
nanoInstant.getNano();
+
byte byteStart = 'a';
// setVectorValue
@@ -239,6 +279,21 @@ public class ArrowToSeatunnelRowReaderTest {
((TimeStampSecVector) vector).setSafe(i,
epochSecond);
} else if (vector instanceof TimeStampSecTZVector) {
((TimeStampSecTZVector) vector).setSafe(i,
epochSecond);
+ } else if (vector instanceof TimeStampMicroTZVector) {
+ TimeStampMicroTZVector timestampVector =
+ (TimeStampMicroTZVector) vector;
+ if (i == 0) {
+ timestampVector.setNull(i);
+ } else {
+ timestampVector.setSafe(i, epochMicro);
+ }
+ } else if (vector instanceof TimeStampNanoTZVector) {
+ TimeStampNanoTZVector timestampVector =
(TimeStampNanoTZVector) vector;
+ if (i == 0) {
+ timestampVector.setNull(i);
+ } else {
+ timestampVector.setSafe(i, epochNano);
+ }
}
}
});
@@ -414,11 +469,26 @@ public class ArrowToSeatunnelRowReaderTest {
rows.stream().map(s ->
s.getField(19)).distinct().collect(Collectors.toList());
Assertions.assertEquals(
Collections.singletonList(localDateTime),
actualTimestampSecTzData);
-
// todo check map
// The java api has problems building MapVectors,and there are no
examples on the
// official website
// @see https://github.com/apache/arrow/issues/44664
+
+ // Check null microsecond timestamp with timezone
+ Assertions.assertNull(rows.get(0).getField(21));
+
+ // Check non-null microsecond timestamps with timezone
+ for (int i = 1; i < rows.size(); i++) {
+ Assertions.assertEquals(localDateTimeMicro,
rows.get(i).getField(21));
+ }
+
+ // Check null nanosecond timestamp with timezone
+ Assertions.assertNull(rows.get(0).getField(22));
+
+ // Check non-null nanosecond timestamps with timezone
+ for (int i = 1; i < rows.size(); i++) {
+ Assertions.assertEquals(localDateTimeNano,
rows.get(i).getField(22));
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/SeaTunnelDataTypeHolder.java
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/SeaTunnelDataTypeHolder.java
index 308ba54647..33e14f8ad7 100644
---
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/SeaTunnelDataTypeHolder.java
+++
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/SeaTunnelDataTypeHolder.java
@@ -67,6 +67,8 @@ public class SeaTunnelDataTypeHolder {
case "timestamp2":
case "timestampSec":
case "timestampSecTz":
+ case "timestampMicroTz":
+ case "timestampNanoTz":
return LocalTimeType.LOCAL_DATE_TIME_TYPE;
case "time":
return LocalTimeType.LOCAL_TIME_TYPE;
