This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13043-50315b321bc68d910352d78f0fcbff80872451f5 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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" + } +}
