[doris-flink-connector] branch master updated: fix timezone and add sqlserver type (#185)

[zykkk]

This is an automated email from the ASF dual-hosted git repository.

zykkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git


The following commit(s) were added to refs/heads/master by this push:
     new fd6ddbc  fix timezone and add sqlserver type (#185)
fd6ddbc is described below

commit fd6ddbc658ca89b9e9873586be81d6bfc8bf19e7
Author: wudi <676366...@qq.com>
AuthorDate: Mon Aug 28 10:22:40 2023 +0800

    fix timezone and add sqlserver type (#185)
---
 .../java/org/apache/doris/flink/tools/cdc/DateToStringConverter.java   | 2 +-
 .../java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java | 3 +++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git 
a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DateToStringConverter.java
 
b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DateToStringConverter.java
index ed5b2b6..60a9718 100644
--- 
a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DateToStringConverter.java
+++ 
b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DateToStringConverter.java
@@ -139,7 +139,7 @@ public class DateToStringConverter implements 
CustomConverter<SchemaBuilder, Rel
             LocalDateTime localDateTime = 
zonedDateTime.withZoneSameInstant(timestampZoneId).toLocalDateTime();
             return timestampFormatter.format(localDateTime);
         } else if (input instanceof Timestamp) {
-            return timestampFormatter.format(((Timestamp) 
input).toLocalDateTime());
+            return timestampFormatter.format(((Timestamp) 
input).toInstant().atZone(timestampZoneId).toLocalDateTime());
         }
         return null;
     }
diff --git 
a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java
 
b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java
index aedb16f..f09bd76 100644
--- 
a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java
+++ 
b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java
@@ -41,6 +41,7 @@ public class SqlServerType {
     private static final String TEXT = "text";
     private static final String NTEXT = "ntext";
     private static final String TIME = "time";
+    private static final String TIMESTAMP = "timestamp";
     private static final String DATETIMEOFFSET = "datetimeoffset";
     private static final String IMAGE = "image";
     private static final String BINARY = "binary";
@@ -67,6 +68,7 @@ public class SqlServerType {
                 return String.format("%s(%s,%s)", DorisType.DECIMAL_V3, 19, 4);
             case SMALLMONEY:
                 return String.format("%s(%s,%s)", DorisType.DECIMAL_V3, 10, 4);
+            case DECIMAL:
             case NUMERIC:
                 return precision != null && precision > 0 && precision <= 38
                         ? String.format("%s(%s,%s)", DorisType.DECIMAL_V3, 
precision, scale != null && scale >= 0 ? scale : 0)
@@ -86,6 +88,7 @@ public class SqlServerType {
             case NTEXT:
             case TIME:
             case DATETIMEOFFSET:
+            case TIMESTAMP:
                 return DorisType.STRING;
             default:
                 throw new UnsupportedOperationException("Unsupported SqlServer 
Type: " + sqlServerType);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org