This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 f7a575432f [Fix][Connector-V2] Keep TIMESTAMP microseconds in the
Lance sink (#13043)
f7a575432f is described below
commit f7a575432fba5c35d36659c16e1590e83ad1f296
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Wed Oct 28 13:46:01 2026 +0000
[Fix][Connector-V2] Keep TIMESTAMP microseconds in the Lance sink (#13043)
---
.../introduction/concepts/incompatible-changes.md | 7 ++
.../introduction/concepts/incompatible-changes.md | 7 ++
.../lance/sink/writers/BaseTypeWriter.java | 17 +++-
.../lance/sink/writers/TimestampTypeWriter.java | 6 +-
.../lance/utils/FragmentConverterTest.java | 90 ++++++++++++++++++++++
.../seatunnel/e2e/connector/lance/LanceIT.java | 51 ++++++++++++
.../resources/lance/fake_timestamp_to_lance.conf | 67 ++++++++++++++++
7 files changed, 240 insertions(+), 5 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 39bdf4731f..b3753c0ba2 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -5,6 +5,13 @@ You need to check this document before you upgrade to related
version.
## dev
+### Lance sink: TIMESTAMP values keep their microseconds
+
+- **Behavior change: a `TIMESTAMP` written by the Lance sink keeps the digits
below the millisecond**
+ - **Description**: The Lance sink declares a `TIMESTAMP` column with
microsecond precision, as documented, but `BaseTypeWriter.convertToEpochMicro`
converted `LocalDateTime` and `java.sql.Timestamp` values through milliseconds,
so every stored value ended in `000` microseconds. Values are now converted
from the instant directly.
+ - **Impact**: For the same input, a newly written value can differ from one
written by an earlier version in its last three microsecond digits. Rows
already in a dataset are not rewritten, so one dataset can hold both.
Millisecond columns, and values with no digits below the millisecond, are
unchanged.
+ - **Migration Guide**: No configuration change is needed. If a consumer
compares or deduplicates new rows against rows written by an earlier version,
compare at millisecond precision, or rewrite the older rows from the source.
+
### JDBC MySQL: DATETIME writes preserve local date/time fields
The MySQL JDBC sink now binds SeaTunnel `TIMESTAMP` (`LocalDateTime`) directly
instead of converting it through `java.sql.Timestamp`. With different worker
JVM and JDBC connection time zones, affected versions could shift `DATETIME`
values or reject valid boundary values. Writes now preserve the local fields
and the fractional seconds supported by the target column. The `DATETIME`
source mapping and `TIMESTAMP_TZ` handling are unchanged.
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 24b14e119e..6145c12f3d 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -4,6 +4,13 @@
## dev
+### Lance Sink：TIMESTAMP 值保留微秒
+
+- **行为变更：Lance Sink 写入的 `TIMESTAMP` 保留毫秒以下的数字**
+ - **描述**：Lance Sink 按文档以微秒精度声明 `TIMESTAMP` 列，但
`BaseTypeWriter.convertToEpochMicro` 会先把 `LocalDateTime` 和 `java.sql.Timestamp`
值转换为毫秒，因此写入的每个值最后三位微秒都是 `000`。现在直接由时间点（Instant）转换。
+ -
**影响**：对同一输入，新写入的值可能与早期版本写入的值在最后三位微秒上不同。数据集中已有的行不会被重写，因此同一个数据集中可能同时存在两种值。毫秒精度的列，以及没有毫秒以下数字的值，不受影响。
+ - **迁移指南**：无需修改配置。如果下游需要把新写入的行与早期版本写入的行进行比较或去重，请按毫秒精度比较，或从数据源重新写入旧数据。
+
### JDBC MySQL：DATETIME 写入保留本地日期和时间字段
MySQL JDBC Sink 现在直接绑定 SeaTunnel `TIMESTAMP`（`LocalDateTime`），不再通过
`java.sql.Timestamp` 转换。当 Worker JVM 与 JDBC 连接的时区不同时，受影响的版本可能导致 `DATETIME`
值偏移，或拒绝合法的边界值。现在写入会保留本地字段及目标列支持的小数秒精度。`DATETIME` 的源端映射和 `TIMESTAMP_TZ` 的处理保持不变。
diff --git
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/BaseTypeWriter.java
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/BaseTypeWriter.java
index d854899a15..04b6b5ea19 100644
---
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/BaseTypeWriter.java
+++
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/BaseTypeWriter.java
@@ -21,6 +21,7 @@ import org.apache.arrow.memory.ArrowBuf;
import org.apache.arrow.memory.BufferAllocator;
import java.nio.charset.StandardCharsets;
+import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -44,10 +45,9 @@ public abstract class BaseTypeWriter implements TypeWriter {
protected long convertToEpochMicro(Object value) {
if (value instanceof LocalDateTime) {
- return ((LocalDateTime)
value).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()
- * 1000;
+ return toEpochMicro(((LocalDateTime)
value).atZone(ZoneId.systemDefault()).toInstant());
} else if (value instanceof java.sql.Timestamp) {
- return ((java.sql.Timestamp) value).getTime() * 1000;
+ return toEpochMicro(((java.sql.Timestamp) value).toInstant());
} else if (value instanceof java.util.Date) {
return ((java.util.Date) value).getTime() * 1000;
} else if (value instanceof Number) {
@@ -57,4 +57,15 @@ public abstract class BaseTypeWriter implements TypeWriter {
"Unsupported timestamp value type: " + value.getClass());
}
}
+
+ /**
+ * Converts an instant to microseconds since the epoch without passing
through milliseconds,
+ * which would drop the sub-millisecond digits. {@code getEpochSecond()}
rounds down and {@code
+ * getNano()} is never negative, so an instant before 1970 converts
exactly as well.
+ */
+ private static long toEpochMicro(Instant instant) {
+ return Math.addExact(
+ Math.multiplyExact(instant.getEpochSecond(), 1_000_000L),
+ instant.getNano() / 1_000);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/TimestampTypeWriter.java
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/TimestampTypeWriter.java
index b504b3ed9f..e76d2eeb22 100644
---
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/TimestampTypeWriter.java
+++
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/sink/writers/TimestampTypeWriter.java
@@ -48,7 +48,9 @@ public class TimestampTypeWriter extends BaseTypeWriter {
((TimeStampMicroVector) vector).setSafe(rowIndex, epochMicro);
}
} else if (unit == TimeUnit.MILLISECOND) {
- long epochMilli = epochMicro / 1000;
+ // floorDiv rather than /, so an instant before 1970 rounds down
to the millisecond it
+ // falls in instead of up towards the epoch.
+ long epochMilli = Math.floorDiv(epochMicro, 1000L);
if (timezone != null && !timezone.isEmpty()) {
((TimeStampMilliTZVector) vector).setSafe(rowIndex,
epochMilli);
} else {
@@ -68,7 +70,7 @@ public class TimestampTypeWriter extends BaseTypeWriter {
if (unit == TimeUnit.MICROSECOND) {
writer.writeTimeStampMicro(epochMicro);
} else if (unit == TimeUnit.MILLISECOND) {
- writer.writeTimeStampMilli(epochMicro / 1000);
+ writer.writeTimeStampMilli(Math.floorDiv(epochMicro, 1000L));
}
}
diff --git
a/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java
b/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java
index 67083313cb..3e042d6733 100644
---
a/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java
+++
b/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.lance.utils;
import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
import org.apache.seatunnel.api.table.type.MapType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -26,10 +27,13 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.lance.exception.LanceConnectorException;
import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.arrow.vector.TimeStampMilliVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.complex.ListVector;
import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.FieldType;
@@ -51,6 +55,11 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -205,6 +214,87 @@ class FragmentConverterTest {
exception.getMessage());
}
+ @Test
+ void shouldKeepTheMicrosecondsOfATimestamp() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"c_timestamp"},
+ new SeaTunnelDataType<?>[]
{LocalTimeType.LOCAL_DATE_TIME_TYPE});
+ LocalDateTime recent = LocalDateTime.of(2025, 2, 15, 10, 21, 23,
123_456_000);
+ LocalDateTime beforeEpoch = LocalDateTime.of(1960, 6, 1, 12, 0, 0,
250_000);
+ List<SeaTunnelRow> rows =
+ Arrays.asList(
+ new SeaTunnelRow(new Object[] {recent}),
+ new SeaTunnelRow(new Object[] {beforeEpoch}),
+ new SeaTunnelRow(new Object[]
{Timestamp.valueOf(recent)}));
+ // The schema the sink declares when it creates the dataset itself.
+ Schema schema = SchemaUtils.convertSchema(rows.get(0), rowType);
+ Assertions.assertEquals(
+ new ArrowType.Timestamp(TimeUnit.MICROSECOND, "Asia/Shanghai"),
+ schema.getFields().get(0).getType());
+
+ try (RootAllocator allocator = new RootAllocator();
+ VectorSchemaRoot root =
+ FragmentConverter.convertToVectorSchemaRoot(
+ rows, rowType, schema, allocator)) {
+ TimeStampMicroTZVector vector = (TimeStampMicroTZVector)
root.getVector("c_timestamp");
+ Assertions.assertEquals(epochMicros(recent), vector.get(0));
+ Assertions.assertEquals(epochMicros(beforeEpoch), vector.get(1));
+ Assertions.assertEquals(epochMicros(recent), vector.get(2));
+ // The fraction of a second does not depend on the zone the value
is encoded in.
+ Assertions.assertEquals(123_456L, Math.floorMod(vector.get(0),
1_000_000L));
+ Assertions.assertEquals(250L, Math.floorMod(vector.get(1),
1_000_000L));
+ }
+ }
+
+ @Test
+ void shouldRoundAMillisecondTimestampBeforeTheEpochDown() {
+ // An existing dataset keeps its own schema, so the sink can meet a
millisecond timestamp
+ // column it did not create, on its own or inside a list.
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"c_millis", "c_millis_list"},
+ new SeaTunnelDataType<?>[] {
+ LocalTimeType.LOCAL_DATE_TIME_TYPE,
ArrayType.LOCAL_DATE_TIME_ARRAY_TYPE
+ });
+ ArrowType millis = new ArrowType.Timestamp(TimeUnit.MILLISECOND, null);
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ Field.nullable("c_millis", millis),
+ new Field(
+ "c_millis_list",
+ FieldType.nullable(new
ArrowType.List()),
+ Collections.singletonList(
+ Field.nullable("item",
millis)))));
+ // 250 microseconds past a millisecond, before 1970 in every zone.
+ LocalDateTime beforeEpoch = LocalDateTime.of(1960, 6, 1, 12, 0, 0,
250_000);
+ long expected =
beforeEpoch.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+
+ try (RootAllocator allocator = new RootAllocator();
+ VectorSchemaRoot root =
+ FragmentConverter.convertToVectorSchemaRoot(
+ Collections.singletonList(
+ new SeaTunnelRow(
+ new Object[] {
+ beforeEpoch,
+
Collections.singletonList(beforeEpoch)
+ })),
+ rowType,
+ schema,
+ allocator)) {
+ Assertions.assertEquals(
+ expected, ((TimeStampMilliVector)
root.getVector("c_millis")).get(0));
+ ListVector list = (ListVector) root.getVector("c_millis_list");
+ Assertions.assertEquals(expected, ((TimeStampMilliVector)
list.getDataVector()).get(0));
+ }
+ }
+
+ private static long epochMicros(LocalDateTime value) {
+ Instant instant = value.atZone(ZoneId.systemDefault()).toInstant();
+ return ChronoUnit.MICROS.between(Instant.EPOCH, instant);
+ }
+
private static void createEmptyDataset(
RootAllocator allocator, String datasetPath, Schema schema) {
try (Dataset ignored =
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/java/org/apache/seatunnel/e2e/connector/lance/LanceIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/java/org/apache/seatunnel/e2e/connector/lance/LanceIT.java
index 5cb19b387f..dddca4518d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/java/org/apache/seatunnel/e2e/connector/lance/LanceIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/java/org/apache/seatunnel/e2e/connector/lance/LanceIT.java
@@ -24,6 +24,13 @@ import
org.apache.seatunnel.e2e.common.container.TestContainerId;
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.condition.DisabledOnOs;
@@ -31,9 +38,12 @@ import org.junit.jupiter.api.condition.OS;
import org.testcontainers.containers.Container;
import com.lancedb.lance.Dataset;
+import com.lancedb.lance.ipc.LanceScanner;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.given;
@@ -57,6 +67,7 @@ public class LanceIT extends TestSuiteBase {
private static final String DATASET_PATH = "/tmp/seatunnel_mnt/lanceTest/";
private static final String TABLE_NAME = "lance_sink_table";
+ private static final String TIMESTAMP_TABLE_NAME = "lance_timestamp_table";
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
@@ -106,6 +117,46 @@ public class LanceIT extends TestSuiteBase {
});
}
+ @TestTemplate
+ public void testTimestampKeepsItsMicroseconds(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult writeResult =
+ container.executeJob("/lance/fake_timestamp_to_lance.conf");
+ Assertions.assertEquals(0, writeResult.getExitCode(),
writeResult.getStderr());
+
+ // The fraction of a second, in microseconds, that each id must keep.
Only the fraction is
+ // compared, because the whole seconds depend on the zone of the JVM
that wrote the value.
+ Map<Integer, Long> expected = new HashMap<>();
+ expected.put(1, 123_456L);
+ expected.put(2, 250L);
+ expected.put(3, 999_500L);
+
+ // Unlike the row count check above, this does not skip when the
dataset cannot be read
+ // in the test JVM, so a pass always means the stored values were
checked.
+ given().ignoreExceptions()
+ .await()
+ .atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(expected,
readTimestampFractions()));
+ }
+
+ private Map<Integer, Long> readTimestampFractions() throws Exception {
+ Map<Integer, Long> fractions = new HashMap<>();
+ try (BufferAllocator allocator = new RootAllocator();
+ Dataset dataset = Dataset.open(DATASET_PATH +
TIMESTAMP_TABLE_NAME, allocator);
+ LanceScanner scanner = dataset.newScan();
+ ArrowReader reader = scanner.scanBatches()) {
+ while (reader.loadNextBatch()) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ IntVector ids = (IntVector) root.getVector("id");
+ TimeStampVector timestamps = (TimeStampVector)
root.getVector("c_timestamp");
+ for (int i = 0; i < root.getRowCount(); i++) {
+ fractions.put(ids.get(i), Math.floorMod(timestamps.get(i),
1_000_000L));
+ }
+ }
+ }
+ return fractions;
+ }
+
private long loadLanceTableCount() {
long count = 0;
try {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/resources/lance/fake_timestamp_to_lance.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/resources/lance/fake_timestamp_to_lance.conf
new file mode 100644
index 0000000000..24f1ef5b1e
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-lance-e2e/src/test/resources/lance/fake_timestamp_to_lance.conf
@@ -0,0 +1,67 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+
+ # You can set spark configuration here
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 2
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ FakeSource {
+ schema = {
+ fields {
+ id = int
+ c_timestamp = timestamp
+ }
+ }
+ # Each value has digits below the millisecond. The second and third are
before 1970.
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, "2025-02-15T10:21:23.123456"]
+ },
+ {
+ kind = INSERT
+ fields = [2, "1960-06-01T12:00:00.000250"]
+ },
+ {
+ kind = INSERT
+ fields = [3, "1969-12-31T23:59:59.999500"]
+ }
+ ]
+ plugin_output = "fake"
+ }
+}
+
+transform {
+}
+
+sink {
+ Lance {
+ dataset_path = "/tmp/seatunnel_mnt/lanceTest/lance_timestamp_table"
+ namespace_type = "dir"
+ namespace_id = "root"
+ table = "lance_timestamp_table"
+ }
+}
