[FLINK-34572] Support OceanBase Jdbc Catalog

docs/content.zh/docs/connectors/table/jdbc.md

@@ -432,10 +432,9 @@ JDBC Catalog
 
 `JdbcCatalog` 允许用户通过 JDBC 协议将 Flink 连接到关系数据库。
 
-目前，JDBC Catalog 有两个实现，即 Postgres Catalog 和 MySQL Catalog。目前支持如下 catalog 方法。其他方法目前尚不支持。
+目前，JDBC Catalog 有四个实现：Postgres Catalog、MySQL Catalog、CrateDB Catalog 和 OceanBase Catalog。目前支持如下 catalog 方法。其他方法目前尚不支持。
 
 ```java
-// Postgres Catalog & MySQL Catalog 支持的方法
 databaseExists(String databaseName);
 listDatabases();
 getDatabase(String databaseName);
@@ -450,17 +449,19 @@ tableExists(ObjectPath tablePath);
 
 ### JDBC Catalog 的使用
 
-本小节主要描述如果创建并使用 Postgres Catalog 或 MySQL Catalog。
+本小节主要描述如何创建并使用 JDBC Catalog。
 请参阅 [依赖](#依赖) 部分了解如何配置 JDBC 连接器和相应的驱动。
 
 JDBC catalog 支持以下参数:
 - `name`：必填，catalog 的名称。
 - `default-database`：必填，默认要连接的数据库。
-- `username`：必填，Postgres/MySQL 账户的用户名。
+- `username`：必填，数据库账户的用户名。
 - `password`：必填，账户的密码。
 - `base-url`：必填，（不应该包含数据库名）
   - 对于 Postgres Catalog `base-url` 应为 `"jdbc:postgresql://<ip>:<port>"` 的格式。
   - 对于 MySQL Catalog `base-url` 应为 `"jdbc:mysql://<ip>:<port>"` 的格式。
+  - 对于 OceanBase Catalog `base-url` 应为 `"jdbc:oceanbase://<ip>:<port>"` 的格式。
+- `compatible-mode`: 选填，数据库的兼容模式。
 
 {{< tabs "10bd8bfb-674c-46aa-8a36-385537df5791" >}}
 {{< tab "SQL" >}}
@@ -656,6 +657,42 @@ SELECT * FROM mycatalog.crate.`custom_schema.test_table2`
 SELECT * FROM crate.`custom_schema.test_table2`;
 SELECT * FROM `custom_schema.test_table2`;
 ```
+
+<a name="jdbc-catalog-for-oceanbase"></a>
+
+### JDBC Catalog for OceanBase
+
+<a name="oceanbase-metaspace-mapping"></a>
+
+#### OceanBase 元空间映射
+
+OceanBase 数据库支持多租户管理，每个租户可以工作在 MySQL 兼容模式或 Oracle 兼容模式。在 OceanBase 的 MySQL 模式上，一个租户中有数据库和表，就像 MySQL 数据库中的数据库和表一样，但没有 schema。在 OceanBase 的 Oracle 模式下，一个租户中有 schema 和表，就像 Oracle 数据库中的 schema 和表一样，但没有数据库。
+
+在 Flink 中，查询 OceanBase Catalog 注册的表时，OceanBase MySQL 模式下可以使用 `database.table_name` 或只使用 `table_name`，OceanBase Oracle 模式下可以使用 `schema.table_name` 或只使用 `table_name`。
+
+因此，Flink Catalog 和 OceanBase catalog 之间的元空间映射如下：
+
+| Flink Catalog Metaspace Structure    | OceanBase Metaspace Structure (MySQL Mode) | OceanBase Metaspace Structure (Oracle Mode) |
+|:-------------------------------------|:-------------------------------------------|---------------------------------------------|
+| catalog name (defined in Flink only) | N/A                                        | N/A                                         |
+| database name                        | database name                              | schema name                                 |
+| table name                           | table name                                 | table name                                  |
+
+Flink 中的 OceanBase 表的完整路径应该是 ``"`<catalog>`.`<db or schema>`.`<table>`"``。
+
+这里提供了一些访问 OceanBase 表的例子：
+
+```sql
+-- 扫描 默认数据库 'mydb' 中的 'test_table' 表
+SELECT * FROM oceanbase_catalog.mydb.test_table;
+SELECT * FROM mydb.test_table;
+SELECT * FROM test_table;
+
+-- 扫描 'given_database' 数据库中的 'test_table2' 表，
+SELECT * FROM oceanbase_catalog.given_database.test_table2;
+SELECT * FROM given_database.test_table2;
+```
+
 <a name="data-type-mapping"></a>
 
 数据类型映射

docs/content/docs/connectors/table/jdbc.md

@@ -441,10 +441,9 @@ JDBC Catalog
 
 The `JdbcCatalog` enables users to connect Flink to relational databases over JDBC protocol.
 
