commit 0e9a69bdefe6d1f4acb0605ba00ab21f3605df16
Author: feichanghong.fch <feichanghong.fch@alibaba-inc.com>
Date:   Fri Sep 27 01:05:22 2024 +0800

    support polardbo-cdc for polardb-oracle

diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml
index b0763b1d..55aea0d8 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml
@@ -69,6 +69,7 @@ limitations under the License.
             </exclusions>
         </dependency>
 
+        <!-- postgresql jdbc dependencies -->
         <dependency>
             <!-- fix CVE-2022-26520 https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-26520  -->
             <groupId>org.postgresql</groupId>
@@ -76,6 +77,12 @@ limitations under the License.
             <version>42.5.1</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.checkerframework</groupId>
+            <artifactId>checker-qual</artifactId>
+            <version>3.5.0</version>
+        </dependency>
+
         <!-- test dependencies on Debezium -->
 
         <dependency>
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
index f6e1e562..9b0e57be 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java
@@ -20,6 +20,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;
@@ -115,6 +116,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;
@@ -128,6 +136,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);
 
@@ -169,6 +185,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);
@@ -197,6 +222,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-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
index e0e1b6b9..be5ab4dc 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
@@ -71,7 +71,7 @@ import static org.apache.flink.util.Preconditions.checkState;
 /** 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";
 
     @Override
     public DynamicTableSource createDynamicTableSource(DynamicTableFactory.Context context) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
index f6d6dbed..26d039b4 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java
@@ -302,6 +302,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;
@@ -898,6 +908,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 */
+
   /**
    * Since PG14 there is GUC_REPORT ParamStatus {@code in_hot_standby} which is set to "on"
    * when the server is in archive recovery or standby mode. In driver's lingo such server is called
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
index b17ce210..79b359f6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
@@ -158,6 +158,12 @@ public class QueryExecutorImpl extends QueryExecutorBase {
 
   private final AdaptiveFetchCache adaptiveFetchCache;
 
+  /* 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,
@@ -2149,6 +2155,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);
           }
 
@@ -2633,6 +2640,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);
@@ -2996,4 +3004,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-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
index 3588f464..68646770 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/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);
@@ -1564,7 +1566,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);
@@ -1586,6 +1589,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
@@ -1736,7 +1740,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);
@@ -2165,6 +2170,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, "
@@ -2517,6 +2528,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) "
@@ -2535,6 +2550,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, "
@@ -2566,6 +2586,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;
@@ -3037,6 +3058,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);
