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-12791-f4474404f3772b26283289c56475a57784c1bb1f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c14fef24f5b6bb8e60cb2aa1ef7fcafaf389c1a0 Author: Stochastic <[email protected]> AuthorDate: Sat Oct 17 15:49:36 2026 +0000 [Fix][Connector-V2] Support Dameng local time zone and interval types (#12791) Co-authored-by: yancy <[email protected]> Co-authored-by: Asthenia0412 <[email protected]> --- docs/en/connectors/source/Jdbc.md | 1 + docs/zh/connectors/source/Jdbc.md | 1 + .../internal/dialect/dm/DmdbTypeConverter.java | 15 ++++ .../internal/dialect/dm/DmdbTypeConverterTest.java | 72 ++++++++++++++++++ .../connectors/seatunnel/jdbc/JdbcDmIT.java | 87 ++++++++++++++++++++++ 5 files changed, 176 insertions(+) diff --git a/docs/en/connectors/source/Jdbc.md b/docs/en/connectors/source/Jdbc.md index 3a89100647..9f1274c6b8 100644 --- a/docs/en/connectors/source/Jdbc.md +++ b/docs/en/connectors/source/Jdbc.md @@ -291,6 +291,7 @@ If one dialect not supported by SeaTunnel, it will use the default dialect `Gene | YashanDB | | | Dameng `NCHAR` source columns are mapped to SeaTunnel `STRING`. +Dameng `TIMESTAMP WITH LOCAL TIME ZONE` source columns are mapped to `TIMESTAMP_TZ`, and `INTERVAL` source columns are mapped to `STRING`. ## Parallel Reader diff --git a/docs/zh/connectors/source/Jdbc.md b/docs/zh/connectors/source/Jdbc.md index 82c1cf5ea1..5b4612b1c9 100644 --- a/docs/zh/connectors/source/Jdbc.md +++ b/docs/zh/connectors/source/Jdbc.md @@ -287,6 +287,7 @@ int_type_narrowing = false | YashanDB | | | 达梦 `NCHAR` 源字段会映射为 SeaTunnel `STRING`。 +达梦 `TIMESTAMP WITH LOCAL TIME ZONE` 源字段会映射为 `TIMESTAMP_TZ`，`INTERVAL` 源字段会映射为 `STRING`。 ## 并行读取器 diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverter.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverter.java index 496ad642af..2267893283 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverter.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverter.java @@ -91,6 +91,8 @@ public class DmdbTypeConverter implements TypeConverter<BasicTypeDefine> { public static final String DM_TIMESTAMP = "TIMESTAMP"; public static final String DM_DATETIME = "DATETIME"; public static final String DM_DATETIME_WITH_TIME_ZONE = "DATETIME WITH TIME ZONE"; + public static final String DM_TIMESTAMP_WITH_LOCAL_TIME_ZONE = "TIMESTAMP WITH LOCAL TIME ZONE"; + private static final String DM_INTERVAL_PREFIX = "INTERVAL "; public static final int DEFAULT_PRECISION = 38; public static final int MAX_PRECISION = 38; @@ -124,6 +126,14 @@ public class DmdbTypeConverter implements TypeConverter<BasicTypeDefine> { .comment(typeDefine.getComment()); String dmType = typeDefine.getDataType().toUpperCase(); + if (dmType.startsWith("TIMESTAMP(") && dmType.endsWith(" WITH LOCAL TIME ZONE")) { + dmType = DM_TIMESTAMP_WITH_LOCAL_TIME_ZONE; + } + // Interval variants can include precision, so match their shared type prefix. + if (dmType.startsWith(DM_INTERVAL_PREFIX)) { + builder.sourceType(typeDefine.getColumnType()).dataType(BasicType.STRING_TYPE); + return builder.build(); + } switch (dmType) { case DM_BIT: builder.sourceType(DM_BIT); @@ -320,6 +330,11 @@ public class DmdbTypeConverter implements TypeConverter<BasicTypeDefine> { builder.dataType(LocalTimeType.OFFSET_DATE_TIME_TYPE); builder.scale(typeDefine.getScale()); break; + case DM_TIMESTAMP_WITH_LOCAL_TIME_ZONE: + builder.sourceType(typeDefine.getColumnType()); + builder.dataType(LocalTimeType.OFFSET_DATE_TIME_TYPE); + builder.scale(typeDefine.getScale()); + break; default: throw CommonError.convertToSeaTunnelTypeError( DatabaseIdentifier.DAMENG, typeDefine.getDataType(), typeDefine.getName()); diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverterTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverterTest.java index 5eb75170b0..c54bd8afb4 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverterTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbTypeConverterTest.java @@ -31,7 +31,39 @@ import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.Arrays; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + public class DmdbTypeConverterTest { + @Test + public void testMapResultSetMetadataTypes() throws SQLException { + ResultSetMetaData metadata = mock(ResultSetMetaData.class); + when(metadata.getColumnLabel(1)).thenReturn("LTZ_VALUE"); + when(metadata.getColumnTypeName(1)).thenReturn("TIMESTAMP(6) WITH LOCAL TIME ZONE"); + when(metadata.isNullable(1)).thenReturn(ResultSetMetaData.columnNullable); + when(metadata.getPrecision(1)).thenReturn(26); + when(metadata.getScale(1)).thenReturn(6); + + Column localTimeZoneColumn = new DmdbTypeMapper().mappingColumn(metadata, 1); + + Assertions.assertEquals( + LocalTimeType.OFFSET_DATE_TIME_TYPE, localTimeZoneColumn.getDataType()); + Assertions.assertEquals( + "TIMESTAMP(6) WITH LOCAL TIME ZONE", localTimeZoneColumn.getSourceType()); + + when(metadata.getColumnLabel(1)).thenReturn("INTERVAL_VALUE"); + when(metadata.getColumnTypeName(1)).thenReturn("INTERVAL DAY(3) TO SECOND(6)"); + + Column intervalColumn = new DmdbTypeMapper().mappingColumn(metadata, 1); + + Assertions.assertEquals(BasicType.STRING_TYPE, intervalColumn.getDataType()); + Assertions.assertEquals("INTERVAL DAY(3) TO SECOND(6)", intervalColumn.getSourceType()); + } + @Test public void testConvertUnsupported() { BasicTypeDefine<Object> typeDefine = @@ -663,6 +695,46 @@ public class DmdbTypeConverterTest { Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType().toLowerCase()); } + @Test + public void testConvertTimestampWithLocalTimeZone() { + BasicTypeDefine<Object> typeDefine = + BasicTypeDefine.builder() + .name("test") + .columnType("timestamp with local time zone") + .dataType("timestamp with local time zone") + .scale(6) + .build(); + + Column column = DmdbTypeConverter.INSTANCE.convert(typeDefine); + + Assertions.assertEquals(LocalTimeType.OFFSET_DATE_TIME_TYPE, column.getDataType()); + Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType().toLowerCase()); + Assertions.assertEquals(typeDefine.getScale(), column.getScale()); + } + + @Test + public void testConvertIntervalTypes() { + for (String intervalType : + Arrays.asList( + "INTERVAL YEAR TO MONTH", + "INTERVAL YEAR", + "INTERVAL DAY TO SECOND", + "INTERVAL HOUR TO MINUTE", + "INTERVAL MINUTE TO SECOND")) { + BasicTypeDefine<Object> typeDefine = + BasicTypeDefine.builder() + .name("test") + .columnType(intervalType) + .dataType(intervalType.toLowerCase()) + .build(); + + Column column = DmdbTypeConverter.INSTANCE.convert(typeDefine); + + Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType(), intervalType); + Assertions.assertEquals(intervalType, column.getSourceType(), intervalType); + } + } + @Test public void testReconvertBoolean() { Column column = diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-5/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcDmIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-5/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcDmIT.java index 9355607f97..e38a59268c 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-5/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcDmIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-5/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcDmIT.java @@ -21,8 +21,17 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc; import org.apache.seatunnel.shade.com.google.common.collect.Lists; import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Pair; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.LocalTimeType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.utils.JdbcUrlUtil; +import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.dm.DamengCatalog; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dm.DmdbJdbcRowConverter; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.output.Slf4jLogConsumer; import org.testcontainers.containers.wait.strategy.Wait; @@ -30,10 +39,13 @@ import org.testcontainers.utility.DockerLoggerFactory; import java.math.BigDecimal; import java.sql.Date; +import java.sql.ResultSet; +import java.sql.Statement; import java.sql.Timestamp; import java.time.Duration; import java.time.LocalDate; import java.time.LocalDateTime; +import java.time.OffsetDateTime; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -227,6 +239,81 @@ public class JdbcDmIT extends AbstractJdbcIT { return Pair.of(fieldNames, rows); } + @Test + public void testLocalTimeZoneAndIntervalTypeConversion() throws Exception { + String tableName = "E2E_DM_LOCAL_TZ_INTERVAL"; + String tableInfo = DM_USERNAME + "." + tableName; + // Build a local catalog scoped to this test so we don't inadvertently + // enable the inherited testCreateIndex (which fires whenever catalog != null). + String jdbcUrl = jdbcCase.getJdbcUrl().replace(HOST, dbServer.getHost()); + DamengCatalog localCatalog = + new DamengCatalog( + "dameng", + jdbcCase.getUserName(), + jdbcCase.getPassword(), + JdbcUrlUtil.getUrlInfo(jdbcUrl), + DM_USERNAME, + DRIVER_CLASS); + localCatalog.open(); + try { + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE " + + tableInfo + + " (LTZ_VALUE TIMESTAMP(6) WITH LOCAL TIME ZONE," + + " IYM_VALUE INTERVAL YEAR TO MONTH," + + " IDS_VALUE INTERVAL DAY TO SECOND)"); + statement.execute( + "INSERT INTO " + + tableInfo + + " VALUES (TIMESTAMP '2024-05-06 07:08:09.123456 +08:00'," + + " INTERVAL '2-3' YEAR TO MONTH," + + " INTERVAL '4 05:06:07.123456' DAY TO SECOND)"); + connection.commit(); + + CatalogTable catalogTable = + localCatalog.getTable(TablePath.of(DM_DATABASE, DM_USERNAME, tableName)); + Assertions.assertEquals( + LocalTimeType.OFFSET_DATE_TIME_TYPE, + catalogTable.getTableSchema().getColumn("LTZ_VALUE").getDataType()); + Assertions.assertEquals( + "TIMESTAMP WITH LOCAL TIME ZONE", + catalogTable.getTableSchema().getColumn("LTZ_VALUE").getSourceType()); + Assertions.assertEquals( + BasicType.STRING_TYPE, + catalogTable.getTableSchema().getColumn("IYM_VALUE").getDataType()); + Assertions.assertEquals( + BasicType.STRING_TYPE, + catalogTable.getTableSchema().getColumn("IDS_VALUE").getDataType()); + + try (ResultSet resultSet = + statement.executeQuery( + "SELECT LTZ_VALUE, IYM_VALUE, IDS_VALUE FROM " + tableInfo)) { + Assertions.assertTrue(resultSet.next()); + String iymDriverString = resultSet.getString("IYM_VALUE"); + String idsDriverString = resultSet.getString("IDS_VALUE"); + SeaTunnelRow row = + new DmdbJdbcRowConverter() + .toInternal(resultSet, catalogTable.getTableSchema()); + OffsetDateTime convertedTimestamp = (OffsetDateTime) row.getField(0); + Assertions.assertEquals( + OffsetDateTime.parse("2024-05-06T07:08:09.123456+08:00").toInstant(), + convertedTimestamp.toInstant()); + Assertions.assertEquals(iymDriverString, row.getField(1)); + Assertions.assertEquals(idsDriverString, row.getField(2)); + Assertions.assertFalse(resultSet.next()); + } + } finally { + try (Statement statement = connection.createStatement()) { + statement.execute("DROP TABLE " + tableInfo); + connection.commit(); + } + } + } finally { + localCatalog.close(); + } + } + protected String buildTableInfoWithSchema(String database, String schema, String table) { return buildTableInfoWithSchema(schema, table); }
