[feat] add kingbase sink

[feat] add kingbase sink

[feat] add kingbase sink

Author: FlechazoW

kingbase/kingbase-sink/pom.xml

@@ -9,7 +9,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>kingbase-sink</artifactId>
+    <artifactId>sql.sink.kingbase</artifactId>
     <properties>
         <sql.sink.rdb.version>1.0-SNAPSHOT</sql.sink.rdb.version>
     </properties>
@@ -21,4 +21,66 @@
             <version>${sql.sink.rdb.version}</version>
         </dependency>
     </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.1</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
+                            <artifactSet>
+                                <excludes>
+
+                                </excludes>
+                            </artifactSet>
+                            <filters>
+                                <filter>
+                                    <artifact>*:*</artifact>
+                                    <excludes>
+                                        <exclude>META-INF/*.SF</exclude>
+                                        <exclude>META-INF/*.DSA</exclude>
+                                        <exclude>META-INF/*.RSA</exclude>
+                                    </excludes>
+                                </filter>
+                            </filters>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+
+            <plugin>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.2</version>
+                <executions>
+                    <execution>
+                        <id>copy-resources</id>
+                        <!-- here the phase you need -->
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <tasks>
+                                <copy todir="${basedir}/../../sqlplugins/kingbasesink">
+                                    <fileset dir="target/">
+                                        <include name="${project.artifactId}-${project.version}.jar"/>
+                                    </fileset>
+                                </copy>
+
+                                <move file="${basedir}/../../sqlplugins/kingbasesink/${project.artifactId}-${project.version}.jar"
+                                      tofile="${basedir}/../../sqlplugins/kingbasesink/${project.name}-${git.branch}.jar"/>
+                            </tasks>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
 </project>

kingbase/kingbase-sink/src/main/java/com/dtstack/flink/sql/sink/kingbase/KingbaseDialect.java

@@ -61,11 +61,15 @@ public Optional<String> buildDuplicateUpsertStatement(String tableName,
                                                           String[] fieldNames,
                                                           String[] uniqueKeyFields) {
         String updateClause = Arrays.stream(fieldNames)
-                .map(f -> quoteIdentifier(f) + "=IFNULL(VALUES(" + quoteIdentifier(f) + ")," + quoteIdentifier(f) + ")")
+                .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
+                .collect(Collectors.joining(", "));
+
+        String uniqueColumns = Arrays.stream(uniqueKeyFields)
+                .map(this::quoteIdentifier)
                 .collect(Collectors.joining(", "));
 
         return Optional.of(getInsertIntoStatement("", tableName, fieldNames, null) +
-                " ON CONFLICT (" + Arrays.toString(uniqueKeyFields) + ") DO UPDATE SET " + updateClause
+                " ON CONFLICT (" + uniqueColumns + ") DO UPDATE SET " + updateClause
         );
     }
 }

kingbase/kingbase-sink/src/main/java/com/dtstack/flink/sql/sink/kingbase/table/KingbaseSinkParser.java

@@ -18,10 +18,24 @@
 
 package com.dtstack.flink.sql.sink.kingbase.table;
 
+import com.dtstack.flink.sql.sink.rdb.table.RdbSinkParser;
+import com.dtstack.flink.sql.table.AbstractTableInfo;
+
+import java.util.Map;
+
 /**
  * Date: 2020/09/10
  * Company: www.dtstack.com
+ *
  * @author tiezhu
  */
-public class KingbaseSinkParser  {
+public class KingbaseSinkParser extends RdbSinkParser {
+    private static final String CURRENT_TYPE = "kingbase";
+
+    @Override
+    public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+        AbstractTableInfo kingbaseTableInfo = super.getTableInfo(tableName, fieldsInfo, props);
+        kingbaseTableInfo.setType(CURRENT_TYPE);
+        return kingbaseTableInfo;
+    }
 }