-Currently, there are two JDBC catalog implementations, Postgres Catalog and MySQL Catalog. They support the following catalog methods. Other methods are currently not supported.
+Currently, there are following JDBC catalog implementations: Postgres Catalog, MySQL Catalog, CrateDB Catalog and OceanBase Catalog. They support the following catalog methods. Other methods are currently not supported.
 
 ```java
-// The supported methods by Postgres & MySQL Catalog.
 databaseExists(String databaseName);
 listDatabases();
 getDatabase(String databaseName);
@@ -457,17 +456,19 @@ Other `Catalog` methods are currently not supported.
 
 ### Usage of JDBC Catalog
 
-The section mainly describes how to create and use a Postgres Catalog or MySQL Catalog.
+The section mainly describes how to create and use a JDBC Catalog.
 Please refer to [Dependencies](#dependencies) section for how to setup a JDBC connector and the corresponding driver.
 
 The JDBC catalog supports the following options:
 - `name`: required, name of the catalog.
 - `default-database`: required, default database to connect to.
-- `username`: required, username of Postgres/MySQL account.
+- `username`: required, username of database account.
 - `password`: required, password of the account.
 - `base-url`: required (should not contain the database name)
   - for Postgres Catalog this should be `"jdbc:postgresql://<ip>:<port>"`
   - for MySQL Catalog this should be `"jdbc:mysql://<ip>:<port>"`
+  - for OceanBase Catalog this should be `jdbc:oceanbase://<ip>:<port>`
+- `compatible-mode`: optional, the compatible mode of database.
 
 {{< tabs "10bd8bfb-674c-46aa-8a36-385537df5791" >}}
 {{< tab "SQL" >}}
