[Connector/JDBC] Support passing arbitrary database options to JDBC Catalog

[och5351]

---

(Note: I am currently waiting for my Jira account registration to be approved, so I am unable to create a ticket for this issue at the moment. I will update the title with the Jira ticket number as soon as I gain access. If someone else would like to create it, that would also be great.)

---

1. Motivation

Currently, when creating a JDBC Catalog in Flink SQL, there is no way to pass arbitrary connection properties to the JDBC driver. This is a significant limitation compared to JDBC Dynamic Tables, which support a generic properties option.

This missing feature can lead to connection failures for certain databases that require specific driver properties. A common example is connecting to a MySQL or MariaDB server with a non-UTC timezone (e.g., 'KST'), which results in the following error:

com.mysql.cj.exceptions.InvalidConnectionAttributeException: The server time zone value 'KST' is unrecognized or represents more than one time zone. You must configure either the server or JDBC driver (via the serverTimezone configuration property) to use a more specific time zone value if you want to utilize time zone support.

To resolve this, the user needs to be able to pass properties like serverTimezone=UTC. Similarly, other use cases, such as passing stringtype=unspecified for PostgreSQL, are not possible with the current JDBC Catalog implementation.

Previous attempts to address this (e.g., PR #74, PR #83) were initiated but not completed. This PR provides a complete and robust solution to this long-standing issue.

2. Solution

This PR introduces a new configuration option, database-options, to the JdbcCatalogFactory. This option allows users to specify a string of key-value pairs that will be appended to the JDBC connection URL.

• For most databases, the options are appended as a query string, starting with ? for the first parameter and using & for subsequent ones.
• For DB2, which uses a semicolon (;) as a separator, the implementation correctly handles this specific syntax.
  This approach provides a flexible and universal way to configure any required JDBC driver property without adding database-specific options.

3. Implementation Details

The changes were implemented as follows:

1. JdbcCatalogFactory.java:

   A new ConfigOption named database-options was added.
   This option is registered in optionalOptions() to make it non-mandatory.

2. AbstractJdbcCatalog.java:

   The getDatabaseUrl() method was modified to check for the presence of database-options.
   If the options exist, they are appended to the base JDBC URL using the correct separator (? or ; based on the database dialect).

3. Factories and Loaders:

   The database-options value is now passed through the factory chain: JdbcCatalogFactory -> JdbcFactoryLoader -> database-specific factories (MySqlFactory, PostgresFactory, etc.).
   The method signatures in JdbcFactory.java and JdbcFactoryLoader.java were overloaded to accept the new dbOptions parameter.

4. Database-Specific Implementations:

   All existing database-specific catalog implementations (MySQL, PostgreSQL, Derby, OceanBase, etc.) and their corresponding factories were updated to accept and pass the dbOptions parameter to the AbstractJdbcCatalog constructor.

5. Usage Example

   With this change, a user can now create a JDBC catalog for MySQL and specify the required timezone property as follows:

CREATE CATALOG my_mariadb_catalog WITH (
  'type' = 'jdbc',
  'base-url' = 'jdbc:mysql://localhost:3306',
  'username' = 'user',
  'password' = 'pass',
  'default-database' = 'my_db',
  'database-options' = 'serverTimezone=UTC&tinyInt1isBit=false'
);

6. Verification

   The changes have been tested locally by creating a JDBC catalog for a MariaDB database that requires the serverTimezone property.

mysql

oceanbase

postgresql

8. Related Issues/PRs

   JIRA: FLINK-38616
   Stale PRs: [FLINK-33069]Mysql and Postgres catalog support url extra parameters #74, [FLINK-33800][JDBC/Connector] Allow passing parameters to database via jdbc url #83