You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2024/07/29 01:49:28 UTC
(flink-connector-jdbc) 04/14: [FLINK-35365] Reorganize catalog code
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
commit d8e5665ce9c04058f5aab6c2e696fa0a6b038728
Author: Joao Boto <bo...@boto.pro>
AuthorDate: Tue May 7 10:33:46 2024 +0200
[FLINK-35365] Reorganize catalog code
---
.../2047daea-3713-4298-bdb3-e269f539363f | 303 ---------------------
.../archunit-violations/stored.rules | 11 +-
flink-connector-jdbc-core/pom.xml | 15 +-
.../connector/jdbc/core/table/JdbcFactory.java | 30 +-
.../jdbc/core/table/JdbcFactoryLoader.java | 60 ++--
.../table}/catalog/AbstractJdbcCatalog.java | 37 ++-
.../jdbc/core/table/catalog/JdbcCatalog.java | 10 +
.../table/catalog/JdbcCatalogTypeMapper.java} | 6 +-
.../table}/catalog/factory/JdbcCatalogFactory.java | 27 +-
.../catalog/factory/JdbcCatalogFactoryOptions.java | 4 +-
.../databases/cratedb/dialect/CrateDBFactory.java | 14 +
.../jdbc/databases/db2/dialect/Db2Factory.java | 12 +
.../jdbc/databases/derby/dialect/DerbyFactory.java | 12 +
.../jdbc/databases/mysql/catalog/MySqlCatalog.java | 6 +-
.../databases/mysql/catalog/MySqlTypeMapper.java | 4 +-
.../jdbc/databases/mysql/dialect/MySqlFactory.java | 13 +
.../oceanbase/dialect/OceanBaseFactory.java | 12 +
.../databases/oracle/dialect/OracleFactory.java | 12 +
.../postgres/catalog/PostgresCatalog.java | 10 +-
.../postgres/catalog/PostgresTypeMapper.java | 4 +-
.../postgres/dialect/PostgresFactory.java | 14 +
.../sqlserver/dialect/SqlServerFactory.java | 12 +
.../jdbc/databases/trino/dialect/TrinoFactory.java | 12 +
.../flink/connector/jdbc/utils/JdbcUtils.java | 4 +-
.../org.apache.flink.table.factories.Factory | 2 +-
.../table/catalog/AbstractJdbcCatalogTest.java} | 15 +-
.../table/dialect/JdbcDialectConverterTest.java} | 6 +-
.../table/dialect/JdbcDialectTest.java} | 4 +-
...ialectTypeTest.java => CrateDBDialectTest.java} | 6 +-
...Db2DialectTypeTest.java => Db2DialectTest.java} | 6 +-
.../databases/mysql/dialect/MySqlDialectTest.java | 46 +++-
.../mysql/dialect/MySqlDialectTypeTest.java | 65 -----
...ypeTest.java => OceanBaseMysqlDialectTest.java} | 8 +-
...peTest.java => OceanBaseOracleDialectTest.java} | 8 +-
...DialectTypeTest.java => OracleDialectTest.java} | 6 +-
.../catalog/factory/JdbcCatalogFactoryTest.java | 17 +-
...alectTypeTest.java => PostgresDialectTest.java} | 6 +-
...oDialectTypeTest.java => TrinoDialectTest.java} | 6 +-
flink-connector-jdbc/pom.xml | 42 +++
.../AbstractJdbcCatalog.java} | 25 +-
.../flink/connector/jdbc/catalog/JdbcCatalog.java | 13 +-
.../connector/jdbc/catalog/JdbcCatalogUtils.java | 37 ++-
.../jdbc/catalog/factory/JdbcCatalogFactory.java | 23 +-
.../jdbc/converter/AbstractJdbcRowConverter.java | 2 +-
.../jdbc/dialect/JdbcDialectTypeMapper.java | 19 +-
.../org.apache.flink.table.factories.Factory | 1 -
.../jdbc/dialect/JdbcDialectTypeTest.java | 20 +-
pom.xml | 9 +
48 files changed, 474 insertions(+), 562 deletions(-)
diff --git a/flink-connector-jdbc-core/archunit-violations/2047daea-3713-4298-bdb3-e269f539363f b/flink-connector-jdbc-core/archunit-violations/2047daea-3713-4298-bdb3-e269f539363f
deleted file mode 100644
index 662348fa..00000000
--- a/flink-connector-jdbc-core/archunit-violations/2047daea-3713-4298-bdb3-e269f539363f
+++ /dev/null
@@ -1,303 +0,0 @@
-Class <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog> extends class <org.apache.flink.table.catalog.AbstractCatalog> in (AbstractJdbcCatalog.java:0)
-Class <org.apache.flink.connector.jdbc.table.JdbcFilterPushdownPreparedStatementVisitor> extends class <org.apache.flink.table.expressions.ExpressionDefaultVisitor> in (JdbcFilterPushdownPreparedStatementVisitor.java:0)
-Constructor <org.apache.flink.connector.jdbc.JdbcConnectionOptions.<init>(java.lang.String, java.lang.String, int, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcConnectionOptions.java:50)
-Constructor <org.apache.flink.connector.jdbc.JdbcConnectionOptions.<init>(java.lang.String, java.lang.String, int, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcConnectionOptions.java:53)
-Constructor <org.apache.flink.connector.jdbc.JdbcConnectionOptions.<init>(java.lang.String, java.lang.String, int, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcConnectionOptions.java:57)
-Constructor <org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions.<init>(boolean, int, boolean, java.util.Optional, boolean)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcExactlyOnceOptions.java:75)
-Constructor <org.apache.flink.connector.jdbc.JdbcExecutionOptions.<init>(long, int, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (JdbcExecutionOptions.java:38)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls constructor <org.apache.flink.table.catalog.AbstractCatalog.<init>(java.lang.String, java.lang.String)> in (AbstractJdbcCatalog.java:121)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AbstractJdbcCatalog.java:124)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AbstractJdbcCatalog.java:132)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AbstractJdbcCatalog.java:134)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AbstractJdbcCatalog.java:123)
-Constructor <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AbstractJdbcCatalog.java:131)
-Constructor <org.apache.flink.connector.jdbc.catalog.JdbcCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcCatalog.java:0)
-Constructor <org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.<init>(org.apache.flink.table.types.logical.RowType)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AbstractJdbcRowConverter.java:60)
-Constructor <org.apache.flink.connector.jdbc.databases.cratedb.catalog.CrateDBCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (CrateDBCatalog.java:0)
-Constructor <org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (MySqlCatalog.java:0)
-Constructor <org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (MySqlCatalog.java:92)
-Constructor <org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (MySqlCatalog.java:94)
-Constructor <org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresCatalog.<init>(java.lang.ClassLoader, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (PostgresCatalog.java:0)
-Constructor <org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresTablePath.<init>(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (PostgresTablePath.java:41)
-Constructor <org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresTablePath.<init>(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (PostgresTablePath.java:44)
-Constructor <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.<init>(java.util.function.Supplier, java.lang.Integer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (SimpleXaConnectionProvider.java:81)
-Constructor <org.apache.flink.connector.jdbc.internal.GenericJdbcSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (GenericJdbcSinkFunction.java:44)
-Constructor <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.internal.JdbcOutputFormat$StatementExecutorFactory)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcOutputFormat.java:87)
-Constructor <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.internal.JdbcOutputFormat$StatementExecutorFactory)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcOutputFormat.java:88)
-Constructor <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.internal.JdbcOutputFormat$StatementExecutorFactory)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcOutputFormat.java:89)
-Constructor <org.apache.flink.connector.jdbc.internal.TableJdbcUpsertOutputFormat.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.internal.JdbcOutputFormat$StatementExecutorFactory, org.apache.flink.connector.jdbc.internal.JdbcOutputFormat$StatementExecutorFactory)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (TableJdbcUpsertOutputFormat.java:0)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:76)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:77)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:78)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:79)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:80)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:81)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:82)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.InsertOrUpdateJdbcExecutor.<init>(java.lang.String, java.lang.String, java.lang.String, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, org.apache.flink.connector.jdbc.JdbcStatementBuilder, java.util.function.Function, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (InsertOrUpdateJdbcExecutor.java:83)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, org.apache.flink.connector.jdbc.converter.JdbcRowConverter, java.util.function.Functi [...]
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (TableSimpleStatementExecutor.java:47)
-Constructor <org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor.<init>(org.apache.flink.connector.jdbc.statement.StatementFactory, org.apache.flink.connector.jdbc.converter.JdbcRowConverter)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (TableSimpleStatementExecutor.java:48)
-Constructor <org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions.<init>(java.lang.String, org.apache.flink.connector.jdbc.dialect.JdbcDialect, [Ljava.lang.String;, [I, [Ljava.lang.String;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcDmlOptions.java:51)
-Constructor <org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions.<init>(java.lang.String, org.apache.flink.connector.jdbc.dialect.JdbcDialect, [Ljava.lang.String;, [I, [Ljava.lang.String;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcDmlOptions.java:52)
-Constructor <org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions.<init>(java.lang.String, org.apache.flink.connector.jdbc.dialect.JdbcDialect, [Ljava.lang.String;, [I, [Ljava.lang.String;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcDmlOptions.java:53)
-Constructor <org.apache.flink.connector.jdbc.internal.options.JdbcInsertOptions.<init>(java.lang.String, [I)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcInsertOptions.java:33)
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.<init>(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.JdbcExecutionOptions, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement, org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer, org.apache.flink.connector.base.DeliveryGuarantee, java.util.Collection, org.apache.flink.a [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.JdbcSource.<init>(org.apache.flink.configuration.Configuration, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase$Provider, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink. [...]
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumStateSerializer.<init>(org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumStateSerializer.java:47)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumerator.<init>(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase, java.util.List)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumerator.java:54)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumerator.<init>(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase, java.util.List)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumerator.java:55)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumerator.<init>(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.jdbc.source.enumerator.JdbcSqlSplitEnumeratorBase, java.util.List)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumerator.java:57)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumeratorState.<init>(java.util.List, java.util.List, java.util.List, java.io.Serializable)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumeratorState.java:47)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumeratorState.<init>(java.util.List, java.util.List, java.util.List, java.io.Serializable)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumeratorState.java:48)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumeratorState.<init>(java.util.List, java.util.List, java.util.List, java.io.Serializable)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumeratorState.java:49)
-Constructor <org.apache.flink.connector.jdbc.source.enumerator.SqlTemplateSplitEnumerator.<init>(java.io.Serializable, java.lang.String, org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (SqlTemplateSplitEnumerator.java:47)
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.<init>(org.apache.flink.api.connector.source.SourceReaderContext, org.apache.flink.configuration.Configuration, org.apache.flink.api.common.typeinfo.TypeInformation, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider, org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preco [...]
-Constructor <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitState.<init>(org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceSplitState.java:38)
-Constructor <org.apache.flink.connector.jdbc.split.CompositeJdbcParameterValuesProvider.<init>(org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider, org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (CompositeJdbcParameterValuesProvider.java:34)
-Constructor <org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider.<init>(long, long)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcNumericBetweenParametersProvider.java:55)
-Constructor <org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider.<init>(long, long, long)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcNumericBetweenParametersProvider.java:68)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcDynamicTableSource$JdbcDataStreamScanProvider.<init>(org.apache.flink.connector.jdbc.source.JdbcSource)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcDynamicTableSource.java:319)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcFilterPushdownPreparedStatementVisitor.<init>(java.util.function.Function)> calls constructor <org.apache.flink.table.expressions.ExpressionDefaultVisitor.<init>()> in (JdbcFilterPushdownPreparedStatementVisitor.java:52)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcRowDataLookupFunction.java:95)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:82)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:83)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:84)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:85)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:86)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcRowDataLookupFunction.java:87)
-Constructor <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.<init>(org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions, int, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, [Ljava.lang.String;, org.apache.flink.table.types.logical.RowType, java.util.List, [Ljava.io.Serializable;)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (JdbcRowDataLookupFunction.java:0)
-Constructor <org.apache.flink.connector.jdbc.table.RowDataResultExtractor.<init>(org.apache.flink.connector.jdbc.converter.JdbcRowConverter)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (RowDataResultExtractor.java:35)
-Constructor <org.apache.flink.connector.jdbc.xa.CheckpointAndXid.<init>(long, javax.transaction.xa.Xid, int, boolean)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (CheckpointAndXid.java:46)
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcXaSink [...]
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcXaSinkFunction.j [...]
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcXaSinkFunction.j [...]
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcXaSinkFunction.j [...]
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcXaSinkFunction.j [...]
-Constructor <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.<init>(org.apache.flink.connector.jdbc.internal.JdbcOutputFormat, org.apache.flink.connector.jdbc.xa.XaFacade, org.apache.flink.connector.jdbc.xa.XidGenerator, org.apache.flink.connector.jdbc.xa.XaSinkStateHandler, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.xa.XaGroupOps)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcXaSinkFunction.j [...]
-Constructor <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.<init>(java.util.function.Supplier, java.lang.Integer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (XaFacadeImpl.java:90)
-Constructor <org.apache.flink.connector.jdbc.xa.XidImpl.<init>(int, [B, [B)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (XidImpl.java:46)
-Constructor <org.apache.flink.connector.jdbc.xa.XidImpl.<init>(int, [B, [B)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (XidImpl.java:47)
-Field <org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.fieldTypes> has type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (AbstractJdbcRowConverter.java:0)
-Field <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.fieldDataTypes> has type <[Lorg.apache.flink.table.types.DataType;> in (JdbcOutputFormatBuilder.java:0)
-Interface <org.apache.flink.connector.jdbc.JdbcStatementBuilder> extends interface <org.apache.flink.util.function.BiConsumerWithException> in (JdbcStatementBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.JdbcConnectionOptions$JdbcConnectionOptionsBuilder.withProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcConnectionOptions.java:116)
-Method <org.apache.flink.connector.jdbc.JdbcConnectionOptions$JdbcConnectionOptionsBuilder.withProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcConnectionOptions.java:117)
-Method <org.apache.flink.connector.jdbc.JdbcInputFormat$JdbcInputFormatBuilder.setFetchSize(int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcInputFormat.java:406)
-Method <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.databaseExists(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AbstractJdbcCatalog.java:221)
-Method <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.getDatabase(java.lang.String)> calls constructor <org.apache.flink.table.catalog.CatalogDatabaseImpl.<init>(java.util.Map, java.lang.String)> in (AbstractJdbcCatalog.java:252)
-Method <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.getDatabase(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (AbstractJdbcCatalog.java:248)
-Method <org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog.getPrimaryKey(java.sql.DatabaseMetaData, java.lang.String, java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (AbstractJdbcCatalog.java:194)
-Method <org.apache.flink.connector.jdbc.catalog.JdbcCatalog.getInternal()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcCatalog.java:0)
-Method <org.apache.flink.connector.jdbc.catalog.JdbcCatalogUtils.validateJdbcUrl(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (JdbcCatalogUtils.java:44)
-Method <org.apache.flink.connector.jdbc.converter.AbstractPostgresCompatibleRowConverter.createPostgresArrayConverter(org.apache.flink.table.types.logical.ArrayType)> calls method <org.apache.flink.table.types.logical.utils.LogicalTypeUtils.toInternalConversionClass(org.apache.flink.table.types.logical.LogicalType)> in (AbstractPostgresCompatibleRowConverter.java:75)
-Method <org.apache.flink.connector.jdbc.databases.cratedb.catalog.CrateDBTablePath.fromFlinkTableName(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (CrateDBTablePath.java:41)
-Method <org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog.listTables(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (MySqlCatalog.java:114)
-Method <org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresCatalog.listTables(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (PostgresCatalog.java:184)
-Method <org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresTablePath.fromFlinkTableName(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (PostgresTablePath.java:56)
-Method <org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider.loadDriver(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (SimpleJdbcConnectionProvider.java:86)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.PoolingXaConnectionProvider.endAndPrepare(javax.transaction.xa.Xid)> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (PoolingXaConnectionProvider.java:109)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.PoolingXaConnectionProvider.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (PoolingXaConnectionProvider.java:79)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.PoolingXaConnectionProvider.peekPooled()> calls method <org.apache.flink.util.ExceptionUtils.rethrow(java.lang.Throwable)> in (PoolingXaConnectionProvider.java:202)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.PoolingXaConnectionProvider.start(javax.transaction.xa.Xid)> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (PoolingXaConnectionProvider.java:91)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.execute(org.apache.flink.connector.jdbc.datasource.connections.xa.XaCommand)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (SimpleXaConnectionProvider.java:261)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.from(javax.sql.XADataSource)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (SimpleXaConnectionProvider.java:0)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.getConnection()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (SimpleXaConnectionProvider.java:128)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (SimpleXaConnectionProvider.java:101)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (SimpleXaConnectionProvider.java:87)
-Method <org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider.recover()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (SimpleXaConnectionProvider.java:246)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.checkState()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaTransaction.java:121)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.checkState()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaTransaction.java:122)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.createTx(long)> calls method <org.apache.flink.util.ExceptionUtils.rethrowIOException(java.lang.Throwable)> in (XaTransaction.java:135)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.createTx(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaTransaction.java:130)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.open(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.util.ExceptionUtils.rethrowIOException(java.lang.Throwable)> in (XaTransaction.java:86)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction.prepareTx()> calls method <org.apache.flink.util.ExceptionUtils.rethrowIOException(java.lang.Throwable)> in (XaTransaction.java:151)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.deserialize([B)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (TransactionId.java:96)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.deserialize([B)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (TransactionId.java:101)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.deserialize([B)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (TransactionId.java:98)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.deserialize([B)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (TransactionId.java:99)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.deserialize([B)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readLong()> in (TransactionId.java:100)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.fromXid(int, [B, [B, boolean)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (TransactionId.java:83)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.fromXid(int, [B, [B, boolean)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (TransactionId.java:87)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.fromXid(int, [B, [B, boolean)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (TransactionId.java:85)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.fromXid(int, [B, [B, boolean)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (TransactionId.java:88)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.fromXid(int, [B, [B, boolean)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readLong()> in (TransactionId.java:89)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getBranchQualifier()> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (TransactionId.java:172)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getBranchQualifier()> calls method <org.apache.flink.core.memory.DataOutputSerializer.getSharedBuffer()> in (TransactionId.java:175)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getBranchQualifier()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (TransactionId.java:173)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getBranchQualifier()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeLong(long)> in (TransactionId.java:174)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getBranchQualifier()> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (TransactionId.java:169)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getGlobalTransactionId()> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (TransactionId.java:158)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getGlobalTransactionId()> calls method <org.apache.flink.core.memory.DataOutputSerializer.getSharedBuffer()> in (TransactionId.java:161)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getGlobalTransactionId()> calls method <org.apache.flink.core.memory.DataOutputSerializer.write([B, int, int)> in (TransactionId.java:159)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.getGlobalTransactionId()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (TransactionId.java:160)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.readJobId(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.read([B)> in (TransactionId.java:110)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.readJobId(org.apache.flink.core.memory.DataInputDeserializer)> has parameter of type <org.apache.flink.core.memory.DataInputDeserializer> in (TransactionId.java:0)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (TransactionId.java:183)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.getSharedBuffer()> in (TransactionId.java:189)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.write([B, int, int)> in (TransactionId.java:184)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (TransactionId.java:185)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (TransactionId.java:186)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (TransactionId.java:188)
-Method <org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId.serialize()> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeLong(long)> in (TransactionId.java:187)
-Method <org.apache.flink.connector.jdbc.dialect.AbstractDialect$Range.of(int, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AbstractDialect.java:256)
-Method <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.getConnection()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcOutputFormat.java:0)
-Method <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.open(org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer)> calls constructor <org.apache.flink.util.concurrent.ExecutorThreadFactory.<init>(java.lang.String)> in (JdbcOutputFormat.java:103)
-Method <org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.open(org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcOutputFormat.java:94)
-Method <org.apache.flink.connector.jdbc.internal.RowJdbcOutputFormat$Builder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (RowJdbcOutputFormat.java:132)
-Method <org.apache.flink.connector.jdbc.internal.TableJdbcUpsertOutputFormat.createUpsertRowExecutor(org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (TableJdbcUpsertOutputFormat.java:152)
-Method <org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions$Builder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (InternalJdbcConnectionOptions.java:202)
-Method <org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions$Builder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (InternalJdbcConnectionOptions.java:203)
-Method <org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions$Builder.setProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (InternalJdbcConnectionOptions.java:195)
-Method <org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions$Builder.setProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (InternalJdbcConnectionOptions.java:196)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.build(org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:111)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.build(org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:112)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.build(org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:113)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.build(org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:114)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.build(org.apache.flink.connector.base.DeliveryGuarantee, org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:115)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildAtLeastOnce(org.apache.flink.connector.jdbc.JdbcConnectionOptions)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:67)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildAtLeastOnce(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:73)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildAtLeastOnce(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:78)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildExactlyOnce(org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.xa.XaConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:101)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildExactlyOnce(org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.connector.jdbc.datasource.connections.xa.XaConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:102)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildExactlyOnce(org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.util.function.SerializableSupplier)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:85)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.buildExactlyOnce(org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions, org.apache.flink.util.function.SerializableSupplier)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:86)
-Method <org.apache.flink.connector.jdbc.sink.JdbcSinkBuilder.withExecutionOptions(org.apache.flink.connector.jdbc.JdbcExecutionOptions)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSinkBuilder.java:51)
-Method <org.apache.flink.connector.jdbc.sink.committer.JdbcCommitableSerializer.deserialize(int, [B)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (JdbcCommitableSerializer.java:34)
-Method <org.apache.flink.connector.jdbc.sink.committer.JdbcCommitableSerializer.serialize(org.apache.flink.connector.jdbc.sink.committer.JdbcCommitable)> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (JdbcCommitableSerializer.java:27)
-Method <org.apache.flink.connector.jdbc.sink.committer.JdbcCommitableSerializer.serialize(org.apache.flink.connector.jdbc.sink.committer.JdbcCommitable)> calls method <org.apache.flink.core.memory.DataOutputSerializer.getSharedBuffer()> in (JdbcCommitableSerializer.java:29)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriter.snapshotState(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcWriter.java:192)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserialize(int, [B)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (JdbcWriterStateSerializer.java:48)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.read([B, int, int)> in (JdbcWriterStateSerializer.java:62)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.read([B, int, int)> in (JdbcWriterStateSerializer.java:70)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readByte()> in (JdbcWriterStateSerializer.java:60)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readByte()> in (JdbcWriterStateSerializer.java:68)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (JdbcWriterStateSerializer.java:57)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (JdbcWriterStateSerializer.java:65)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.deserializeV2(org.apache.flink.core.memory.DataInputDeserializer)> has parameter of type <org.apache.flink.core.memory.DataInputDeserializer> in (JdbcWriterStateSerializer.java:0)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (JdbcWriterStateSerializer.java:29)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.getSharedBuffer()> in (JdbcWriterStateSerializer.java:42)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.write([B, int, int)> in (JdbcWriterStateSerializer.java:34)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.write([B, int, int)> in (JdbcWriterStateSerializer.java:40)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeByte(int)> in (JdbcWriterStateSerializer.java:33)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeByte(int)> in (JdbcWriterStateSerializer.java:39)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (JdbcWriterStateSerializer.java:30)
-Method <org.apache.flink.connector.jdbc.sink.writer.JdbcWriterStateSerializer.serialize(org.apache.flink.connector.jdbc.sink.writer.JdbcWriterState)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (JdbcWriterStateSerializer.java:36)
-Method <org.apache.flink.connector.jdbc.source.JdbcSource.equals(java.lang.Object)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSource.java:0)
-Method <org.apache.flink.connector.jdbc.source.JdbcSource.getConfiguration()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSource.java:0)
-Method <org.apache.flink.connector.jdbc.source.JdbcSource.getResultExtractor()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSource.java:0)
-Method <org.apache.flink.connector.jdbc.source.JdbcSource.getSqlSplitEnumeratorProvider()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSource.java:0)
-Method <org.apache.flink.connector.jdbc.source.JdbcSource.getTypeInformation()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSource.java:0)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:250)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:251)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcSourceBuilder.java:248)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setConnectionProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:187)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setConnectionProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:188)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setConnectionProvider(org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceBuilder.java:234)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setDBUrl(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcSourceBuilder.java:164)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setDriverName(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcSourceBuilder.java:156)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setJdbcParameterValuesProvider(org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceBuilder.java:195)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setResultExtractor(org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:140)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setResultSetFetchSize(int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcSourceBuilder.java:224)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setSplitReaderFetchBatchSize(int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcSourceBuilder.java:200)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setSql(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcSourceBuilder.java:131)
-Method <org.apache.flink.connector.jdbc.source.JdbcSourceBuilder.setTypeInformation(org.apache.flink.api.common.typeinfo.TypeInformation)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcSourceBuilder.java:173)
-Method <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumStateSerializer.deserialize(int, [B)> calls method <org.apache.flink.util.InstantiationUtil.deserializeObject([B, java.lang.ClassLoader)> in (JdbcSourceEnumStateSerializer.java:110)
-Method <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumStateSerializer.serialize(org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumeratorState)> calls method <org.apache.flink.util.InstantiationUtil.serializeObject(java.lang.Object)> in (JdbcSourceEnumStateSerializer.java:70)
-Method <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumStateSerializer.serialize(org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumeratorState)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcSourceEnumStateSerializer.java:58)
-Method <org.apache.flink.connector.jdbc.source.enumerator.JdbcSourceEnumStateSerializer.serializeJdbcSourceSplits(java.io.DataOutputStream, java.util.List)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (JdbcSourceEnumStateSerializer.java:86)
-Method <org.apache.flink.connector.jdbc.source.enumerator.SqlTemplateSplitEnumerator$TemplateSqlSplitEnumeratorProvider.setSqlTemplate(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (SqlTemplateSplitEnumerator.java:92)
-Method <org.apache.flink.connector.jdbc.source.enumerator.SqlTemplateSplitEnumerator.getParameterValuesProvider()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (SqlTemplateSplitEnumerator.java:0)
-Method <org.apache.flink.connector.jdbc.source.enumerator.SqlTemplateSplitEnumerator.getSqlTemplate()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (SqlTemplateSplitEnumerator.java:0)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.fetch()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcSourceSplitReader.java:131)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.finishSplit()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcSourceSplitReader.java:157)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.getConnection()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSourceSplitReader.java:0)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.getResultSet()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSourceSplitReader.java:0)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.getSplits()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSourceSplitReader.java:0)
-Method <org.apache.flink.connector.jdbc.source.reader.JdbcSourceSplitReader.getStatement()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcSourceSplitReader.java:0)
-Method <org.apache.flink.connector.jdbc.source.reader.extractor.RowResultExtractor.extract(java.sql.ResultSet)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (RowResultExtractor.java:33)
-Method <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer.deserializeJdbcSourceSplit(java.io.DataInputStream)> calls method <org.apache.flink.util.InstantiationUtil.deserializeObject([B, java.lang.ClassLoader)> in (JdbcSourceSplitSerializer.java:100)
-Method <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer.deserializeJdbcSourceSplit(java.io.DataInputStream)> calls method <org.apache.flink.util.InstantiationUtil.deserializeObject([B, java.lang.ClassLoader)> in (JdbcSourceSplitSerializer.java:109)
-Method <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer.serialize(org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcSourceSplitSerializer.java:46)
-Method <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer.serializeJdbcSourceSplit(java.io.DataOutputStream, org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit)> calls method <org.apache.flink.util.InstantiationUtil.serializeObject(java.lang.Object)> in (JdbcSourceSplitSerializer.java:80)
-Method <org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer.serializeJdbcSourceSplit(java.io.DataOutputStream, org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit)> calls method <org.apache.flink.util.InstantiationUtil.serializeObject(java.lang.Object)> in (JdbcSourceSplitSerializer.java:87)
-Method <org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider.getParameterValues()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcNumericBetweenParametersProvider.java:100)
-Method <org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider.ofBatchNum(int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcNumericBetweenParametersProvider.java:87)
-Method <org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider.ofBatchSize(long)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcNumericBetweenParametersProvider.java:75)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.parseNamedStatement(java.lang.String, java.util.Map)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (FieldNamedPreparedStatementImpl.java:247)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.prepareStatement(java.sql.Connection, java.lang.String, [Ljava.lang.String;, java.lang.String, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (FieldNamedPreparedStatementImpl.java:202)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.prepareStatement(java.sql.Connection, java.lang.String, [Ljava.lang.String;, java.lang.String, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (FieldNamedPreparedStatementImpl.java:212)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.prepareStatement(java.sql.Connection, java.lang.String, [Ljava.lang.String;, java.lang.String, int)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (FieldNamedPreparedStatementImpl.java:187)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.prepareStatement(java.sql.Connection, java.lang.String, [Ljava.lang.String;, java.lang.String, int)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (FieldNamedPreparedStatementImpl.java:188)
-Method <org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.prepareStatement(java.sql.Connection, java.lang.String, [Ljava.lang.String;, java.lang.String, int)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (FieldNamedPreparedStatementImpl.java:189)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.checkAllOrNone(org.apache.flink.configuration.ReadableConfig, [Lorg.apache.flink.configuration.ConfigOption;)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcDynamicTableFactory.java:351)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.checkAllOrNone(org.apache.flink.configuration.ReadableConfig, [Lorg.apache.flink.configuration.ConfigOption;)> has parameter of type <[Lorg.apache.flink.configuration.ConfigOption;> in (JdbcDynamicTableFactory.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.createDynamicTableSource(org.apache.flink.table.factories.DynamicTableFactory$Context)> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.MAX_RETRIES> in (JdbcDynamicTableFactory.java:127)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.getLookupCache(org.apache.flink.configuration.ReadableConfig)> calls method <org.apache.flink.table.connector.source.lookup.LookupOptions$LookupCacheType.equals(java.lang.Object)> in (JdbcDynamicTableFactory.java:216)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.getLookupCache(org.apache.flink.configuration.ReadableConfig)> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions$LookupCacheType.PARTIAL> in (JdbcDynamicTableFactory.java:215)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.getLookupCache(org.apache.flink.configuration.ReadableConfig)> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.CACHE_TYPE> in (JdbcDynamicTableFactory.java:214)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.CACHE_TYPE> in (JdbcDynamicTableFactory.java:257)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.MAX_RETRIES> in (JdbcDynamicTableFactory.java:262)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.PARTIAL_CACHE_CACHE_MISSING_KEY> in (JdbcDynamicTableFactory.java:261)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_ACCESS> in (JdbcDynamicTableFactory.java:258)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_WRITE> in (JdbcDynamicTableFactory.java:259)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.optionalOptions()> gets field <org.apache.flink.table.connector.source.lookup.LookupOptions.PARTIAL_CACHE_MAX_ROWS> in (JdbcDynamicTableFactory.java:260)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableSink.validatePrimaryKey(org.apache.flink.table.connector.ChangelogMode)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcDynamicTableSink.java:69)
-Method <org.apache.flink.connector.jdbc.table.JdbcDynamicTableSource.getLookupRuntimeProvider(org.apache.flink.table.connector.source.LookupTableSource$LookupContext)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (JdbcDynamicTableSource.java:112)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcOutputFormatBuilder.java:81)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcOutputFormatBuilder.java:82)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (JdbcOutputFormatBuilder.java:83)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createBufferReduceExecutor(org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions, [Lorg.apache.flink.table.types.logical.LogicalType;)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (JdbcOutputFormatBuilder.java:116)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createBufferReduceExecutor(org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions, [Lorg.apache.flink.table.types.logical.LogicalType;)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createDeleteExecutor(org.apache.flink.connector.jdbc.dialect.JdbcDialect, java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createInsertOrUpdateExecutor(org.apache.flink.connector.jdbc.dialect.JdbcDialect, java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;, [I, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createRowKeyExtractor([Lorg.apache.flink.table.types.logical.LogicalType;, [I)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createSimpleBufferedExecutor(org.apache.flink.connector.jdbc.dialect.JdbcDialect, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;, java.lang.String)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createSimpleRowExecutor(org.apache.flink.connector.jdbc.dialect.JdbcDialect, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;, java.lang.String)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.createUpsertRowExecutor(org.apache.flink.connector.jdbc.dialect.JdbcDialect, java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;, [I, [Ljava.lang.String;, [Lorg.apache.flink.table.types.logical.LogicalType;)> has parameter of type <[Lorg.apache.flink.table.types.logical.LogicalType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.getPrimaryKey(org.apache.flink.table.data.RowData, [Lorg.apache.flink.table.data.RowData$FieldGetter;)> has parameter of type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcOutputFormatBuilder.setFieldDataTypes([Lorg.apache.flink.table.types.DataType;)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (JdbcOutputFormatBuilder.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat$Builder.setFetchSize(int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.String, [Ljava.lang.Object;)> in (JdbcRowDataInputFormat.java:356)
-Method <org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.createInputSplits(int)> has return type <[Lorg.apache.flink.core.io.InputSplit;> in (JdbcRowDataInputFormat.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.getInputSplitAssigner([Lorg.apache.flink.core.io.InputSplit;)> calls constructor <org.apache.flink.api.common.io.DefaultInputSplitAssigner.<init>([Lorg.apache.flink.core.io.InputSplit;)> in (JdbcRowDataInputFormat.java:287)
-Method <org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.getInputSplitAssigner([Lorg.apache.flink.core.io.InputSplit;)> has parameter of type <[Lorg.apache.flink.core.io.InputSplit;> in (JdbcRowDataInputFormat.java:0)
-Method <org.apache.flink.connector.jdbc.table.JdbcRowDataLookupFunction.getDbConnection()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (JdbcRowDataLookupFunction.java:0)
-Method <org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties(org.apache.flink.configuration.ReadableConfig)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (JdbcUtils.java:49)
-Method <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.beginTx(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcXaSinkFunction.java:312)
-Method <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.invoke(java.lang.Object, org.apache.flink.streaming.api.functions.sink.SinkFunction$Context)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcXaSinkFunction.java:263)
-Method <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.prepareCurrentTx(long)> calls method <org.apache.flink.util.ExceptionUtils.rethrowIOException(java.lang.Throwable)> in (JdbcXaSinkFunction.java:305)
-Method <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.prepareCurrentTx(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcXaSinkFunction.java:290)
-Method <org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction.prepareCurrentTx(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (JdbcXaSinkFunction.java:291)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.execute(org.apache.flink.connector.jdbc.xa.XaFacadeImpl$Command)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaFacadeImpl.java:270)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.fromXaDataSource(javax.sql.XADataSource)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (XaFacadeImpl.java:0)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.getConnection()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (XaFacadeImpl.java:137)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (XaFacadeImpl.java:110)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaFacadeImpl.java:96)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadeImpl.recover()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (XaFacadeImpl.java:250)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadePoolingImpl.endAndPrepare(javax.transaction.xa.Xid)> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (XaFacadePoolingImpl.java:95)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadePoolingImpl.open()> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (XaFacadePoolingImpl.java:65)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadePoolingImpl.peekPooled()> calls method <org.apache.flink.util.ExceptionUtils.rethrow(java.lang.Throwable)> in (XaFacadePoolingImpl.java:187)
-Method <org.apache.flink.connector.jdbc.xa.XaFacadePoolingImpl.start(javax.transaction.xa.Xid)> calls method <org.apache.flink.util.Preconditions.checkState(boolean)> in (XaFacadePoolingImpl.java:77)
\ No newline at end of file
diff --git a/flink-connector-jdbc-core/archunit-violations/stored.rules b/flink-connector-jdbc-core/archunit-violations/stored.rules
index 1b7a0e39..0f9bf9b4 100644
--- a/flink-connector-jdbc-core/archunit-violations/stored.rules
+++ b/flink-connector-jdbc-core/archunit-violations/stored.rules
@@ -1,11 +1,10 @@
#
-#Sat May 18 04:39:49 CEST 2024
-Classes\ in\ API\ packages\ should\ have\ at\ least\ one\ API\ visibility\ annotation.=0c7b60d6-a12c-432d-8a72-747cabaeb7ed
-Connector\ production\ code\ must\ depend\ only\ on\ public\ API\ when\ outside\ of\ connector\ packages=2047daea-3713-4298-bdb3-e269f539363f
+#Tue May 28 16:21:17 CEST 2024
+Return\ and\ argument\ types\ of\ methods\ annotated\ with\ @Public\ must\ be\ annotated\ with\ @Public.=596bd7eb-1ed5-4963-a56a-941766d61233
Connector\ production\ code\ must\ not\ depend\ on\ non-public\ API\ outside\ of\ connector\ packages=f242ee43-a122-4d2f-a291-6439cdb9029f
ITCASE\ tests\ should\ use\ a\ MiniCluster\ resource\ or\ extension=6b9ab1b0-c14d-4667-bab5-407b81fba98b
-Options\ for\ connectors\ and\ formats\ should\ reside\ in\ a\ consistent\ package\ and\ be\ public\ API.=b10c194a-0517-4744-830f-81abfb71ab18
Production\ code\ must\ not\ call\ methods\ annotated\ with\ @VisibleForTesting=4bca2274-65a9-4a61-81ef-767d58233ea0
-Return\ and\ argument\ types\ of\ methods\ annotated\ with\ @Public\ must\ be\ annotated\ with\ @Public.=596bd7eb-1ed5-4963-a56a-941766d61233
-Return\ and\ argument\ types\ of\ methods\ annotated\ with\ @PublicEvolving\ must\ be\ annotated\ with\ @Public(Evolving).=d45c3af5-52c6-45fd-9926-75e75e77473a
+Options\ for\ connectors\ and\ formats\ should\ reside\ in\ a\ consistent\ package\ and\ be\ public\ API.=b10c194a-0517-4744-830f-81abfb71ab18
Tests\ inheriting\ from\ AbstractTestBase\ should\ have\ name\ ending\ with\ ITCase=0f94f11b-9bc8-445a-8be8-e7dfd8a384b0
+Return\ and\ argument\ types\ of\ methods\ annotated\ with\ @PublicEvolving\ must\ be\ annotated\ with\ @Public(Evolving).=d45c3af5-52c6-45fd-9926-75e75e77473a
+Classes\ in\ API\ packages\ should\ have\ at\ least\ one\ API\ visibility\ annotation.=0c7b60d6-a12c-432d-8a72-747cabaeb7ed
diff --git a/flink-connector-jdbc-core/pom.xml b/flink-connector-jdbc-core/pom.xml
index 990582d8..c7e767df 100644
--- a/flink-connector-jdbc-core/pom.xml
+++ b/flink-connector-jdbc-core/pom.xml
@@ -35,9 +35,7 @@ under the License.
<packaging>jar</packaging>
<properties>
- <scala.binary.version>2.12</scala.binary.version>
- <scala-library.version>2.12.7</scala-library.version>
- <assertj.version>3.23.1</assertj.version>
+
<postgres.version>42.7.3</postgres.version>
<oracle.version>21.8.0.0</oracle.version>
<trino.version>418</trino.version>
@@ -284,17 +282,6 @@ under the License.
</dependencies>
- <dependencyManagement>
- <dependencies>
- <!-- For dependency convergence -->
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>${scala-library.version}</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
<build>
<plugins>
<plugin>
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactory.java
index 448ecaa5..250c6d65 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactory.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.core.table;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
import org.apache.flink.util.StringUtils;
@@ -63,11 +64,26 @@ public interface JdbcFactory {
"Not supported option 'compatible-mode' with value: " + compatibleMode);
}
- // JdbcCatalog createCatalog(
- // ClassLoader classLoader,
- // String catalogName,
- // String defaultDatabase,
- // String username,
- // String pwd,
- // String baseUrl);
+ JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl);
+
+ default JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl,
+ String compatibleMode) {
+ if (StringUtils.isNullOrWhitespaceOnly(compatibleMode)) {
+ return createCatalog(classLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
+ throw new UnsupportedOperationException(
+ "Not supported option 'compatible-mode' with value: " + compatibleMode);
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactoryLoader.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactoryLoader.java
index 3549b76d..a9db62cc 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactoryLoader.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/JdbcFactoryLoader.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.core.table;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
import org.slf4j.Logger;
@@ -57,37 +58,34 @@ public final class JdbcFactoryLoader {
return load(url, classLoader).createDialect(compatibleMode);
}
- // public static JdbcCatalog loadCatalog(
- // ClassLoader classLoader,
- // String catalogName,
- // String defaultDatabase,
- // String username,
- // String pwd,
- // String baseUrl) {
- // return load(baseUrl, classLoader)
- // .createCatalog(
- // classLoader, catalogName, defaultDatabase, username, pwd, baseUrl,
- // null);
- // }
- //
- // public static JdbcCatalog loadCatalog(
- // ClassLoader classLoader,
- // String catalogName,
- // String defaultDatabase,
- // String username,
- // String pwd,
- // String baseUrl,
- // String compatibleMode) {
- // return load(baseUrl, classLoader)
- // .createCatalog(
- // classLoader,
- // catalogName,
- // defaultDatabase,
- // username,
- // pwd,
- // baseUrl,
- // compatibleMode);
- // }
+ public static JdbcCatalog loadCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ return loadCatalog(classLoader, catalogName, defaultDatabase, username, pwd, baseUrl, null);
+ }
+
+ public static JdbcCatalog loadCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl,
+ String compatibleMode) {
+ return load(baseUrl, classLoader)
+ .createCatalog(
+ classLoader,
+ catalogName,
+ defaultDatabase,
+ username,
+ pwd,
+ baseUrl,
+ compatibleMode);
+ }
private static JdbcFactory load(String url, ClassLoader classLoader) {
List<JdbcFactory> foundFactories = discoverFactories(classLoader);
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalog.java
similarity index 95%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
rename to flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalog.java
index a0521484..a3c27c47 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalog.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.catalog;
+package org.apache.flink.connector.jdbc.core.table.catalog;
import org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory;
import org.apache.flink.table.api.Schema;
@@ -87,7 +87,7 @@ import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
/** Abstract catalog for any JDBC catalogs. */
-public abstract class AbstractJdbcCatalog extends AbstractCatalog {
+public abstract class AbstractJdbcCatalog extends AbstractCatalog implements JdbcCatalog {
private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcCatalog.class);
@@ -123,7 +123,7 @@ public abstract class AbstractJdbcCatalog extends AbstractCatalog {
checkNotNull(userClassLoader);
checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
- JdbcCatalogUtils.validateJdbcUrl(baseUrl);
+ validateJdbcUrl(baseUrl);
this.userClassLoader = userClassLoader;
this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
@@ -565,4 +565,35 @@ public abstract class AbstractJdbcCatalog extends AbstractCatalog {
protected String getSchemaTableName(ObjectPath tablePath) {
throw new UnsupportedOperationException();
}
+
+ /**
+ * URL has to be without database, like "jdbc:dialect://localhost:1234/" or
+ * "jdbc:dialect://localhost:1234" rather than "jdbc:dialect://localhost:1234/db".
+ */
+ protected static void validateJdbcUrl(String url) {
+ String[] parts = url.trim().split("\\/+");
+
+ checkArgument(parts.length == 2);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof AbstractJdbcCatalog)) {
+ return false;
+ }
+ AbstractJdbcCatalog that = (AbstractJdbcCatalog) o;
+ return Objects.equals(getName(), that.getName())
+ && Objects.equals(getDefaultDatabase(), that.getDefaultDatabase())
+ && Objects.equals(connectionProperties, that.connectionProperties)
+ && Objects.equals(baseUrl, that.baseUrl);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ getName(), getDefaultDatabase(), userClassLoader, connectionProperties, baseUrl);
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalog.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalog.java
new file mode 100644
index 00000000..5c70deee
--- /dev/null
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalog.java
@@ -0,0 +1,10 @@
+package org.apache.flink.connector.jdbc.core.table.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.Catalog;
+
+import java.io.Serializable;
+
+/** Catalogs for relational databases via JDBC. */
+@PublicEvolving
+public interface JdbcCatalog extends Catalog, Serializable {}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalogTypeMapper.java
similarity index 87%
copy from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
copy to flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalogTypeMapper.java
index 11fc46ef..fa3e4cf4 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/JdbcCatalogTypeMapper.java
@@ -16,8 +16,9 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.dialect;
+package org.apache.flink.connector.jdbc.core.table.catalog;
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.types.DataType;
@@ -25,7 +26,8 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
/** Separate the jdbc meta-information type to flink table type into the interface. */
-public interface JdbcDialectTypeMapper {
+@PublicEvolving
+public interface JdbcCatalogTypeMapper {
DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)
throws SQLException;
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactory.java
similarity index 65%
copy from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
copy to flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactory.java
index a19e0bae..b1fd7ca5 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactory.java
@@ -16,10 +16,11 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.catalog.factory;
+package org.apache.flink.connector.jdbc.core.table.catalog.factory;
import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.JdbcFactoryLoader;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.FactoryUtil;
@@ -30,12 +31,11 @@ import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Set;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.BASE_URL;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.COMPATIBLE_MODE;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.DEFAULT_DATABASE;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
-import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.BASE_URL;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.COMPATIBLE_MODE;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.DEFAULT_DATABASE;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
/** Factory for {@link JdbcCatalog}. */
@@ -72,12 +72,13 @@ public class JdbcCatalogFactory implements CatalogFactory {
FactoryUtil.createCatalogFactoryHelper(this, context);
helper.validate();
- return new JdbcCatalog(
+ return JdbcFactoryLoader.loadCatalog(
context.getClassLoader(),
context.getName(),
- helper.getOptions().get(DEFAULT_DATABASE),
- helper.getOptions().get(BASE_URL),
- helper.getOptions().get(COMPATIBLE_MODE),
- getConnectionProperties(helper.getOptions()));
+ helper.getOptions().get(JdbcCatalogFactoryOptions.DEFAULT_DATABASE),
+ helper.getOptions().get(JdbcCatalogFactoryOptions.USERNAME),
+ helper.getOptions().get(JdbcCatalogFactoryOptions.PASSWORD),
+ helper.getOptions().get(JdbcCatalogFactoryOptions.BASE_URL),
+ helper.getOptions().get(JdbcCatalogFactoryOptions.COMPATIBLE_MODE));
}
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactoryOptions.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactoryOptions.java
similarity index 93%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactoryOptions.java
rename to flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactoryOptions.java
index 197f6b9f..b13f5c68 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactoryOptions.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/catalog/factory/JdbcCatalogFactoryOptions.java
@@ -16,12 +16,12 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.catalog.factory;
+package org.apache.flink.connector.jdbc.core.table.catalog.factory;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.table.JdbcConnectorOptions;
import org.apache.flink.table.catalog.CommonCatalogOptions;
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBFactory.java
index fcf31b24..94179408 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBFactory.java
@@ -20,7 +20,9 @@ package org.apache.flink.connector.jdbc.databases.cratedb.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
+import org.apache.flink.connector.jdbc.databases.cratedb.catalog.CrateDBCatalog;
/** Factory for {@link CrateDBDialect}. */
@Internal
@@ -34,4 +36,16 @@ public class CrateDBFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new CrateDBDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ return new CrateDBCatalog(
+ classLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2Factory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2Factory.java
index 268a20c2..4b0165cc 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2Factory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2Factory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.db2.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
/** Factory for {@link Db2Dialect}. */
@@ -34,4 +35,15 @@ public class Db2Factory implements JdbcFactory {
public JdbcDialect createDialect() {
return new Db2Dialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for DB2 is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/derby/dialect/DerbyFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/derby/dialect/DerbyFactory.java
index 639d8695..7df224fc 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/derby/dialect/DerbyFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/derby/dialect/DerbyFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.derby.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
/** Factory for {@link DerbyDialect}. */
@@ -34,4 +35,15 @@ public class DerbyFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new DerbyDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for Derby is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlCatalog.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlCatalog.java
index a511e77c..982b92e9 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlCatalog.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlCatalog.java
@@ -20,8 +20,8 @@ package org.apache.flink.connector.jdbc.databases.mysql.catalog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper;
+import org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalogTypeMapper;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
@@ -52,7 +52,7 @@ public class MySqlCatalog extends AbstractJdbcCatalog {
private static final Logger LOG = LoggerFactory.getLogger(MySqlCatalog.class);
- private final JdbcDialectTypeMapper dialectTypeMapper;
+ private final JdbcCatalogTypeMapper dialectTypeMapper;
private static final Set<String> builtinDatabases =
new HashSet<String>() {
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlTypeMapper.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlTypeMapper.java
index 235136dd..f3ca932d 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlTypeMapper.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/catalog/MySqlTypeMapper.java
@@ -19,7 +19,7 @@
package org.apache.flink.connector.jdbc.databases.mysql.catalog;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalogTypeMapper;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
@@ -34,7 +34,7 @@ import java.sql.SQLException;
/** MySQLTypeMapper util class. */
@Internal
-public class MySqlTypeMapper implements JdbcDialectTypeMapper {
+public class MySqlTypeMapper implements JdbcCatalogTypeMapper {
private static final Logger LOG = LoggerFactory.getLogger(MySqlTypeMapper.class);
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlFactory.java
index a3544a01..87051f23 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlFactory.java
@@ -20,7 +20,9 @@ package org.apache.flink.connector.jdbc.databases.mysql.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
+import org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog;
/** Factory for {@link MySqlDialect}. */
@Internal
@@ -34,4 +36,15 @@ public class MySqlFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new MySqlDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ return new MySqlCatalog(classLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseFactory.java
index 1455e310..2f30aa3c 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.oceanbase.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
import javax.annotation.Nonnull;
@@ -43,4 +44,15 @@ public class OceanBaseFactory implements JdbcFactory {
public JdbcDialect createDialect(@Nonnull String compatibleMode) {
return new OceanBaseDialect(compatibleMode);
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for OceanBase is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleFactory.java
index 78f124b1..fa63b9dc 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.oracle.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
/** Factory for {@link OracleDialect}. */
@@ -34,4 +35,15 @@ public class OracleFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new OracleDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for Oracle is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresCatalog.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresCatalog.java
index 27aa1f4d..301691b7 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresCatalog.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresCatalog.java
@@ -20,8 +20,8 @@ package org.apache.flink.connector.jdbc.databases.postgres.catalog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper;
+import org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalogTypeMapper;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
@@ -74,7 +74,7 @@ public class PostgresCatalog extends AbstractJdbcCatalog {
}
};
- protected final JdbcDialectTypeMapper dialectTypeMapper;
+ protected final JdbcCatalogTypeMapper dialectTypeMapper;
@VisibleForTesting
public PostgresCatalog(
@@ -115,7 +115,7 @@ public class PostgresCatalog extends AbstractJdbcCatalog {
String username,
String pwd,
String baseUrl,
- JdbcDialectTypeMapper dialectTypeMapper) {
+ JdbcCatalogTypeMapper dialectTypeMapper) {
this(
userClassLoader,
catalogName,
@@ -130,7 +130,7 @@ public class PostgresCatalog extends AbstractJdbcCatalog {
String catalogName,
String defaultDatabase,
String baseUrl,
- JdbcDialectTypeMapper dialectTypeMapper,
+ JdbcCatalogTypeMapper dialectTypeMapper,
Properties connectProperties) {
super(userClassLoader, catalogName, defaultDatabase, baseUrl, connectProperties);
this.dialectTypeMapper = dialectTypeMapper;
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresTypeMapper.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresTypeMapper.java
index f213128a..c6b92605 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresTypeMapper.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/PostgresTypeMapper.java
@@ -19,7 +19,7 @@
package org.apache.flink.connector.jdbc.databases.postgres.catalog;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalogTypeMapper;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.types.DataType;
@@ -33,7 +33,7 @@ import java.sql.SQLException;
/** PostgresTypeMapper util class. */
@Internal
-public class PostgresTypeMapper implements JdbcDialectTypeMapper {
+public class PostgresTypeMapper implements JdbcCatalogTypeMapper {
private static final Logger LOG = LoggerFactory.getLogger(PostgresTypeMapper.class);
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresFactory.java
index 5fe6df8f..5c5310ed 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresFactory.java
@@ -20,7 +20,9 @@ package org.apache.flink.connector.jdbc.databases.postgres.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
+import org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresCatalog;
/** Factory for {@link PostgresDialect}. */
@Internal
@@ -34,4 +36,16 @@ public class PostgresFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new PostgresDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ return new PostgresCatalog(
+ classLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/sqlserver/dialect/SqlServerFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/sqlserver/dialect/SqlServerFactory.java
index 695cca72..43d66827 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/sqlserver/dialect/SqlServerFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/sqlserver/dialect/SqlServerFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.sqlserver.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
/** Factory for {@link SqlServerDialect}. */
@@ -34,4 +35,15 @@ public class SqlServerFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new SqlServerDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for SqlServer is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoFactory.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoFactory.java
index 9283b8b4..d6ef4da8 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoFactory.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.databases.trino.dialect;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.core.table.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialect;
/** Factory for {@link TrinoDialect}. */
@@ -34,4 +35,15 @@ public class TrinoFactory implements JdbcFactory {
public JdbcDialect createDialect() {
return new TrinoDialect();
}
+
+ @Override
+ public JdbcCatalog createCatalog(
+ ClassLoader classLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ throw new UnsupportedOperationException("Catalog for Trino is not supported yet.");
+ }
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/utils/JdbcUtils.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/utils/JdbcUtils.java
index d86c7319..bf405a20 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/utils/JdbcUtils.java
+++ b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/utils/JdbcUtils.java
@@ -33,8 +33,8 @@ import java.util.Objects;
import java.util.Properties;
import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
/** Utils for jdbc connectors. */
public class JdbcUtils {
diff --git a/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index a54d9f57..41de0904 100644
--- a/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -14,4 +14,4 @@
# limitations under the License.
org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory
-org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory
+org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactory
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtilsTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalogTest.java
similarity index 71%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtilsTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalogTest.java
index 23407143..cde094a0 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtilsTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/catalog/AbstractJdbcCatalogTest.java
@@ -16,28 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.catalog;
+package org.apache.flink.connector.jdbc.core.table.catalog;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Test for {@link JdbcCatalogUtils}. */
-class JdbcCatalogUtilsTest {
+/** Test for {@link AbstractJdbcCatalog}. */
+class AbstractJdbcCatalogTest {
@Test
void testJdbcUrl() {
- JdbcCatalogUtils.validateJdbcUrl("jdbc:postgresql://localhost:5432/");
-
- JdbcCatalogUtils.validateJdbcUrl("jdbc:postgresql://localhost:5432");
+ AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234/");
+ AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234");
}
@Test
void testInvalidJdbcUrl() {
assertThatThrownBy(
() ->
- JdbcCatalogUtils.validateJdbcUrl(
- "jdbc:postgresql://localhost:5432/db"))
+ AbstractJdbcCatalog.validateJdbcUrl(
+ "jdbc:dialect://localhost:1234/db"))
.isInstanceOf(IllegalArgumentException.class);
}
}
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/converter/AbstractDialectConverterTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectConverterTest.java
similarity index 89%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/converter/AbstractDialectConverterTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectConverterTest.java
index bfb77392..d39bf355 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/converter/AbstractDialectConverterTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectConverterTest.java
@@ -16,10 +16,8 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.converter;
+package org.apache.flink.connector.jdbc.core.table.dialect;
-import org.apache.flink.connector.jdbc.core.table.dialect.AbstractDialectConverter;
-import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectConverter;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.RowType;
@@ -34,7 +32,7 @@ import java.time.LocalDateTime;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link AbstractDialectConverter}. */
-class AbstractDialectConverterTest {
+class JdbcDialectConverterTest {
@Test
void testExternalLocalDateTimeToTimestamp() throws Exception {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectTest.java
similarity index 98%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectTest.java
index d73c61e4..99ad42ff 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/dialect/JdbcDialectTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.dialect;
+package org.apache.flink.connector.jdbc.core.table.dialect;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
@@ -35,7 +35,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for all DataTypes and Dialects of JDBC connector. */
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class JdbcDialectTypeTest {
+public class JdbcDialectTest {
protected String ddlFormat =
"CREATE TABLE T (f0 %s)"
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTest.java
similarity index 92%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTest.java
index 9e90e336..65922f52 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/cratedb/dialect/CrateDBDialectTest.java
@@ -18,13 +18,13 @@
package org.apache.flink.connector.jdbc.databases.cratedb.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The CrateDB params for {@link JdbcDialectTypeTest}. */
-class CrateDBDialectTypeTest extends JdbcDialectTypeTest {
+/** The CrateDB params for {@link JdbcDialectTest}. */
+class CrateDBDialectTest extends JdbcDialectTest {
@Override
protected String testDialect() {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTest.java
similarity index 91%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTest.java
index 31e364a2..04fbe3c9 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/db2/dialect/Db2DialectTest.java
@@ -18,13 +18,13 @@
package org.apache.flink.connector.jdbc.databases.db2.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The Db2 params for {@link JdbcDialectTypeTest}. */
-class Db2DialectTypeTest extends JdbcDialectTypeTest {
+/** The Db2 params for {@link JdbcDialectTest}. */
+class Db2DialectTest extends JdbcDialectTest {
@Override
protected String testDialect() {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTest.java
index a91d9b41..5491d398 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTest.java
@@ -18,12 +18,54 @@
package org.apache.flink.connector.jdbc.databases.mysql.dialect;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
+
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.List;
+
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for {@link MySqlDialect}. */
-class MySqlDialectTest {
+/** The MySql params for {@link JdbcDialectTest}. */
+class MySqlDialectTest extends JdbcDialectTest {
+
+ @Override
+ protected String testDialect() {
+ return "mysql";
+ }
+
+ @Override
+ protected List<TestItem> testData() {
+ return Arrays.asList(
+ createTestItem("CHAR"),
+ createTestItem("VARCHAR"),
+ createTestItem("BOOLEAN"),
+ createTestItem("TINYINT"),
+ createTestItem("SMALLINT"),
+ createTestItem("INTEGER"),
+ createTestItem("BIGINT"),
+ createTestItem("FLOAT"),
+ createTestItem("DOUBLE"),
+ createTestItem("DECIMAL(10, 4)"),
+ createTestItem("DECIMAL(38, 18)"),
+ createTestItem("DATE"),
+ createTestItem("TIME"),
+ createTestItem("TIMESTAMP(3)"),
+ createTestItem("TIMESTAMP WITHOUT TIME ZONE"),
+ createTestItem("VARBINARY"),
+
+ // Not valid data
+ createTestItem("BINARY", "The MySQL dialect doesn't support type: BINARY(1)."),
+ createTestItem(
+ "VARBINARY(10)", "The MySQL dialect doesn't support type: VARBINARY(10)."),
+ createTestItem(
+ "TIMESTAMP(9) WITHOUT TIME ZONE",
+ "The precision of field 'f0' is out of the TIMESTAMP precision range [0, 6] supported by MySQL dialect."),
+ createTestItem(
+ "TIMESTAMP_LTZ(3)",
+ "The MySQL dialect doesn't support type: TIMESTAMP_LTZ(3)."));
+ }
@Test
void testAppendDefaultUrlProperties() {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTypeTest.java
deleted file mode 100644
index dac699cb..00000000
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/mysql/dialect/MySqlDialectTypeTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.flink.connector.jdbc.databases.mysql.dialect;
-
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
-
-import java.util.Arrays;
-import java.util.List;
-
-/** The MySql params for {@link JdbcDialectTypeTest}. */
-class MySqlDialectTypeTest extends JdbcDialectTypeTest {
-
- @Override
- protected String testDialect() {
- return "mysql";
- }
-
- @Override
- protected List<TestItem> testData() {
- return Arrays.asList(
- createTestItem("CHAR"),
- createTestItem("VARCHAR"),
- createTestItem("BOOLEAN"),
- createTestItem("TINYINT"),
- createTestItem("SMALLINT"),
- createTestItem("INTEGER"),
- createTestItem("BIGINT"),
- createTestItem("FLOAT"),
- createTestItem("DOUBLE"),
- createTestItem("DECIMAL(10, 4)"),
- createTestItem("DECIMAL(38, 18)"),
- createTestItem("DATE"),
- createTestItem("TIME"),
- createTestItem("TIMESTAMP(3)"),
- createTestItem("TIMESTAMP WITHOUT TIME ZONE"),
- createTestItem("VARBINARY"),
-
- // Not valid data
- createTestItem("BINARY", "The MySQL dialect doesn't support type: BINARY(1)."),
- createTestItem(
- "VARBINARY(10)", "The MySQL dialect doesn't support type: VARBINARY(10)."),
- createTestItem(
- "TIMESTAMP(9) WITHOUT TIME ZONE",
- "The precision of field 'f0' is out of the TIMESTAMP precision range [0, 6] supported by MySQL dialect."),
- createTestItem(
- "TIMESTAMP_LTZ(3)",
- "The MySQL dialect doesn't support type: TIMESTAMP_LTZ(3)."));
- }
-}
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTest.java
similarity index 91%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTest.java
index 09daadaf..7b74b3a7 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseMysqlDialectTest.java
@@ -18,15 +18,15 @@
package org.apache.flink.connector.jdbc.databases.oceanbase.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The OceanBase MySql mode params for {@link JdbcDialectTypeTest}. */
-class OceanBaseMysqlDialectTypeTest extends JdbcDialectTypeTest {
+/** The OceanBase MySql mode params for {@link JdbcDialectTest}. */
+class OceanBaseMysqlDialectTest extends JdbcDialectTest {
- public OceanBaseMysqlDialectTypeTest() {
+ public OceanBaseMysqlDialectTest() {
ddlFormat =
"CREATE TABLE T (f0 %s)"
+ " WITH ("
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTest.java
similarity index 90%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTest.java
index 21d46a3b..ebc519f1 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oceanbase/dialect/OceanBaseOracleDialectTest.java
@@ -18,15 +18,15 @@
package org.apache.flink.connector.jdbc.databases.oceanbase.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The OceanBase Oracle mode params for {@link JdbcDialectTypeTest}. */
-class OceanBaseOracleDialectTypeTest extends JdbcDialectTypeTest {
+/** The OceanBase Oracle mode params for {@link JdbcDialectTest}. */
+class OceanBaseOracleDialectTest extends JdbcDialectTest {
- public OceanBaseOracleDialectTypeTest() {
+ public OceanBaseOracleDialectTest() {
ddlFormat =
"CREATE TABLE T (f0 %s)"
+ " WITH ("
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTest.java
similarity index 91%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTest.java
index d4526791..6619b262 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/oracle/dialect/OracleDialectTest.java
@@ -18,13 +18,13 @@
package org.apache.flink.connector.jdbc.databases.oracle.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The Oracle params for {@link JdbcDialectTypeTest}. */
-class OracleDialectTypeTest extends JdbcDialectTypeTest {
+/** The Oracle params for {@link JdbcDialectTest}. */
+class OracleDialectTest extends JdbcDialectTest {
@Override
protected String testDialect() {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/factory/JdbcCatalogFactoryTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/factory/JdbcCatalogFactoryTest.java
index cef7e3b9..0cb69d80 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/factory/JdbcCatalogFactoryTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/catalog/factory/JdbcCatalogFactoryTest.java
@@ -18,9 +18,10 @@
package org.apache.flink.connector.jdbc.databases.postgres.catalog.factory;
-import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
-import org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory;
-import org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions;
+import org.apache.flink.connector.jdbc.core.table.JdbcFactoryLoader;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactory;
+import org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions;
import org.apache.flink.connector.jdbc.databases.postgres.PostgresTestBase;
import org.apache.flink.connector.jdbc.databases.postgres.catalog.PostgresCatalog;
import org.apache.flink.table.catalog.Catalog;
@@ -51,7 +52,7 @@ class JdbcCatalogFactoryTest implements PostgresTestBase {
baseUrl = jdbcUrl.substring(0, jdbcUrl.lastIndexOf("/"));
catalog =
- new JdbcCatalog(
+ JdbcFactoryLoader.loadCatalog(
Thread.currentThread().getContextClassLoader(),
TEST_CATALOG_NAME,
PostgresCatalog.DEFAULT_DATABASE,
@@ -80,14 +81,10 @@ class JdbcCatalogFactoryTest implements PostgresTestBase {
checkEquals(catalog, (JdbcCatalog) actualCatalog);
- assertThat(((JdbcCatalog) actualCatalog).getInternal()).isInstanceOf(PostgresCatalog.class);
+ assertThat((JdbcCatalog) actualCatalog).isInstanceOf(PostgresCatalog.class);
}
private static void checkEquals(JdbcCatalog c1, JdbcCatalog c2) {
- assertThat(c2.getName()).isEqualTo(c1.getName());
- assertThat(c2.getDefaultDatabase()).isEqualTo(c1.getDefaultDatabase());
- assertThat(c2.getUsername()).isEqualTo(c1.getUsername());
- assertThat(c2.getPassword()).isEqualTo(c1.getPassword());
- assertThat(c2.getBaseUrl()).isEqualTo(c1.getBaseUrl());
+ assertThat(c2).isEqualTo(c1);
}
}
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTest.java
similarity index 92%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTest.java
index 1da9b971..a0973a28 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/postgres/dialect/PostgresDialectTest.java
@@ -18,13 +18,13 @@
package org.apache.flink.connector.jdbc.databases.postgres.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The PostgresSql params for {@link JdbcDialectTypeTest}. */
-class PostgresDialectTypeTest extends JdbcDialectTypeTest {
+/** The PostgresSql params for {@link JdbcDialectTest}. */
+class PostgresDialectTest extends JdbcDialectTest {
@Override
protected String testDialect() {
diff --git a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTypeTest.java b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTest.java
similarity index 91%
rename from flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTypeTest.java
rename to flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTest.java
index 6133285a..88a386db 100644
--- a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTypeTest.java
+++ b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/databases/trino/dialect/TrinoDialectTest.java
@@ -18,13 +18,13 @@
package org.apache.flink.connector.jdbc.databases.trino.dialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeTest;
+import org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest;
import java.util.Arrays;
import java.util.List;
-/** The Oracle params for {@link JdbcDialectTypeTest}. */
-class TrinoDialectTypeTest extends JdbcDialectTypeTest {
+/** The Oracle params for {@link JdbcDialectTest}. */
+class TrinoDialectTest extends JdbcDialectTest {
@Override
protected String testDialect() {
diff --git a/flink-connector-jdbc/pom.xml b/flink-connector-jdbc/pom.xml
index 5e507f93..60211302 100644
--- a/flink-connector-jdbc/pom.xml
+++ b/flink-connector-jdbc/pom.xml
@@ -42,20 +42,58 @@ under the License.
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-jdbc-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-annotations</artifactId>
<version>${flink.version}</version>
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
</dependency>
</dependencies>
@@ -95,6 +133,10 @@ under the License.
</excludes>
</filter>
</filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
similarity index 57%
copy from flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
copy to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
index f45f0c09..6a7e1714 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
@@ -16,19 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.connector.jdbc.converter;
-
-import org.apache.flink.connector.jdbc.core.table.dialect.AbstractDialectConverter;
-import org.apache.flink.table.types.logical.RowType;
+package org.apache.flink.connector.jdbc.catalog;
/**
- * Base class for all converters that convert between JDBC object and Flink internal object.
+ * Abstract catalog for any JDBC catalogs.
*
- * @deprecated use AbstractDialectConverter
- * */
+ * @deprecated use org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog
+ */
@Deprecated
-public abstract class AbstractJdbcRowConverter extends AbstractDialectConverter {
- public AbstractJdbcRowConverter(RowType rowType) {
- super(rowType);
+public abstract class AbstractJdbcCatalog
+ extends org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog {
+
+ public AbstractJdbcCatalog(
+ ClassLoader userClassLoader,
+ String catalogName,
+ String defaultDatabase,
+ String username,
+ String pwd,
+ String baseUrl) {
+ super(userClassLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
}
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
similarity index 92%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
rename to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
index 44c0fa64..34688cc2 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.jdbc.catalog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogDatabase;
import org.apache.flink.table.catalog.ObjectPath;
@@ -33,11 +34,17 @@ import java.util.Properties;
import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
-/** Catalogs for relational databases via JDBC. */
+/**
+ * Catalogs for relational databases via JDBC.
+ *
+ * @deprecated user org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalog
+ */
+@Deprecated
@PublicEvolving
-public class JdbcCatalog extends AbstractJdbcCatalog {
+public class JdbcCatalog
+ extends org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog {
- private final AbstractJdbcCatalog internal;
+ private final org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog internal;
@Deprecated
/**
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
similarity index 76%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
rename to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
index af94b9cb..c45cabd8 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
@@ -32,7 +32,12 @@ import java.util.Properties;
import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
import static org.apache.flink.util.Preconditions.checkArgument;
-/** Utils for {@link JdbcCatalog}. */
+/**
+ * Utils for {@link JdbcCatalog}.
+ *
+ * @deprecated
+ */
+@Deprecated
public class JdbcCatalogUtils {
/**
* URL has to be without database, like "jdbc:postgresql://localhost:5432/" or
@@ -75,18 +80,36 @@ public class JdbcCatalogUtils {
JdbcDialect dialect =
JdbcFactoryLoader.loadDialect(baseUrl, userClassLoader, compatibleMode);
+ org.apache.flink.connector.jdbc.core.table.catalog.AbstractJdbcCatalog catalog;
if (dialect instanceof PostgresDialect) {
- return new PostgresCatalog(
- userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
+ catalog =
+ new PostgresCatalog(
+ userClassLoader,
+ catalogName,
+ defaultDatabase,
+ baseUrl,
+ connectionProperties);
} else if (dialect instanceof CrateDBDialect) {
- return new CrateDBCatalog(
- userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
+ catalog =
+ new CrateDBCatalog(
+ userClassLoader,
+ catalogName,
+ defaultDatabase,
+ baseUrl,
+ connectionProperties);
} else if (dialect instanceof MySqlDialect) {
- return new MySqlCatalog(
- userClassLoader, catalogName, defaultDatabase, baseUrl, connectionProperties);
+ catalog =
+ new MySqlCatalog(
+ userClassLoader,
+ catalogName,
+ defaultDatabase,
+ baseUrl,
+ connectionProperties);
} else {
throw new UnsupportedOperationException(
String.format("Catalog for '%s' is not supported yet.", dialect));
}
+
+ return (AbstractJdbcCatalog) catalog;
}
}
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
similarity index 72%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
rename to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
index a19e0bae..5f14840e 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.jdbc.catalog.factory;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
+import org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.FactoryUtil;
@@ -30,18 +31,26 @@ import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Set;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.BASE_URL;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.COMPATIBLE_MODE;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.DEFAULT_DATABASE;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
-import static org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.BASE_URL;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.COMPATIBLE_MODE;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.DEFAULT_DATABASE;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
+import static org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties;
import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
-/** Factory for {@link JdbcCatalog}. */
+/**
+ * Factory for {@link JdbcCatalog}.
+ *
+ * @deprecated use org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactory
+ */
+@Deprecated
public class JdbcCatalogFactory implements CatalogFactory {
- private static final Logger LOG = LoggerFactory.getLogger(JdbcCatalogFactory.class);
+ private static final Logger LOG =
+ LoggerFactory.getLogger(
+ org.apache.flink.connector.jdbc.core.table.catalog.factory.JdbcCatalogFactory
+ .class);
@Override
public String factoryIdentifier() {
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
index f45f0c09..cccf06d3 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
@@ -25,7 +25,7 @@ import org.apache.flink.table.types.logical.RowType;
* Base class for all converters that convert between JDBC object and Flink internal object.
*
* @deprecated use AbstractDialectConverter
- * */
+ */
@Deprecated
public abstract class AbstractJdbcRowConverter extends AbstractDialectConverter {
public AbstractJdbcRowConverter(RowType rowType) {
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
similarity index 67%
copy from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
copy to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
index 11fc46ef..5b598602 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
@@ -18,15 +18,12 @@
package org.apache.flink.connector.jdbc.dialect;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.types.DataType;
+import org.apache.flink.connector.jdbc.core.table.catalog.JdbcCatalogTypeMapper;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-
-/** Separate the jdbc meta-information type to flink table type into the interface. */
-public interface JdbcDialectTypeMapper {
-
- DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)
- throws SQLException;
-}
+/**
+ * Separate the jdbc meta-information type to flink table type into the interface.
+ *
+ * @deprecated use JdbcCatalogTypeMapper
+ */
+@Deprecated
+public interface JdbcDialectTypeMapper extends JdbcCatalogTypeMapper {}
diff --git a/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
similarity index 93%
copy from flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
copy to flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index a54d9f57..44a97bdb 100644
--- a/flink-connector-jdbc-core/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -13,5 +13,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory
org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
similarity index 67%
rename from flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
rename to flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
index 11fc46ef..80edaea5 100644
--- a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
+++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
@@ -18,15 +18,11 @@
package org.apache.flink.connector.jdbc.dialect;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.types.DataType;
-
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-
-/** Separate the jdbc meta-information type to flink table type into the interface. */
-public interface JdbcDialectTypeMapper {
-
- DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)
- throws SQLException;
-}
+/**
+ * Tests for all DataTypes and Dialects of JDBC connector.
+ *
+ * @deprecated use org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest
+ */
+@Deprecated
+public class JdbcDialectTypeTest
+ extends org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest {}
diff --git a/pom.xml b/pom.xml
index 2f41ad2e..107ac34c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -49,6 +49,8 @@ under the License.
<properties>
<flink.version>1.18.0</flink.version>
+ <scala.binary.version>2.12</scala.binary.version>
+ <scala-library.version>2.12.7</scala-library.version>
<jackson-bom.version>2.13.4.20221013</jackson-bom.version>
<junit4.version>4.13.2</junit4.version>
@@ -171,6 +173,13 @@ under the License.
<dependencyManagement>
<dependencies>
+ <!-- For dependency convergence -->
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala-library.version}</version>
+ </dependency>
+
<!-- Flink dependencies -->
<dependency>