[Fix][Connector] Add PostgreSql Type Converter (#379)

datavane · Mar 14, 2024 · d908b9a · d908b9a
1 parent 5599c0b
commit d908b9a
Show file tree

Hide file tree

Showing 4 changed files with 92 additions and 0 deletions.
diff --git a/...or-postgresql/src/main/java/io/datavines/connector/plugin/PostgreSqlConnectorFactory.java b/...or-postgresql/src/main/java/io/datavines/connector/plugin/PostgreSqlConnectorFactory.java
@@ -44,4 +44,9 @@ public Executor getExecutor() {
     public ConfigBuilder getConfigBuilder() {
         return new PostgreSqlConfigBuilder();
     }
+
+    @Override
+    public TypeConverter getTypeConverter() {
+        return new PostgreSqlTypeConverter();
+    }
 }
diff --git a/...ector-postgresql/src/main/java/io/datavines/connector/plugin/PostgreSqlTypeConverter.java b/...ector-postgresql/src/main/java/io/datavines/connector/plugin/PostgreSqlTypeConverter.java
@@ -0,0 +1,78 @@
+/*
+ * 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.
+ */
+package io.datavines.connector.plugin;
+
+import io.datavines.common.enums.DataType;
+import io.datavines.common.utils.StringUtils;
+
+public class PostgreSqlTypeConverter extends JdbcTypeConverter{
+    @Override
+    public DataType convert(String originType) {
+        if (StringUtils.isEmpty(originType)) {
+            throw new UnsupportedOperationException("sql type id null error");
+        }
+        switch (originType.toUpperCase()) {
+            case "INT4":
+            case "INT2":
+            case "OID":
+            case "SERIAL":
+                return DataType.INT_TYPE;
+            case "BIGSERIAL":
+            case "INT8":
+                return DataType.LONG_TYPE;
+            case "BOOL":
+                return DataType.BOOLEAN_TYPE;
+            case "FLOAT8":
+            case "FLOAT4":
+            case "REAL":
+                return DataType.FLOAT_TYPE;
+            case "NUMBER":
+            case "MONEY":
+                return DataType.DOUBLE_TYPE;
+            case "TIMESTAMPTZ":
+                return DataType.TIMESTAMP_TYPE;
+            case "TIMETZ":
+                return DataType.TIME_TYPE;
+            case "BPCHAR":
+            case "UUID":
+            case "JSONB":
+            case "XML":
+                return DataType.STRING_TYPE;
+            case "NUMERIC":
+                return DataType.BIG_DECIMAL_TYPE;
+            case "CIDR":
+            case "INET":
+            case "JSONPATH":
+            case "CIRCLE":
+            case "POINT":
+            case "LINE":
+            case "BOX":
+            case "PATH":
+            case "POLYGON":
+            case "LSEG":
+            case "VARBIT":
+                return DataType.OBJECT;
+            default:
+                return super.convert(originType);
+        }
+    }
+
+    @Override
+    public String convertToOriginType(DataType dataType) {
+        return  super.convertToOriginType(dataType);
+    }
+}
diff --git a/...dbc/src/main/java/io/datavines/engine/local/connector/executor/ErrorDataSinkExecutor.java b/...dbc/src/main/java/io/datavines/engine/local/connector/executor/ErrorDataSinkExecutor.java
@@ -271,6 +271,11 @@ private void sinkErrorData() {
                                     }
                                     break;
                                 case OBJECT:
+                                    if (StringUtils.isNotEmpty(rowContent)) {
+                                        errorDataPreparedStatement.setObject(j+1, rowContent);
+                                    } else {
+                                        errorDataPreparedStatement.setNull(j+1, Types.JAVA_OBJECT);
+                                    }
                                     break;
                                 default:
                                     break;

diff --git a/datavines-runner/pom.xml b/datavines-runner/pom.xml
@@ -110,6 +110,10 @@
             <artifactId>mysql-connector-java</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.datavines</groupId>
             <artifactId>datavines-notification-plugin-email</artifactId>