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-12541-07f32fffbda5452bbf4a3f4dc6de5e36ca69e5ab in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit d1009b873caed13cffc6a296873e2534d83656eb Author: hutiefang76 <[email protected]> AuthorDate: Thu Oct 8 13:35:49 2026 +0000 [Fix][Connector-V2] Preserve DuckDB TIME microseconds (#12541) Co-authored-by: David Zollo <[email protected]> Co-authored-by: Claude Sonnet 5.5 <[email protected]> --- docs/en/connectors/sink/DuckDB.md | 2 + docs/en/connectors/source/DuckDB.md | 2 + docs/zh/connectors/sink/DuckDB.md | 2 + docs/zh/connectors/source/DuckDB.md | 2 + .../dialect/duckdb/DuckDBJdbcRowConverter.java | 18 ++++ .../dialect/duckdb/DuckDBSourceAndSinkTest.java | 95 ++++++++++++++++++++++ 6 files changed, 121 insertions(+) diff --git a/docs/en/connectors/sink/DuckDB.md b/docs/en/connectors/sink/DuckDB.md index e44679be70..60a14d55c8 100644 --- a/docs/en/connectors/sink/DuckDB.md +++ b/docs/en/connectors/sink/DuckDB.md @@ -63,6 +63,8 @@ works against a local database file path (`jdbc:duckdb:/path/to/database.db`) or | TIMESTAMP | TIMESTAMP | | BYTES<br/>ARRAY<br/>ROW<br/>MAP | BLOB | +DuckDB `TIME` values preserve microsecond precision when read or written through the JDBC connector. They represent a local time of day without a time zone. + ## Sink Options | Name | Type | Required | Default | Description | diff --git a/docs/en/connectors/source/DuckDB.md b/docs/en/connectors/source/DuckDB.md index 64675855bf..1bca13ce7c 100644 --- a/docs/en/connectors/source/DuckDB.md +++ b/docs/en/connectors/source/DuckDB.md @@ -74,6 +74,8 @@ MySQL automatic DDL cannot create a full-column primary key on `LONGTEXT`. If a | TIMESTAMP<br/>TIMESTAMP WITH TIME ZONE | TIMESTAMP | | BLOB<br/>ARRAY<br/>STRUCT<br/>MAP | BYTES | +DuckDB `TIME` values preserve microsecond precision when read or written through the JDBC connector. They represent a local time of day without a time zone. + > Type names are matched without regard to case or the JVM default locale. For > example, `integer` and `INTEGER` both map to `INT`, including under `tr-TR`. ## Source Options diff --git a/docs/zh/connectors/sink/DuckDB.md b/docs/zh/connectors/sink/DuckDB.md index 8f5df43252..b0894d59e4 100644 --- a/docs/zh/connectors/sink/DuckDB.md +++ b/docs/zh/connectors/sink/DuckDB.md @@ -59,6 +59,8 @@ import ChangeLog from '../changelog/connector-jdbc.md'; | TIMESTAMP | TIMESTAMP | | BYTES<br/>ARRAY<br/>ROW<br/>MAP | BLOB | +JDBC 连接器读取和写入 DuckDB `TIME` 时保留微秒精度。该类型表示不带时区的本地时刻。 + ## Sink 选项 | 名称 | 类型 | 是否必需 | 默认值 | 描述 | diff --git a/docs/zh/connectors/source/DuckDB.md b/docs/zh/connectors/source/DuckDB.md index 78c03ed450..755731a123 100644 --- a/docs/zh/connectors/source/DuckDB.md +++ b/docs/zh/connectors/source/DuckDB.md @@ -71,6 +71,8 @@ MySQL 自动建表无法在 `LONGTEXT` 上创建使用完整列值的主键。 | TIMESTAMP<br/>TIMESTAMP WITH TIME ZONE | TIMESTAMP | | BLOB<br/>ARRAY<br/>STRUCT<br/>MAP | BYTES | +JDBC 连接器读取和写入 DuckDB `TIME` 时保留微秒精度。该类型表示不带时区的本地时刻。 + > 类型名识别不区分大小写，也不受 JVM 默认区域设置影响。例如，在 `tr-TR` 下，`integer` 和 `INTEGER` 均映射为 `INT`。 ## 源选项 diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBJdbcRowConverter.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBJdbcRowConverter.java index 59250279af..6c853e6ebc 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBJdbcRowConverter.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBJdbcRowConverter.java @@ -19,12 +19,30 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.duckdb; import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter; import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier; +import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.JdbcFieldTypeUtils; import lombok.extern.slf4j.Slf4j; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.time.LocalTime; + @Slf4j public class DuckDBJdbcRowConverter extends AbstractJdbcRowConverter { + @Override + protected LocalTime readTime(ResultSet resultSet, int index) throws SQLException { + // java.sql.Time discards DuckDB TIME's fractional seconds. + return JdbcFieldTypeUtils.getLocalTime(resultSet, index); + } + + @Override + protected void writeTime(PreparedStatement statement, int index, LocalTime time) + throws SQLException { + statement.setObject(index, time); + } + @Override public String converterName() { return DatabaseIdentifier.DUCKDB; diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBSourceAndSinkTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBSourceAndSinkTest.java index 33dd8ed038..cab6f35522 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBSourceAndSinkTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/duckdb/DuckDBSourceAndSinkTest.java @@ -22,6 +22,7 @@ import org.apache.seatunnel.api.sink.DataSaveMode; import org.apache.seatunnel.api.sink.SchemaSaveMode; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.common.utils.JdbcUrlUtil; import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.duckdb.DuckDBCatalog; @@ -44,10 +45,12 @@ import java.sql.Connection; import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.Statement; +import java.time.LocalTime; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.TimeZone; @TestInstance(TestInstance.Lifecycle.PER_CLASS) public class DuckDBSourceAndSinkTest { @@ -81,6 +84,98 @@ public class DuckDBSourceAndSinkTest { } } + @Test + public void testSourceTimePrecision() throws Exception { + verifyTimePrecision(true); + } + + @Test + public void testSinkTimePrecision() throws Exception { + verifyTimePrecision(false); + } + + private void verifyTimePrecision(boolean verifySource) throws Exception { + TimeZone original = TimeZone.getDefault(); + LocalTime[] times = { + LocalTime.parse("00:00:00"), + LocalTime.parse("00:00:00.000001"), + LocalTime.parse("12:34:56.123456"), + LocalTime.parse("23:59:59.999999"), + null + }; + try { + for (String zone : new String[] {"UTC", "Asia/Shanghai", "America/New_York"}) { + TimeZone.setDefault(TimeZone.getTimeZone(zone)); + try (Connection connection = DriverManager.getConnection(jdbcUrl); + Statement statement = connection.createStatement()) { + statement.execute( + "CREATE OR REPLACE TABLE main.time_precision_source (id INTEGER, value TIME)"); + statement.execute( + "CREATE OR REPLACE TABLE main.time_precision_sink (id INTEGER, value TIME)"); + statement.execute( + "INSERT INTO main.time_precision_source VALUES (0, TIME '00:00:00'), (1, TIME '00:00:00.000001'), (2, TIME '12:34:56.123456'), (3, TIME '23:59:59.999999'), (4, NULL)"); + } + Map<String, Object> sourceOptions = new HashMap<>(); + sourceOptions.put("url", jdbcUrl); + sourceOptions.put("driver", "org.duckdb.DuckDBDriver"); + sourceOptions.put("table_path", "main.time_precision_source"); + ReadonlyConfig config = ReadonlyConfig.fromMap(sourceOptions); + CatalogTable table = + new JdbcSourceFactory() + .inferSchemaForDryRun( + new TableSourceFactoryContext( + config, getClass().getClassLoader())) + .get(0); + List<SeaTunnelRow> rows = + SourceFlowTestUtils.runBatchWithCheckpointDisabled( + config, new JdbcSourceFactory()); + Assertions.assertEquals(times.length, rows.size()); + if (verifySource) { + for (SeaTunnelRow row : rows) { + Assertions.assertEquals( + times[(Integer) row.getField(0)], + row.getField(1), + "Source precision in " + zone); + } + } + // Use the original values independently of Source so a matching truncation on + // both sides cannot make the Sink regression pass. + List<SeaTunnelRow> sinkRows = new ArrayList<>(); + for (int id = 0; id < times.length; id++) { + sinkRows.add(new SeaTunnelRow(new Object[] {id, times[id]})); + } + Map<String, Object> sinkOptions = new HashMap<>(); + sinkOptions.put("url", jdbcUrl); + sinkOptions.put("driver", "org.duckdb.DuckDBDriver"); + sinkOptions.put("query", "INSERT INTO main.time_precision_sink VALUES (?, ?)"); + sinkOptions.put("schema_save_mode", SchemaSaveMode.IGNORE); + sinkOptions.put("data_save_mode", DataSaveMode.APPEND_DATA); + SinkFlowTestUtils.runBatchWithCheckpointDisabled( + table, + ReadonlyConfig.fromMap(sinkOptions), + new JdbcSinkFactory(), + sinkRows); + try (Connection connection = DriverManager.getConnection(jdbcUrl); + Statement statement = connection.createStatement(); + ResultSet result = + statement.executeQuery( + "SELECT value::VARCHAR FROM main.time_precision_sink ORDER BY id")) { + for (LocalTime time : times) { + Assertions.assertTrue(result.next()); + String value = result.getString(1); + Assertions.assertEquals( + time, + value == null ? null : LocalTime.parse(value), + "Sink precision in " + zone); + } + Assertions.assertFalse(result.next()); + } + } + } finally { + TimeZone.setDefault(original); + } + } + @SneakyThrows @Test public void testFlow() {
