[Bug] [connector-v2] PostgreSQL versions below 9.5 are compatible use…

… cdc sync problem add jdbc doc add postgresql 9.5 version below test
apache · Jul 20, 2023 · afd5ff7 · afd5ff7
1 parent 725ca9b
commit afd5ff7
Show file tree

Hide file tree

Showing 4 changed files with 80 additions and 1 deletion.
diff --git a/docs/en/connector-v2/sink/Jdbc.md b/docs/en/connector-v2/sink/Jdbc.md
@@ -74,6 +74,8 @@ Use this sql write upstream input datas to database. e.g `INSERT ...`
 
 The compatible mode of database, required when the database supports multiple compatible modes. For example, when using OceanBase database, you need to set it to 'mysql' or 'oracle'.
 
+Postgres 9.5 version or below,please set it to `postgresLow` to support cdc
+
 ### database [string]
 
 Use this `database` and `table-name` auto-generate sql and receive upstream input datas write to database.
@@ -226,6 +228,26 @@ sink {
 }
 ```
 
+Postgresql 9.5 version below support CDC(Change data capture) event
+
+```
+sink {
+    jdbc {
+        url = "jdbc:postgresql://localhost:5432"
+        driver = "org.postgresql.Driver"
+        user = "root"
+        password = "123456"
+        compatible_mode="postgresLow"
+        database = "sink_database"
+        table = "sink_table"
+        support_upsert_by_query_primary_key_exist = true
+        generate_sink_sql = true
+        primary_keys = ["key1", "key2", ...]
+    }
+}
+
+```
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

diff --git a/...che/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresDialectFactory.java b/...che/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresDialectFactory.java
@@ -19,9 +19,9 @@
 
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psqllow.PostgresLowDialect;
 
 import com.google.auto.service.AutoService;
-import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psqllow.PostgresLowDialect;
 
 import javax.annotation.Nonnull;
 
@@ -37,6 +37,7 @@ public JdbcDialect create() {
         throw new UnsupportedOperationException(
                 "Can't create JdbcDialect without compatible mode for Postgres");
     }
+
     @Override
     public JdbcDialect create(@Nonnull String compatibleMode) {
         if ("postgresLow".equalsIgnoreCase(compatibleMode)) {

diff --git a/...ache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psqllow/PostgresLowDialect.java b/...ache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psqllow/PostgresLowDialect.java
@@ -1,3 +1,20 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psqllow;
 
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psql.PostgresDialect;

diff --git a/...ache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/PostgresDialectFactoryTest.java b/...ache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/PostgresDialectFactoryTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psql.PostgresDialectFactory;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+public class PostgresDialectFactoryTest {
+
+    @Test
+    public void testPostgresDialectCreate() {
+        PostgresDialectFactory postgresDialectFactory = new PostgresDialectFactory();
+        JdbcDialect postgresLow = postgresDialectFactory.create("postgresLow");
+        String[] fields = {"id", "name", "age"};
+        String[] uniqueKeyField = {"id"};
+        Optional<String> upsertStatement =
+                postgresLow.getUpsertStatement("test", "test_a", fields, uniqueKeyField);
+        System.out.println(upsertStatement);
+    }
+}