@@ -654,6 +655,37 @@ SELECT * FROM crate.`custom_schema.test_table2`;
 SELECT * FROM `custom_schema.test_table2`;
 ```
 
+### JDBC Catalog for OceanBase
+
+#### OceanBase Metaspace Mapping
+
+OceanBase database supports multiple tenant management, and each tenant can work at MySQL compatible mode or Oracle compatible mode. On MySQL mode of OceanBase, there are databases and tables but no schema in one tenant, these objects just like databases and tables in the MySQL database. On Oracle mode of OceanBase, there are schemas and tables but no database in one tenant, these objects just like schemas and tables in the Oracle database.
+
+In Flink, when querying tables registered by OceanBase Catalog, users can use either `database.table_name` or just `table_name` on OceanBase MySQL mode, or use either `schema.table_name` or just `table_name` on OceanBase Oracle mode.
+
+Therefore, the metaspace mapping between Flink Catalog and OceanBase is as following:
+
+| Flink Catalog Metaspace Structure    | OceanBase Metaspace Structure (MySQL Mode) | OceanBase Metaspace Structure (Oracle Mode) |
+|:-------------------------------------|:-------------------------------------------|---------------------------------------------|
+| catalog name (defined in Flink only) | N/A                                        | N/A                                         |
+| database name                        | database name                              | schema name                                 |
+| table name                           | table name                                 | table name                                  |
+
+The full path of OceanBase table in Flink should be "`<catalog>`.`<db or schema>`.`<table>`".
+
+Here are some examples to access OceanBase tables:
+
+```sql
+-- scan table 'test_table', the default database or schema is 'mydb'.
+SELECT * FROM oceanbase_catalog.mydb.test_table;
+SELECT * FROM mydb.test_table;
+SELECT * FROM test_table;
+
+-- scan table 'test_table' with the given database or schema.
+SELECT * FROM oceanbase_catalog.given_database.test_table2;
+SELECT * FROM given_database.test_table2;
+```
+
 Data Type Mapping
 ----------------
 Flink supports connect to several databases which uses dialect like MySQL, Oracle, PostgreSQL, CrateDB, Derby, SQL Server, Db2 and OceanBase. The Derby dialect usually used for testing purpose. The field data type mappings from relational databases data types to Flink SQL data types are listed in the following table, the mapping table can help define JDBC table in Flink easily.

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.connector.jdbc.catalog;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory;
 import org.apache.flink.table.api.Schema;
 import org.apache.flink.table.api.ValidationException;
@@ -87,6 +88,7 @@
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** Abstract catalog for any JDBC catalogs. */
+@PublicEvolving
 public abstract class AbstractJdbcCatalog extends AbstractCatalog {
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcCatalog.class);
@@ -298,20 +300,23 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
             primaryKey.ifPresent(
                     pk -> schemaBuilder.primaryKeyNamed(pk.getName(), pk.getColumns()));
             Schema tableSchema = schemaBuilder.build();
-
-            Map<String, String> props = new HashMap<>();
-            props.put(CONNECTOR.key(), IDENTIFIER);
-            props.put(URL.key(), dbUrl);
-            props.put(USERNAME.key(), connectionProperties.getProperty(USER_KEY));
-            props.put(PASSWORD.key(), connectionProperties.getProperty(PASSWORD_KEY));
-            props.put(TABLE_NAME.key(), getSchemaTableName(tablePath));
-            return CatalogTable.of(tableSchema, null, Lists.newArrayList(), props);
+            return CatalogTable.of(tableSchema, null, Lists.newArrayList(), getOptions(tablePath));
         } catch (Exception e) {
             throw new CatalogException(
                     String.format("Failed getting table %s", tablePath.getFullName()), e);
         }
     }
 
+    protected Map<String, String> getOptions(ObjectPath tablePath) {
+        Map<String, String> props = new HashMap<>();
+        props.put(CONNECTOR.key(), IDENTIFIER);
+        props.put(URL.key(), baseUrl + tablePath.getDatabaseName());
+        props.put(USERNAME.key(), connectionProperties.getProperty(USER_KEY));
+        props.put(PASSWORD.key(), connectionProperties.getProperty(PASSWORD_KEY));
+        props.put(TABLE_NAME.key(), getSchemaTableName(tablePath));
+        return props;
+    }
+
     @Override
     public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
             throws TableNotExistException, CatalogException {

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java

@@ -18,12 +18,6 @@
 
 package org.apache.flink.connector.jdbc.catalog;
 
-import org.apache.flink.connector.jdbc.databases.cratedb.catalog.CrateDBCatalog;
-import org.apache.flink.connector.jdbc.databases.cratedb.dialect.CrateDBDialect;
-import org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog;
-import org.apache.flink.connector.jdbc.databases.mysql.dialect.MySqlDialect;
-import org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresCatalog;
-import org.apache.flink.connector.jdbc.databases.postgres.dialect.PostgresDialect;
 import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
 import org.apache.flink.connector.jdbc.dialect.JdbcDialectLoader;
 
@@ -73,18 +67,7 @@ public static AbstractJdbcCatalog createCatalog(
             Properties connectionProperties) {
         JdbcDialect dialect = JdbcDialectLoader.load(baseUrl, compatibleMode, userClassLoader);
 
-        if (dialect instanceof PostgresDialect) {
-            return new PostgresCatalog(
-                    userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
-        } else if (dialect instanceof CrateDBDialect) {
-            return new CrateDBCatalog(
-                    userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
-        } else if (dialect instanceof MySqlDialect) {
-            return new MySqlCatalog(
-                    userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
-        } else {
-            throw new UnsupportedOperationException(
-                    String.format("Catalog for '%s' is not supported yet.", dialect));
-        }
+        return dialect.createCatalog(
+                userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
     }
 }

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialect.java

@@ -18,10 +18,13 @@
 
 package org.apache.flink.connector.jdbc.databases.cratedb.dialect;
 
+import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
+import org.apache.flink.connector.jdbc.databases.cratedb.catalog.CrateDBCatalog;
 import org.apache.flink.connector.jdbc.dialect.AbstractPostgresCompatibleDialect;
 import org.apache.flink.table.types.logical.RowType;
 
 import java.util.Optional;
+import java.util.Properties;
 
 /** JDBC dialect for CrateDB. */
 public class CrateDBDialect extends AbstractPostgresCompatibleDialect {
@@ -33,6 +36,17 @@ public String dialectName() {
         return "CrateDB";
     }
 
+    @Override
+    public AbstractJdbcCatalog createCatalog(
+            ClassLoader userClassLoader,
+            String catalogName,
+            String defaultDatabase,
+            String baseUrl,
+            Properties connectionProperties) {
+        return new CrateDBCatalog(
+                userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
+    }
+
     @Override
     public CrateDBRowConverter getRowConverter(RowType rowType) {
         return new CrateDBRowConverter(rowType);

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialect.java

@@ -19,14 +19,17 @@
 package org.apache.flink.connector.jdbc.databases.mysql.dialect;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
 import org.apache.flink.connector.jdbc.converter.JdbcRowConverter;
+import org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog;
 import org.apache.flink.connector.jdbc.dialect.AbstractDialect;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.RowType;
 
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -50,6 +53,17 @@ public class MySqlDialect extends AbstractDialect {
     // https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-connp-props-performance-extensions.html#cj-conn-prop_rewriteBatchedStatements
     private static final String REWRITE_BATCHED_STATEMENTS = "rewriteBatchedStatements";
 
+    @Override
+    public AbstractJdbcCatalog createCatalog(
+            ClassLoader userClassLoader,
+            String catalogName,
+            String defaultDatabase,
+            String baseUrl,
+            Properties connectionProperties) {
+        return new MySqlCatalog(
+                userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
+    }
+
     @Override
     public JdbcRowConverter getRowConverter(RowType rowType) {
         return new MySQLRowConverter(rowType);