commit 75758d4139a4e3d7de18e2c6407c404e613feeb4
Author: feichanghong.fch <feichanghong.fch@alibaba-inc.com>
Date:   Fri Sep 27 15:31:20 2024 +0800

    support polardbo-cdc for polardb-oracle

diff --git a/flink-connector-postgres-cdc/pom.xml b/flink-connector-postgres-cdc/pom.xml
index f253eee..1308a86 100644
--- a/flink-connector-postgres-cdc/pom.xml
+++ b/flink-connector-postgres-cdc/pom.xml
@@ -61,6 +61,11 @@ under the License.
             <artifactId>postgresql</artifactId>
             <version>42.2.26</version>
         </dependency>
+        <dependency>
+            <groupId>org.checkerframework</groupId>
+            <artifactId>checker-qual</artifactId>
+            <version>3.5.0</version>
+        </dependency>
         <!-- test dependencies on Debezium -->
 
         <dependency>
@@ -159,6 +164,11 @@ under the License.
             <version>${slf4j.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
 
     </dependencies>
 
diff --git a/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
index 624ce26..540cac2 100644
--- a/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
+++ b/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
@@ -38,7 +38,7 @@ import static org.apache.flink.util.Preconditions.checkArgument;
 /** Factory for creating configured instance of {@link PostgreSQLTableSource}. */
 public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
 
-    private static final String IDENTIFIER = "postgres-cdc";
+    private static final String IDENTIFIER = "polardbo-cdc";
 
     private static final ConfigOption<String> HOSTNAME =
             ConfigOptions.key("hostname")
diff --git a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
index 4489c75..a6ba395 100644
--- a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
+++ b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
@@ -21,6 +21,7 @@ import java.util.Set;
 
 import org.apache.kafka.connect.errors.ConnectException;
 import org.postgresql.core.BaseConnection;
+import org.postgresql.core.Oid; // POLAR DIFF
 import org.postgresql.core.TypeInfo;
 import org.postgresql.jdbc.PgDatabaseMetaData;
 import org.slf4j.Logger;
@@ -117,6 +118,13 @@ public class TypeRegistry {
     private int citextArrayOid = Integer.MIN_VALUE;
     private int ltreeArrayOid = Integer.MIN_VALUE;
 
+    /* POLAR DIFF */
+    private boolean mappingDateToTimestamp = false;
+    public static final int ORADATE = 9002;
+    public static final int ORADATE_ARRAY_V1 = 9005;
+    public static final int ORADATE_ARRAY_V2 = 9008;
+    /* POLAR end */
+
     public TypeRegistry(PostgresConnection connection) {
         try {
             this.connection = connection.connection();
@@ -131,6 +139,14 @@ public class TypeRegistry {
     }
 
     private void addType(PostgresType type) {
+        /* POLAR DIFF */
+        int oid = type.getOid();
+        if (oid == ORADATE || oid == ORADATE_ARRAY_V1 || oid == ORADATE_ARRAY_V2) {
+            mappingDateToTimestamp = true;
+            return;
+        }
+        /* POLAR end */
+
         oidToType.put(type.getOid(), type);
         nameToType.put(type.getName(), type);
 
@@ -172,6 +188,15 @@ public class TypeRegistry {
      * @return type associated with the given OID
      */
     public PostgresType get(int oid) {
+        /* POLAR DIFF */
+        if (oid == ORADATE) {
+            oid = Oid.TIMESTAMP;
+        }
+        if (oid == ORADATE_ARRAY_V1 || oid == ORADATE_ARRAY_V2) {
+            oid = Oid.TIMESTAMP_ARRAY;
+        }
+        /* POLAR end */
+
         PostgresType r = oidToType.get(oid);
         if (r == null) {
             r = resolveUnknownType(oid);
@@ -200,6 +225,17 @@ public class TypeRegistry {
                 name = "int8";
                 break;
         }
+
+        /* POLAR DIFF */
+        if (mappingDateToTimestamp) {
+            if (name.equals("date")) {
+                name = "timestamp";
+            } else if (name.equals("_date")) {
+                name = "_timestamp";
+            }
+        }
+        /* POLAR end */
+
         String[] parts = name.split("\\.");
         if (parts.length > 1) {
             name = parts[1];
diff --git a/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java b/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
index 0ffc9e5..87ee8c3 100644
--- a/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
+++ b/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
@@ -286,6 +286,16 @@ public class ConnectionFactoryImpl extends ConnectionFactory {
         }
 
         runInitialQueries(queryExecutor, info);
+        /* POLAR DIFF */
+        try {
+          // Lower versions of walsender will report syntax errors
+          runPolarInitialQueries(queryExecutor);
+        } catch (org.postgresql.util.PSQLException e) {
+          if (!e.getMessage().contains("syntax error")) {
+            LOGGER.log(Level.WARNING, "runPolarInitialQueries failed, due to ", e);
+          }
+        }
+        /* POLAR END */
 
         // And we're done.
         return queryExecutor;
@@ -841,6 +851,37 @@ public class ConnectionFactoryImpl extends ConnectionFactory {
 
   }
 
+  /*
+   * POLAR DIFF
+   * After establishing the connection, set some PolarDB-specific parameters by using
+   * 'select pg_catalog.set_config ... where name = ...' to avoid errors caused by the
+   * StartupMessage not recognizing these parameters.
+   * The old version of the replication process will still raise an error, so it is
+   * necessary to catch exceptions in the outer layer.
+   */
+  private void runPolarInitialQueries(QueryExecutor queryExecutor) throws SQLException {
+    /*
+     * Reset nls_xxx_format like dateStyle to make sys.date\timestamp\timestamptz data
+     * correct Setting this via SQL instead of the standard StartupMessages (like dateStyle) is to
+     * ensure compatibility with older versions.
+     */
+    String paramSetSql =
+            "SELECT pg_catalog.set_config(name,boot_val,'f') FROM pg_catalog.pg_settings WHERE name = '%s' ";
+    SetupQueryRunner.run(queryExecutor, String.format(paramSetSql, "nls_date_format"), false);
+    SetupQueryRunner.run(
+            queryExecutor, String.format(paramSetSql, "nls_timestamp_format"), false);
+    SetupQueryRunner.run(
+            queryExecutor, String.format(paramSetSql, "nls_timestamp_tz_format"), false);
+    /* Set polar_prohibit_rowid_logical on. */
+    paramSetSql =
+            "SELECT pg_catalog.set_config(name,'%s','f') FROM pg_catalog.pg_settings WHERE name = '%s' ";
+    SetupQueryRunner.run(
+            queryExecutor,
+            String.format(paramSetSql, "on", "polar_prohibit_rowid_logical"),
+            false);
+  }
+  /* POLAR END */
+
   private boolean isPrimary(QueryExecutor queryExecutor) throws SQLException, IOException {
     Tuple results = SetupQueryRunner.run(queryExecutor, "show transaction_read_only", true);
     Tuple nonNullResults = castNonNull(results);
diff --git a/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java b/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
index 4d656b0..a1461ea 100644
--- a/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
+++ b/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
@@ -134,6 +134,12 @@ public class QueryExecutorImpl extends QueryExecutorBase {
    */
   private final CommandCompleteParser commandCompleteParser = new CommandCompleteParser();
 
+  /* POLAR DIFF */
+  public static final int ORADATE = 9002;
+  public static final int ORADATE_ARRAY_V1 = 9005;
+  public static final int ORADATE_ARRAY_V2 = 9008;
+  /* POLAR end */
+
   @SuppressWarnings({"assignment.type.incompatible", "argument.type.incompatible",
       "method.invocation.invalid"})
   public QueryExecutorImpl(PGStream pgStream, String user, String database,
@@ -2098,6 +2104,7 @@ public class QueryExecutorImpl extends QueryExecutorBase {
 
           for (int i = 1; i <= numParams; i++) {
             int typeOid = pgStream.receiveInteger4();
+            typeOid = polarMapTypeOid(typeOid); // POLAR DIFF: map sys.date to timestamp
             params.setResolvedType(i, typeOid);
           }
 
@@ -2522,6 +2529,7 @@ public class QueryExecutorImpl extends QueryExecutorBase {
       int typeLength = pgStream.receiveInteger2();
       int typeModifier = pgStream.receiveInteger4();
       int formatType = pgStream.receiveInteger2();
+      typeOid = polarMapTypeOid(typeOid); // POLAR DIFF: map sys.date to timestamp
       fields[i] = new Field(columnLabel,
           typeOid, typeLength, typeModifier, tableOid, positionInTable);
       fields[i].setFormat(formatType);
@@ -2882,4 +2890,23 @@ public class QueryExecutorImpl extends QueryExecutorBase {
       new SimpleQuery(
           new NativeQuery("ROLLBACK TO SAVEPOINT PGJDBC_AUTOSAVE", new int[0], false, SqlCommand.BLANK),
           null, false);
+
+  /* POLAR DIFF: map sys.date to timestamp */
+  public static int polarMapTypeOid(int typeoid) {
+    switch (typeoid) {
+      /* Treat SYS.DATE as TIMESTAMP */
+      case ORADATE:
+        typeoid = Oid.TIMESTAMP;
+        break;
+      /* Treat SYS.DATE_ARRAY as TIMESTAMP_ARRAY */
+      case ORADATE_ARRAY_V1:
+      case ORADATE_ARRAY_V2:
+        typeoid = Oid.TIMESTAMP_ARRAY;
+        break;
+      default:
+        break;
+    }
+    return typeoid;
+  }
+  /* POLAR end */
 }
diff --git a/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java b/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
index 42a54c4..9ee6abe 100644
--- a/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
+++ b/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
@@ -13,6 +13,7 @@ import org.postgresql.core.Oid;
 import org.postgresql.core.ServerVersion;
 import org.postgresql.core.Tuple;
 import org.postgresql.core.TypeInfo;
+import org.postgresql.core.v3.QueryExecutorImpl;
 import org.postgresql.util.ByteConverter;
 import org.postgresql.util.GT;
 import org.postgresql.util.JdbcBlackHole;
@@ -1226,6 +1227,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
       if ("c".equals(returnTypeType) || ("p".equals(returnTypeType) && argModesArray != null)) {
         String columnsql = "SELECT a.attname,a.atttypid FROM pg_catalog.pg_attribute a "
                            + " WHERE a.attrelid = " + returnTypeRelid
+                           + " AND a.attname <> 'polar_sys_rowid_attr' AND a.attname <> 'polarSeqRowid' " // POLAR DIFF
                            + " AND NOT a.attisdropped AND a.attnum > 0 ORDER BY a.attnum ";
         Statement columnstmt = connection.createStatement();
         ResultSet columnrs = columnstmt.executeQuery(columnsql);
@@ -1557,7 +1559,8 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
            + " LEFT JOIN pg_catalog.pg_description dsc ON (c.oid=dsc.objoid AND a.attnum = dsc.objsubid) "
            + " LEFT JOIN pg_catalog.pg_class dc ON (dc.oid=dsc.classoid AND dc.relname='pg_class') "
            + " LEFT JOIN pg_catalog.pg_namespace dn ON (dc.relnamespace=dn.oid AND dn.nspname='pg_catalog') "
-           + " WHERE c.relkind in ('r','p','v','f','m') and a.attnum > 0 AND NOT a.attisdropped ";
+           + " WHERE c.relkind in ('r','p','v','f','m') and a.attnum > 0 AND NOT a.attisdropped "
+           + " AND a.attname <> 'polar_sys_rowid_attr' AND a.attname <> 'polarSeqRowid' "; // POLAR DIFF
 
     if (schemaPattern != null && !schemaPattern.isEmpty()) {
       sql += " AND n.nspname LIKE " + escapeQuotes(schemaPattern);
@@ -1579,6 +1582,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
       byte[] @Nullable [] tuple = new byte[numberOfFields][];
       int typeOid = (int) rs.getLong("atttypid");
       int typeMod = rs.getInt("atttypmod");
+      typeOid = QueryExecutorImpl.polarMapTypeOid(typeOid); // POLAR DIFF
 
       tuple[0] = null; // Catalog name, not supported
       tuple[1] = rs.getBytes("nspname"); // Schema
@@ -1724,7 +1728,8 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
           + " AND c.relowner = r.oid "
           + " AND c.oid = a.attrelid "
           + " AND c.relkind = 'r' "
-          + " AND a.attnum > 0 AND NOT a.attisdropped ";
+          + " AND a.attnum > 0 AND NOT a.attisdropped "
+          + " AND a.attname <> 'polar_sys_rowid_attr' AND a.attname <> 'polarSeqRowid' "; // POLAR DIFF
 
     if (schema != null && !schema.isEmpty()) {
       sql += " AND n.nspname = " + escapeQuotes(schema);
@@ -2153,6 +2158,12 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
       sql += " AND ct.relname = " + escapeQuotes(table);
     }
 
+    /* POLAR DIFF: support rowid and global index */
+    sql += " AND ci.relname <> 'polar_rowid_' || ct.oid || '_index' ";
+    sql += " AND ci.relname <> 'pg_oid_' || ct.oid || '_index' ";
+    sql += " AND ci.relname <> 'Polar_Rowid_' || ct.oid || '_index' ";
+    sql += " AND NOT ('global_index=true' = ANY(ci.reloptions) AND a.attname = 'tableoid') ";
+    /* POLAR END */
     sql += " AND i.indisprimary ";
     sql = "SELECT "
             + "       result.TABLE_CAT, "
@@ -2505,6 +2516,10 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
             + "  pg_catalog.pg_get_expr(i.indpred, i.indrelid) AS FILTER_CONDITION, "
             + "  ci.oid AS CI_OID, "
             + "  i.indoption AS I_INDOPTION, "
+            /* POLAR DIFF: support global index */
+            + "  ('global_index=true' = ANY(ci.reloptions)) AS IS_GLOBAL_INDEX, "
+            + "  (information_schema._pg_expandarray(i.indkey)).x AS COLUMN_ATTNUM, "
+            /* POLAR END */
             + (connection.haveMinimumServerVersion(ServerVersion.v9_6) ? "  am.amname AS AM_NAME " : "  am.amcanorder AS AM_CANORDER ")
             + "FROM pg_catalog.pg_class ct "
             + "  JOIN pg_catalog.pg_namespace n ON (ct.relnamespace = n.oid) "
@@ -2523,6 +2538,11 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
         sql += " AND i.indisunique ";
       }
 
+      /* POLAR DIFF: support rowid */
+      sql += " AND ci.relname <> 'polar_rowid_' || ct.oid || '_index' ";
+      sql += " AND ci.relname <> 'pg_oid_' || ct.oid || '_index' ";
+      sql += " AND ci.relname <> 'Polar_Rowid_' || ct.oid || '_index' ";
+      /* POLAR END */
       sql = "SELECT "
                 + "    tmp.TABLE_CAT, "
                 + "    tmp.TABLE_SCHEM, "
@@ -2554,6 +2574,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
                 + "FROM ("
                 + sql
                 + ") AS tmp";
+      sql += " WHERE NOT (tmp.IS_GLOBAL_INDEX AND tmp.COLUMN_ATTNUM = -6) "; /* POLAR DIFF: support global index */
     } else {
       String select;
       String from;
@@ -3025,6 +3046,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
       if ("c".equals(returnTypeType) || ("p".equals(returnTypeType) && argModesArray != null)) {
         String columnsql = "SELECT a.attname,a.atttypid FROM pg_catalog.pg_attribute a "
             + " WHERE a.attrelid = " + returnTypeRelid
+            + " AND a.attname <> 'polar_sys_rowid_attr' AND a.attname <> 'polarSeqRowid' " // POLAR DIFF
             + " AND NOT a.attisdropped AND a.attnum > 0 ORDER BY a.attnum ";
         Statement columnstmt = connection.createStatement();
         ResultSet columnrs = columnstmt.executeQuery(columnsql);
