You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/05/18 07:25:45 UTC
[rocketmq-connect] branch master updated: Upgrade rocketmq connect jdbc (#139)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
The following commit(s) were added to refs/heads/master by this push:
new 8a72f1c Upgrade rocketmq connect jdbc (#139)
8a72f1c is described below
commit 8a72f1c4038ad9042a0e09aa9bbdd386ebf081d9
Author: xiaoyi <su...@163.com>
AuthorDate: Wed May 18 15:25:39 2022 +0800
Upgrade rocketmq connect jdbc (#139)
* Upgrade rocketmq connect JDBC and be compatible with openmldb JDBC write mode
* add OpenMLDBDatabaseDialect
* Upgrade rocketmq connect API to increase scalability; Add support for openmldb #114
* Update README.md
* delete author information
* remove deprecated method
* reference package expand
* finish debugging openmldb compatibility
* fixed
* reformat code and Check by checkstyle
Co-authored-by: “sunxiaojian” <“sunxiaojian926@163.com”>
---
connectors/rocketmq-connect-jdbc/README.md | 37 +-
connectors/rocketmq-connect-jdbc/pom.xml | 74 +-
.../rocketmq/connect/jdbc/common/Dialect.java | 3 +
.../jdbc/common/JdbcSourceConfigConstants.java | 26 +-
.../connect/jdbc/config/AbstractConfig.java | 26 +-
.../rocketmq/connect/jdbc/config/ConfigDef.java | 33 +-
.../connect/jdbc/connector/JdbcSinkConfig.java | 45 +-
.../connect/jdbc/connector/JdbcSinkConnector.java | 16 +-
.../connect/jdbc/connector/JdbcSinkTask.java | 11 +-
.../connect/jdbc/connector/JdbcSourceConfig.java | 45 +-
.../jdbc/connector/JdbcSourceConnector.java | 22 +-
.../connect/jdbc/connector/JdbcSourceTask.java | 39 +-
.../jdbc/connector/JdbcSourceTaskConfig.java | 22 +-
.../connect/jdbc/dialect/DatabaseDialect.java | 548 +--
.../jdbc/dialect/DatabaseDialectFactory.java | 269 +-
.../rocketmq/connect/jdbc/dialect/DropOptions.java | 134 +-
.../jdbc/dialect/PreparedStatementBinder.java | 156 +-
.../jdbc/dialect/impl/GenericDatabaseDialect.java | 3491 ++++++++++----------
.../jdbc/dialect/impl/MySqlDatabaseDialect.java | 194 +-
.../jdbc/dialect/impl/OpenMLDBDatabaseDialect.java | 55 +-
.../dialect/provider/CachedConnectionProvider.java | 162 +-
.../jdbc/dialect/provider/ConnectionProvider.java | 71 +-
.../dialect/provider/DatabaseDialectProvider.java | 56 +-
.../connect/jdbc/dialect/provider/JdbcUrlInfo.java | 2 +-
.../connect/jdbc/exception/ConfigException.java | 2 +-
.../exception/TableAlterOrCreateException.java | 12 +-
.../jdbc/schema/column/ColumnDefAdjuster.java | 12 +-
.../jdbc/schema/column/ColumnDefinition.java | 560 ++--
.../connect/jdbc/schema/column/ColumnId.java | 126 +-
.../schema/column/parser/DateColumnParser.java | 4 +-
.../schema/column/parser/TimeColumnParser.java | 8 +-
.../connect/jdbc/schema/db/DbStructure.java | 469 +--
.../connect/jdbc/schema/table/TableDefinition.java | 156 +-
.../jdbc/schema/table/TableDefinitions.java | 106 +-
.../connect/jdbc/schema/table/TableId.java | 204 +-
.../connect/jdbc/sink/BufferedRecords.java | 531 +--
.../apache/rocketmq/connect/jdbc/sink/Updater.java | 114 +-
.../connect/jdbc/sink/metadata/FieldsMetadata.java | 362 +-
.../connect/jdbc/sink/metadata/SchemaPair.java | 64 +-
.../jdbc/sink/metadata/SinkRecordField.java | 114 +-
.../jdbc/source/TimestampIncrementingCriteria.java | 546 +--
.../jdbc/source/metadata/ColumnMapping.java | 122 +-
.../jdbc/source/metadata/SchemaMapping.java | 199 +-
.../jdbc/source/offset/SourceOffsetCompute.java | 200 +-
.../source/offset/TimestampIncrementingOffset.java | 140 +-
.../connect/jdbc/source/querier/BulkQuerier.java | 174 +-
.../connect/jdbc/source/querier/Querier.java | 250 +-
.../querier/TimestampIncrementingQuerier.java | 340 +-
.../rocketmq/connect/jdbc/util/BytesUtil.java | 17 +-
.../connect/jdbc/util/ConnectorGroupUtils.java | 4 +-
.../rocketmq/connect/jdbc/util/DateTimeUtils.java | 68 +-
.../connect/jdbc/util/ExpressionBuilder.java | 1061 +++---
.../connect/jdbc/util/IdentifierRules.java | 294 +-
.../rocketmq/connect/jdbc/util/JdbcDriverInfo.java | 202 +-
.../rocketmq/connect/jdbc/util/NumericMapping.java | 29 +-
.../rocketmq/connect/jdbc/util/QuoteMethod.java | 34 +-
.../rocketmq/connect/jdbc/util/TableType.java | 77 +-
.../dialect/DatabaseDialectFactoryTest.java | 5 +-
.../jdbc/connector/sink/OpenMLDBJdbcSinkTest.java | 141 +
.../{ => source}/JdbcSourceConnectorTest.java | 2 +-
.../jdbc/connector/source/JdbcSourceTaskTest.java | 15 +-
61 files changed, 6306 insertions(+), 5995 deletions(-)
diff --git a/connectors/rocketmq-connect-jdbc/README.md b/connectors/rocketmq-connect-jdbc/README.md
index fb8b8f5..cf7ca7b 100644
--- a/connectors/rocketmq-connect-jdbc/README.md
+++ b/connectors/rocketmq-connect-jdbc/README.md
@@ -1,8 +1,11 @@
# rocketmq-connect-jdbc
+
```
注: 目前支持的数据库类型为 mysql ,openMLDB ，其它数据库的jdbc模式持续扩展中
```
+
## rocketmq-connect-jdbc 打包
+
```
mvn clean package -Dmaven.test.skip=true
```
@@ -10,6 +13,7 @@ mvn clean package -Dmaven.test.skip=true
## rocketmq-connect-jdbc 启动
* **jdbc-source-connector** 启动
+
```
POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-source-connector-name}
{
@@ -26,7 +30,6 @@ POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-source-con
}
```
-
* **jdbc-sink-connector** 启动
```
@@ -45,7 +48,7 @@ POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-sink-conne
}
```
->**注：** `rocketmq-jdbc-connect` 的启动依赖于`rocketmq-connect-runtime`项目的启动，需将打好的所有`jar`包放置到`runtime`项目中`pluginPaths`配置的路径后再执行上面的启动请求,该值配置在`runtime`项目下的`connect.conf`文件中
+> **注：** `rocketmq-jdbc-connect` 的启动依赖于`rocketmq-connect-runtime`项目的启动，需将打好的所有`jar`包放置到`runtime`项目中`pluginPaths`配置的路径后再执行上面的启动请求,该值配置在`runtime`项目下的`connect.conf`文件中
## rocketmq-connect-jdbc 停止
@@ -54,8 +57,8 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-connector-name}/
```
## rocketmq-connect-jdbc 参数说明
-* **jdbc-source-connector 参数说明**
+* **jdbc-source-connector 参数说明**
| KEY | TYPE | Must be filled | Description| Example
|------------------------|----|---------|---------------|------------------|
@@ -72,12 +75,13 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-connector-name}/
|table.whitelist | String | YES |需要扫描的表 | db.table,db.table01 |
|max-task | Integer | YES |任务数量，最大不能大于表的数量 | 2 |
|source-record-converter | Integer | YES |data转换器 | org.apache.rocketmq.connect.runtime.converter.JsonConverter |
+
```
注：1.source拉取的数据写入到以表名自动创建的topic中，如果需要写入特定的topic中则需要指定"connect-topicname" 参数
2.topic.prefix参数可以为自动创建的topic增加前缀，用来进行逻辑的隔离
```
-* **jdbc-sink-connector 参数说明**
+* **jdbc-sink-connector 参数说明**
| KEY | TYPE | Must be filled | Description| Example
|------------------------|----|---------|---------------|------------------|
@@ -94,3 +98,28 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-connector-name}/
|source-record-converter | Integer | YES |data转换器 | org.apache.rocketmq.connect.runtime.converter.JsonConverter |
```
+注: openMLDB maven包的引入：
+---------MacOS 系统下运行-------------
+ <dependency>
+ <groupId>com.4paradigm.openmldb</groupId>
+ <artifactId>openmldb-native</artifactId>
+ <version>0.5.0-macos</version>
+ </dependency>
+ <dependency>
+ <groupId>com.4paradigm.openmldb</groupId>
+ <artifactId>openmldb-jdbc</artifactId>
+ <version>0.5.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.4paradigm.openmldb</groupId>
+ <artifactId>openmldb-native</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+---------Linux 系统下运行-------------
+ <dependency>
+ <groupId>com.4paradigm.openmldb</groupId>
+ <artifactId>openmldb-jdbc</artifactId>
+ <version>0.5.0</version>
+ </dependency>
+```
diff --git a/connectors/rocketmq-connect-jdbc/pom.xml b/connectors/rocketmq-connect-jdbc/pom.xml
index 651f5a7..8abed39 100644
--- a/connectors/rocketmq-connect-jdbc/pom.xml
+++ b/connectors/rocketmq-connect-jdbc/pom.xml
@@ -44,7 +44,6 @@
</properties>
-
<build>
<resources>
<resource>
@@ -162,6 +161,27 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <executions>
+ <execution>
+ <id>verify</id>
+ <phase>verify</phase>
+ <configuration>
+ <configLocation>../../style/rmq_checkstyle.xml</configLocation>
+ <encoding>UTF-8</encoding>
+ <consoleOutput>true</consoleOutput>
+ <failsOnError>true</failsOnError>
+ <includeTestSourceDirectory>false</includeTestSourceDirectory>
+ <includeTestResources>false</includeTestResources>
+ </configuration>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
@@ -185,21 +205,21 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>commons-codec</groupId>
- <artifactId>commons-codec</artifactId>
- <version>1.12</version>
- </dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <version>1.12</version>
+ </dependency>
<dependency>
<groupId>io.openmessaging</groupId>
<artifactId>openmessaging-connector</artifactId>
<version>0.1.2</version>
<scope>provided</scope>
</dependency>
- <dependency>
- <groupId>io.openmessaging</groupId>
- <artifactId>openmessaging-api</artifactId>
- <version>0.3.1-alpha</version>
- </dependency>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-api</artifactId>
+ <version>0.3.1-alpha</version>
+ </dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
@@ -246,16 +266,16 @@
<artifactId>commons-cli</artifactId>
<version>1.2</version>
</dependency>
- <dependency>
- <groupId>mysql</groupId>
- <artifactId>mysql-connector-java</artifactId>
- <version>8.0.16</version>
- </dependency>
- <dependency>
- <groupId>io.javalin</groupId>
- <artifactId>javalin</artifactId>
- <version>1.3.0</version>
- </dependency>
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>8.0.16</version>
+ </dependency>
+ <dependency>
+ <groupId>io.javalin</groupId>
+ <artifactId>javalin</artifactId>
+ <version>1.3.0</version>
+ </dependency>
<dependency>
<groupId>com.github.shyiko</groupId>
@@ -271,13 +291,19 @@
<dependency>
<groupId>com.4paradigm.openmldb</groupId>
- <artifactId>openmldb-jdbc</artifactId>
- <version>0.4.4-hotfix1</version>
+ <artifactId>openmldb-native</artifactId>
+ <version>0.5.0-macos</version>
</dependency>
<dependency>
<groupId>com.4paradigm.openmldb</groupId>
- <artifactId>openmldb-native</artifactId>
- <version>0.4.4-hotfix1</version>
+ <artifactId>openmldb-jdbc</artifactId>
+ <version>0.5.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.4paradigm.openmldb</groupId>
+ <artifactId>openmldb-native</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
</dependencies>
</project>
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/Dialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/Dialect.java
index c164462..f6f7dd2 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/Dialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/Dialect.java
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.rocketmq.connect.jdbc.common;
import java.lang.annotation.ElementType;
@@ -29,12 +30,14 @@ import java.lang.annotation.Target;
public @interface Dialect {
/**
* dialect name
+ *
* @return
*/
String name() default "";
/**
* support protocol
+ *
* @return
*/
String protocol() default "";
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/JdbcSourceConfigConstants.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/JdbcSourceConfigConstants.java
index fbe00ec..3b1039b 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/JdbcSourceConfigConstants.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/JdbcSourceConfigConstants.java
@@ -19,21 +19,21 @@ package org.apache.rocketmq.connect.jdbc.common;
import org.apache.commons.lang3.StringUtils;
public class JdbcSourceConfigConstants {
- private static final String TABLE_NAME_KEY_PREFIX = "table";
- private static final String QUERY_NAME_KEY_PREFIX = "query";
- public static final String QUERY_NAME_VALUE = "query";
+ private static final String TABLE_NAME_KEY_PREFIX = "table";
+ private static final String QUERY_NAME_KEY_PREFIX = "query";
+ public static final String QUERY_NAME_VALUE = "query";
- public static String TABLE_NAME_KEY(String suffix){
- if(StringUtils.isEmpty(suffix)){
- return TABLE_NAME_KEY_PREFIX;
+ public static String TABLE_NAME_KEY(String suffix) {
+ if (StringUtils.isEmpty(suffix)) {
+ return TABLE_NAME_KEY_PREFIX;
+ }
+ return TABLE_NAME_KEY_PREFIX + "-" + suffix;
}
- return TABLE_NAME_KEY_PREFIX + "-" + suffix;
- }
- public static String QUERY_NAME_KEY(String suffix){
- if(StringUtils.isEmpty(suffix)){
- return QUERY_NAME_KEY_PREFIX;
+ public static String QUERY_NAME_KEY(String suffix) {
+ if (StringUtils.isEmpty(suffix)) {
+ return QUERY_NAME_KEY_PREFIX;
+ }
+ return QUERY_NAME_KEY_PREFIX + "-" + suffix;
}
- return QUERY_NAME_KEY_PREFIX + "-" + suffix;
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/AbstractConfig.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/AbstractConfig.java
index 6daf89e..e4fd758 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/AbstractConfig.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/AbstractConfig.java
@@ -66,13 +66,13 @@ public abstract class AbstractConfig {
private Long backoffMs;
private String quoteSqlIdentifiers;
- public AbstractConfig(KeyValue config){
- connectionDbUrl=config.getString(CONNECTION_URL_CONFIG);
- connectionDbUser=config.getString(CONNECTION_USER_CONFIG);
- connectionDbPassword=config.getString(CONNECTION_PASSWORD_CONFIG);
- attempts=config.getInt(CONNECTION_ATTEMPTS_CONFIG,CONNECTION_ATTEMPTS_DEFAULT);
- backoffMs=config.getLong(CONNECTION_BACKOFF_CONFIG,CONNECTION_BACKOFF_DEFAULT);
- quoteSqlIdentifiers=config.getString(QUOTE_SQL_IDENTIFIERS_CONFIG,QUOTE_SQL_IDENTIFIERS_DEFAULT);
+ public AbstractConfig(KeyValue config) {
+ connectionDbUrl = config.getString(CONNECTION_URL_CONFIG);
+ connectionDbUser = config.getString(CONNECTION_USER_CONFIG);
+ connectionDbPassword = config.getString(CONNECTION_PASSWORD_CONFIG);
+ attempts = config.getInt(CONNECTION_ATTEMPTS_CONFIG, CONNECTION_ATTEMPTS_DEFAULT);
+ backoffMs = config.getLong(CONNECTION_BACKOFF_CONFIG, CONNECTION_BACKOFF_DEFAULT);
+ quoteSqlIdentifiers = config.getString(QUOTE_SQL_IDENTIFIERS_CONFIG, QUOTE_SQL_IDENTIFIERS_DEFAULT);
}
@@ -102,12 +102,13 @@ public abstract class AbstractConfig {
/**
* get list
+ *
* @param config
* @param key
* @return
*/
- protected List<String> getList(KeyValue config, String key){
- if (!config.containsKey(key) || Objects.isNull(config.getString(key))){
+ protected List<String> getList(KeyValue config, String key) {
+ if (!config.containsKey(key) || Objects.isNull(config.getString(key))) {
return Lists.newArrayList();
}
return Arrays.asList(COMMA_WITH_WHITESPACE.split(config.getString(key), -1));
@@ -115,18 +116,19 @@ public abstract class AbstractConfig {
/**
* get list
+ *
* @param config
* @param key
* @return
*/
- protected List<String> getList(KeyValue config, String key,String defaultValue){
- if (config.containsKey(key) ||Objects.isNull(config.getString(key))){
+ protected List<String> getList(KeyValue config, String key, String defaultValue) {
+ if (config.containsKey(key) || Objects.isNull(config.getString(key))) {
return Lists.newArrayList(defaultValue);
}
return Arrays.asList(COMMA_WITH_WHITESPACE.split(config.getString(key), -1));
}
- protected Boolean getBoolean(KeyValue config, String key,Boolean defaultValue){
+ protected Boolean getBoolean(KeyValue config, String key, Boolean defaultValue) {
return config.containsKey(key) ? Boolean.getBoolean(config.getString(key)) : defaultValue;
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/ConfigDef.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/ConfigDef.java
index c54ce11..504e225 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/ConfigDef.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/config/ConfigDef.java
@@ -45,6 +45,7 @@ public class ConfigDef {
public static final Object NO_DEFAULT_VALUE = new Object();
private final Map<String, ConfigKey> configKeys;
+
public ConfigDef() {
configKeys = new LinkedHashMap<>();
}
@@ -59,7 +60,7 @@ public class ConfigDef {
}
/**
- * 不验证
+ * no validate
* @param name
* @param type
* @param defaultValue
@@ -67,12 +68,12 @@ public class ConfigDef {
* @param documentation
* @return
*/
- public ConfigDef define(String name, Type type, Object defaultValue, Validator validator, String documentation) {
+ public ConfigDef define(String name, Type type, Object defaultValue, Validator validator, String documentation) {
return define(new ConfigKey(name, type, defaultValue, validator, documentation));
}
/**
- * 加验证
+ * validate config
* @param name
* @param type
* @param defaultValue
@@ -86,6 +87,7 @@ public class ConfigDef {
/**
* Get the configuration keys
+ *
* @return a map containing all configuration keys
*/
public Map<String, ConfigKey> configKeys() {
@@ -94,6 +96,7 @@ public class ConfigDef {
/**
* parse props
+ *
* @param props
* @return
*/
@@ -108,6 +111,7 @@ public class ConfigDef {
/**
* convert + validate
+ *
* @param key
* @param value
* @param isSet
@@ -130,9 +134,9 @@ public class ConfigDef {
}
-
/**
* Parse a value according to its expected type.
+ *
* @param name The config name
* @param value The config value
* @param type The expected type
@@ -260,16 +264,14 @@ public class ConfigDef {
}
-
-
-
/**
* Validation logic the user may provide to perform single configuration validation.
*/
public interface Validator {
/**
* Perform single configuration validation.
- * @param name The name of the configuration
+ *
+ * @param name The name of the configuration
* @param value The value of the configuration
* @throws ConfigException if the value is invalid.
*/
@@ -282,17 +284,21 @@ public class ConfigDef {
public static class Range implements Validator {
private final Number min;
private final Number max;
+
/**
- * A numeric range with inclusive upper bound and inclusive lower bound
- * @param min the lower bound
- * @param max the upper bound
+ * A numeric range with inclusive upper bound and inclusive lower bound
+ *
+ * @param min the lower bound
+ * @param max the upper bound
*/
private Range(Number min, Number max) {
this.min = min;
this.max = max;
}
+
/**
* A numeric range that checks only the lower bound
+ *
* @param min The minimum acceptable value
*/
public static Range atLeast(Number min) {
@@ -469,7 +475,7 @@ public class ConfigDef {
@Override
public void ensureValid(String name, Object value) {
- for (Validator validator: validators) {
+ for (Validator validator : validators) {
validator.ensureValid(name, value);
}
}
@@ -480,7 +486,7 @@ public class ConfigDef {
return "";
}
StringBuilder desc = new StringBuilder();
- for (Validator v: validators) {
+ for (Validator v : validators) {
if (desc.length() > 0) {
desc.append(',').append(' ');
}
@@ -511,6 +517,7 @@ public class ConfigDef {
public static NonEmptyStringWithoutControlChars nonEmptyStringWithoutControlChars() {
return new NonEmptyStringWithoutControlChars();
}
+
@Override
public void ensureValid(String name, Object value) {
String s = (String) value;
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConfig.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConfig.java
index d211a23..7c1f0b3 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConfig.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConfig.java
@@ -72,7 +72,7 @@ public class JdbcSinkConfig extends AbstractConfig {
private static final int RETRY_BACKOFF_MS_DEFAULT = 3000;
public static final String BATCH_SIZE = "batch.size";
- private static final int BATCH_SIZE_DEFAULT = 3000;
+ private static final int BATCH_SIZE_DEFAULT = 100;
public static final String DELETE_ENABLED = "delete.enabled";
@@ -184,27 +184,27 @@ public class JdbcSinkConfig extends AbstractConfig {
public JdbcSinkConfig(KeyValue config) {
super(config);
- tableNameFormat = config.getString(TABLE_NAME_FORMAT,TABLE_NAME_FORMAT_DEFAULT).trim();
- tableFromHeader=getBoolean(config,TABLE_NAME_FROM_HEADER,false);
- batchSize = config.getInt(BATCH_SIZE);
- deleteEnabled = getBoolean(config,DELETE_ENABLED,DELETE_ENABLED_DEFAULT);
- maxRetries = config.getInt(MAX_RETRIES,MAX_RETRIES_DEFAULT);
- retryBackoffMs =config.getInt(RETRY_BACKOFF_MS,RETRY_BACKOFF_MS_DEFAULT);
- autoCreate = getBoolean(config,AUTO_CREATE,AUTO_CREATE_DEFAULT);
- autoEvolve = getBoolean(config,AUTO_EVOLVE,AUTO_EVOLVE_DEFAULT);
- if (Objects.nonNull(config.getString(INSERT_MODE))){
- insertMode = InsertMode.valueOf(config.getString(INSERT_MODE,INSERT_MODE_DEFAULT).toUpperCase());
+ tableNameFormat = config.getString(TABLE_NAME_FORMAT, TABLE_NAME_FORMAT_DEFAULT).trim();
+ tableFromHeader = getBoolean(config, TABLE_NAME_FROM_HEADER, false);
+ batchSize = config.getInt(BATCH_SIZE, BATCH_SIZE_DEFAULT);
+ deleteEnabled = getBoolean(config, DELETE_ENABLED, DELETE_ENABLED_DEFAULT);
+ maxRetries = config.getInt(MAX_RETRIES, MAX_RETRIES_DEFAULT);
+ retryBackoffMs = config.getInt(RETRY_BACKOFF_MS, RETRY_BACKOFF_MS_DEFAULT);
+ autoCreate = getBoolean(config, AUTO_CREATE, AUTO_CREATE_DEFAULT);
+ autoEvolve = getBoolean(config, AUTO_EVOLVE, AUTO_EVOLVE_DEFAULT);
+ if (Objects.nonNull(config.getString(INSERT_MODE))) {
+ insertMode = InsertMode.valueOf(config.getString(INSERT_MODE, INSERT_MODE_DEFAULT).toUpperCase());
}
- pkMode = PrimaryKeyMode.valueOf(config.getString(PK_MODE,PK_MODE_DEFAULT).toUpperCase());
- pkFields = getList(config,PK_FIELDS);
+ pkMode = PrimaryKeyMode.valueOf(config.getString(PK_MODE, PK_MODE_DEFAULT).toUpperCase());
+ pkFields = getList(config, PK_FIELDS);
dialectName = config.getString(DIALECT_NAME_CONFIG);
- fieldsWhitelist = new HashSet<>(getList(config,FIELDS_WHITELIST));
+ fieldsWhitelist = new HashSet<>(getList(config, FIELDS_WHITELIST));
// table white list
- tableWhitelist =new HashSet<>(getList(config,TABLE_WHITE_LIST_CONFIG));
- String dbTimeZone = config.getString(DB_TIMEZONE_CONFIG,DB_TIMEZONE_DEFAULT);
+ tableWhitelist = new HashSet<>(getList(config, TABLE_WHITE_LIST_CONFIG));
+ String dbTimeZone = config.getString(DB_TIMEZONE_CONFIG, DB_TIMEZONE_DEFAULT);
timeZone = TimeZone.getTimeZone(ZoneId.of(dbTimeZone));
- tableTypes = TableType.parse(getList(config,TABLE_TYPES_CONFIG,TABLE_TYPES_DEFAULT));
+ tableTypes = TableType.parse(getList(config, TABLE_TYPES_CONFIG, TABLE_TYPES_DEFAULT));
}
@@ -274,21 +274,22 @@ public class JdbcSinkConfig extends AbstractConfig {
/**
* filter white table
+ *
* @param dbDialect
* @param tableId
* @return
*/
public boolean filterWhiteTable(DatabaseDialect dbDialect, TableId tableId) {
// not filter table
- if (tableWhitelist.isEmpty()){
+ if (tableWhitelist.isEmpty()) {
return true;
}
- for (String tableName: tableWhitelist){
- TableId table=dbDialect.parseToTableId(tableName);
- if(table.catalogName() != null && table.catalogName().equals(tableId.catalogName())){
+ for (String tableName : tableWhitelist) {
+ TableId table = dbDialect.parseToTableId(tableName);
+ if (table.catalogName() != null && table.catalogName().equals(tableId.catalogName())) {
return true;
}
- if (table.tableName().equals(tableId.tableName())){
+ if (table.tableName().equals(tableId.tableName())) {
return true;
}
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConnector.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConnector.java
index 0fbbf8d..aaf8560 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConnector.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkConnector.java
@@ -38,11 +38,12 @@ public class JdbcSinkConnector extends SinkConnector {
@Override
public void start(ConnectorContext context) {
- this.context=context;
+ this.context = context;
}
/**
* Should invoke before start the connector.
+ *
* @param config
* @return error message
*/
@@ -58,17 +59,20 @@ public class JdbcSinkConnector extends SinkConnector {
*/
@Override
public void init(KeyValue config) {
- this.connectConfig=config;
+ this.connectConfig = config;
}
@Override
- public void stop() {}
+ public void stop() {
+ }
@Override
- public void pause() {}
+ public void pause() {
+ }
@Override
- public void resume() {}
+ public void resume() {
+ }
/**
* Returns a set of configurations for Tasks based on the current configuration,
@@ -81,7 +85,7 @@ public class JdbcSinkConnector extends SinkConnector {
public List<KeyValue> taskConfigs(int maxTasks) {
log.info("Starting task config !!! ");
List<KeyValue> configs = new ArrayList<>();
- for (int i =0 ; i< maxTasks; i++){
+ for (int i = 0; i < maxTasks; i++) {
configs.add(this.connectConfig);
}
return configs;
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkTask.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkTask.java
index 4fd844a..d69c80a 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkTask.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSinkTask.java
@@ -51,6 +51,7 @@ public class JdbcSinkTask extends SinkTask {
/**
* Put the records to the sink
+ *
* @param records
*/
@Override
@@ -89,7 +90,7 @@ public class JdbcSinkTask extends SinkTask {
@Override
public void start(SinkTaskContext context) {
- this.context=context;
+ this.context = context;
}
/**
@@ -105,6 +106,7 @@ public class JdbcSinkTask extends SinkTask {
/**
* Init the component
+ *
* @param keyValue
*/
@Override
@@ -141,8 +143,11 @@ public class JdbcSinkTask extends SinkTask {
}
@Override
- public void pause() {}
+ public void pause() {
+ }
+
@Override
- public void resume() {}
+ public void resume() {
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConfig.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConfig.java
index bc3028f..3c34c5e 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConfig.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConfig.java
@@ -38,11 +38,12 @@ public class JdbcSourceConfig extends AbstractConfig {
public enum TableLoadMode {
MODE_BULK("bulk"),
MODE_TIMESTAMP("timestamp"),
- MODE_INCREMENTING( "incrementing"),
+ MODE_INCREMENTING("incrementing"),
MODE_TIMESTAMP_INCREMENTING("timestamp+incrementing");
private String name;
+
TableLoadMode(String name) {
- this.name=name;
+ this.name = name;
}
public String getName() {
@@ -52,13 +53,14 @@ public class JdbcSourceConfig extends AbstractConfig {
public void setName(String name) {
this.name = name;
}
- public static TableLoadMode findTableLoadModeByName(String name){
- for (TableLoadMode mode : TableLoadMode.values()){
- if (mode.getName().equals(name)){
+
+ public static TableLoadMode findTableLoadModeByName(String name) {
+ for (TableLoadMode mode : TableLoadMode.values()) {
+ if (mode.getName().equals(name)) {
return mode;
}
}
- throw new IllegalArgumentException("Unsupports mode "+ name);
+ throw new IllegalArgumentException("Unsupports mode " + name);
}
}
@@ -217,30 +219,31 @@ public class JdbcSourceConfig extends AbstractConfig {
private TimeZone timeZone;
private String offsetSuffix;
private String querySuffix;
+
public JdbcSourceConfig(KeyValue config) {
super(config);
- this.pollIntervalMs = config.getInt(POLL_INTERVAL_MS_CONFIG,POLL_INTERVAL_MS_DEFAULT);
- this.batchMaxRows = config.getInt(BATCH_MAX_ROWS_CONFIG,BATCH_MAX_ROWS_DEFAULT);
- this.numericPrecisionMapping = getBoolean(config,NUMERIC_PRECISION_MAPPING_CONFIG,NUMERIC_PRECISION_MAPPING_DEFAULT);
- this.numericMapping = config.getString(NUMERIC_MAPPING_CONFIG,NUMERIC_MAPPING_DEFAULT);
- this.dialectName = config.getString(DIALECT_NAME_CONFIG, DIALECT_NAME_DEFAULT );
+ this.pollIntervalMs = config.getInt(POLL_INTERVAL_MS_CONFIG, POLL_INTERVAL_MS_DEFAULT);
+ this.batchMaxRows = config.getInt(BATCH_MAX_ROWS_CONFIG, BATCH_MAX_ROWS_DEFAULT);
+ this.numericPrecisionMapping = getBoolean(config, NUMERIC_PRECISION_MAPPING_CONFIG, NUMERIC_PRECISION_MAPPING_DEFAULT);
+ this.numericMapping = config.getString(NUMERIC_MAPPING_CONFIG, NUMERIC_MAPPING_DEFAULT);
+ this.dialectName = config.getString(DIALECT_NAME_CONFIG, DIALECT_NAME_DEFAULT);
this.mode = config.getString(MODE_CONFIG);
this.incrementingColumnName = config.getString(INCREMENTING_COLUMN_NAME_CONFIG);
- this.timestampColumnNames = getList(config,TIMESTAMP_COLUMN_NAME_CONFIG);
- timestampDelayIntervalMs=config.getLong(TIMESTAMP_DELAY_INTERVAL_MS_CONFIG);
+ this.timestampColumnNames = getList(config, TIMESTAMP_COLUMN_NAME_CONFIG);
+ timestampDelayIntervalMs = config.getLong(TIMESTAMP_DELAY_INTERVAL_MS_CONFIG);
this.timestampInitial = config.containsKey(TIMESTAMP_INITIAL_CONFIG) ? config.getLong(TIMESTAMP_INITIAL_CONFIG) : TIMESTAMP_INITIAL_DEFAULT;
- this.tableWhitelist = new HashSet<>(getList(config,TABLE_WHITELIST_CONFIG));
- this.tableBlacklist = new HashSet<>(getList(config,TABLE_BLACKLIST_CONFIG));
+ this.tableWhitelist = new HashSet<>(getList(config, TABLE_WHITELIST_CONFIG));
+ this.tableBlacklist = new HashSet<>(getList(config, TABLE_BLACKLIST_CONFIG));
this.schemaPattern = config.getString(SCHEMA_PATTERN_CONFIG);
this.catalogPattern = config.getString(CATALOG_PATTERN_CONFIG);
this.query = config.getString(QUERY_CONFIG);
this.topicPrefix = config.getString(TOPIC_PREFIX_CONFIG);
- this.validateNonNull = getBoolean(config,VALIDATE_NON_NULL_CONFIG,VALIDATE_NON_NULL_DEFAULT);
- tableTypes = TableType.parse(getList(config,TABLE_TYPE_CONFIG,TABLE_TYPE_DEFAULT)) ;
- String dbTimeZone =config.getString(DB_TIMEZONE_CONFIG,DB_TIMEZONE_DEFAULT);
- this.timeZone=TimeZone.getTimeZone(ZoneId.of(dbTimeZone));
- this.querySuffix=config.getString(QUERY_SUFFIX_CONFIG,QUERY_SUFFIX_DEFAULT);
- this.offsetSuffix=config.getString(OFFSET_SUFFIX_CONFIG,OFFSET_SUFFIX_DEFAULT);
+ this.validateNonNull = getBoolean(config, VALIDATE_NON_NULL_CONFIG, VALIDATE_NON_NULL_DEFAULT);
+ tableTypes = TableType.parse(getList(config, TABLE_TYPE_CONFIG, TABLE_TYPE_DEFAULT));
+ String dbTimeZone = config.getString(DB_TIMEZONE_CONFIG, DB_TIMEZONE_DEFAULT);
+ this.timeZone = TimeZone.getTimeZone(ZoneId.of(dbTimeZone));
+ this.querySuffix = config.getString(QUERY_SUFFIX_CONFIG, QUERY_SUFFIX_DEFAULT);
+ this.offsetSuffix = config.getString(OFFSET_SUFFIX_CONFIG, OFFSET_SUFFIX_DEFAULT);
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConnector.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConnector.java
index f7f55eb..c48da71 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConnector.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceConnector.java
@@ -55,8 +55,8 @@ public class JdbcSourceConnector extends SourceConnector {
*/
@Override
public void init(KeyValue config) {
- if (config.containsKey("connect-topicname")){
- config.put("connect-topicname","");
+ if (config.containsKey("connect-topicname")) {
+ config.put("connect-topicname", "");
}
originalConfig = config;
}
@@ -83,23 +83,23 @@ public class JdbcSourceConnector extends SourceConnector {
@Override
public List<KeyValue> taskConfigs(int maxTasks) {
- log.info("Connector task config divide["+maxTasks+"]" );
- List<KeyValue> keyValues=Lists.newArrayList();
+ log.info("Connector task config divide[" + maxTasks + "]");
+ List<KeyValue> keyValues = Lists.newArrayList();
List<String> tables = Lists.newArrayList();
- log.info("Connector table white list["+jdbcSourceConfig.getTableWhitelist()+"]" );
- jdbcSourceConfig.getTableWhitelist().forEach(table->{
+ log.info("Connector table white list[" + jdbcSourceConfig.getTableWhitelist() + "]");
+ jdbcSourceConfig.getTableWhitelist().forEach(table -> {
tables.add(table);
});
- maxTasks = tables.size() > maxTasks ? maxTasks: tables.size();
+ maxTasks = tables.size() > maxTasks ? maxTasks : tables.size();
List<List<String>> tablesGrouped =
ConnectorGroupUtils.groupPartitions(tables, maxTasks);
- for (List<String> tableGroup:tablesGrouped) {
+ for (List<String> tableGroup : tablesGrouped) {
KeyValue keyValue = new DefaultKeyValue();
- for (String key: originalConfig.keySet()){
- keyValue.put(key,originalConfig.getString(key));
+ for (String key : originalConfig.keySet()) {
+ keyValue.put(key, originalConfig.getString(key));
}
- keyValue.put(JdbcSourceTaskConfig.TABLES_CONFIG,StringUtils.join(tableGroup,","));
+ keyValue.put(JdbcSourceTaskConfig.TABLES_CONFIG, StringUtils.join(tableGroup, ","));
keyValues.add(keyValue);
}
return keyValues;
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTask.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTask.java
index 6ea61b1..80a7117 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTask.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTask.java
@@ -1,4 +1,3 @@
-
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -21,7 +20,7 @@ package org.apache.rocketmq.connect.jdbc.connector;
import io.openmessaging.KeyValue;
import io.openmessaging.connector.api.component.task.source.SourceTask;
import io.openmessaging.connector.api.component.task.source.SourceTaskContext;
-import io.openmessaging.connector.api.data.*;
+import io.openmessaging.connector.api.data.ConnectRecord;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.connect.jdbc.dialect.DatabaseDialect;
import org.apache.rocketmq.connect.jdbc.dialect.DatabaseDialectFactory;
@@ -65,7 +64,6 @@ public class JdbcSourceTask extends SourceTask {
@Override
public List<ConnectRecord> poll() {
log.trace(" Polling for new data");
- // 连续空的次数
Map<Querier, Integer> consecutiveEmptyResults = tableQueue.stream().collect(Collectors.toMap(Function.identity(), (q) -> 0));
while (running.get()) {
final Querier querier = tableQueue.peek();
@@ -138,9 +136,9 @@ public class JdbcSourceTask extends SourceTask {
private void resetAndRequeueHead(Querier querier) {
log.debug("Resetting querier {}", querier.toString());
tableQueue.poll();
- if(running.get()){
+ if (running.get()) {
querier.reset(System.currentTimeMillis());
- }else{ //不跑了，可能是暂停，也可能是停止
+ } else {
querier.reset(0);
}
tableQueue.add(querier);
@@ -149,21 +147,24 @@ public class JdbcSourceTask extends SourceTask {
/**
* Should invoke before start the connector.
+ *
* @param config
* @return error message
*/
@Override
- public void validate(KeyValue config) { }
+ public void validate(KeyValue config) {
+ }
/**
* start jdbc task
+ *
* @param context
*/
@Override
public void start(SourceTaskContext context) {
// compute table offset
- Map<String,Map<String, Object>> offsetValues= SourceOffsetCompute.initOffset(config, context, dialect, cachedConnectionProvider);
- for (String tableOrQuery: offsetValues.keySet()) {
+ Map<String, Map<String, Object>> offsetValues = SourceOffsetCompute.initOffset(config, context, dialect, cachedConnectionProvider);
+ for (String tableOrQuery : offsetValues.keySet()) {
this.buildAndAddQuerier(
JdbcSourceConfig.TableLoadMode.findTableLoadModeByName(this.config.getMode()),
this.config.getQuerySuffix(),
@@ -180,6 +181,7 @@ public class JdbcSourceTask extends SourceTask {
/**
* build and add querier
+ *
* @param loadMode
* @param querySuffix
* @param incrementingColumn
@@ -193,7 +195,7 @@ public class JdbcSourceTask extends SourceTask {
String topicPrefix = config.getTopicPrefix();
Querier.QueryMode queryMode = !StringUtils.isEmpty(config.getQuery()) ? Querier.QueryMode.QUERY : Querier.QueryMode.TABLE;
Querier querier = null;
- switch (loadMode){
+ switch (loadMode) {
case MODE_BULK:
querier = new BulkQuerier(
dialect,
@@ -222,7 +224,7 @@ public class JdbcSourceTask extends SourceTask {
tableQueue.add(querier);
break;
case MODE_TIMESTAMP:
- querier = new TimestampIncrementingQuerier(
+ querier = new TimestampIncrementingQuerier(
dialect,
queryMode,
tableOrQuery,
@@ -238,7 +240,7 @@ public class JdbcSourceTask extends SourceTask {
tableQueue.add(querier);
break;
case MODE_TIMESTAMP_INCREMENTING:
- querier = new TimestampIncrementingQuerier(
+ querier = new TimestampIncrementingQuerier(
dialect,
queryMode,
tableOrQuery,
@@ -249,7 +251,7 @@ public class JdbcSourceTask extends SourceTask {
timestampDelayInterval,
timeZone,
querySuffix,
- this.config.getOffsetSuffix()
+ this.config.getOffsetSuffix()
);
tableQueue.add(querier);
break;
@@ -257,13 +259,14 @@ public class JdbcSourceTask extends SourceTask {
}
/**
- * Init the component
- * @param props
- */
+ * Init the component
+ *
+ * @param props
+ */
@Override
public void init(KeyValue props) {
try {
- config=new JdbcSourceTaskConfig(props);
+ config = new JdbcSourceTaskConfig(props);
final String dialectName = config.getDialectName();
final String url = config.getConnectionDbUrl();
if (dialectName != null && !dialectName.trim().isEmpty()) {
@@ -321,11 +324,11 @@ public class JdbcSourceTask extends SourceTask {
@Override
public void pause() {
- //TODO do nothing
+ // do nothing
}
@Override
public void resume() {
- //TODO do nothing
+ // do nothing
}
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTaskConfig.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTaskConfig.java
index d0a2d8e..001beb8 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTaskConfig.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/connector/JdbcSourceTaskConfig.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.connect.jdbc.connector;
import io.openmessaging.KeyValue;
+
import java.util.List;
@@ -24,16 +25,17 @@ import java.util.List;
* task config
*/
public class JdbcSourceTaskConfig extends JdbcSourceConfig {
- public static final String TABLES_CONFIG = "tables";
- private static final String TABLES_DOC = "List of tables for this task to watch for changes.";
+ public static final String TABLES_CONFIG = "tables";
+ private static final String TABLES_DOC = "List of tables for this task to watch for changes.";
+
+ private List<String> tables;
- private List<String> tables;
- public JdbcSourceTaskConfig(KeyValue config) {
- super(config);
- this.tables=getList(config,TABLES_CONFIG);
- }
+ public JdbcSourceTaskConfig(KeyValue config) {
+ super(config);
+ this.tables = getList(config, TABLES_CONFIG);
+ }
- public List<String> getTables() {
- return tables;
- }
+ public List<String> getTables() {
+ return tables;
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialect.java
index fc15fa8..61c9d6e 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialect.java
@@ -49,271 +49,299 @@ import java.util.Map;
*/
public interface DatabaseDialect extends ConnectionProvider {
- /**
- * dialect name
- * @return
- */
- String name();
-
- /**
- * create jdbc prepared statement
- * @param connection
- * @param query
- * @return
- * @throws SQLException
- */
- PreparedStatement createPreparedStatement(Connection connection, String query) throws SQLException;
-
- /**
- * parse to Table Id
- * @param fqn
- * @return
- */
- TableId parseToTableId(String fqn);
-
- /**
- * Get the identifier rules for this database.
- * @return the identifier rules
- */
- IdentifierRules identifierRules();
-
- /**
- * Get a new expression builder that can be used to build expressions with quoted identifiers.
- * @return
- */
- ExpressionBuilder expressionBuilder();
- Timestamp currentTimeOnDB(Connection connection, Calendar cal) throws SQLException;
-
- /**
- * Get a list of identifiers of the non-system tables in the database.
- * @param connection
- * @return
- * @throws SQLException
- */
- List<TableId> tableIds(Connection connection) throws SQLException;
-
- /**
- * table exists
- * @param connection
- * @param tableId
- * @return
- * @throws SQLException
- */
- boolean tableExists(Connection connection, TableId tableId) throws SQLException;
-
-
- /**
- * Create the definition for the columns described by the database metadata.
- */
- Map<ColumnId, ColumnDefinition> describeColumns(Connection connection, String tablePattern, String columnPattern) throws SQLException;
-
- /**
- * Create the definition for the columns described by the database metadata.
- */
- Map<ColumnId, ColumnDefinition> describeColumns(Connection connection, String catalogPattern, String schemaPattern, String tablePattern, String columnPattern) throws SQLException;
-
- /**
- * Create the definition for the columns in the result set.
- */
- Map<ColumnId, ColumnDefinition> describeColumns(Connection conn, TableId tableId, ResultSetMetaData rsMetadata) throws SQLException;
-
- /**
- * describe table info
- * @param connection
- * @param tableId
- * @return
- * @throws SQLException
- */
- TableDefinition describeTable(Connection connection, TableId tableId) throws SQLException;
-
- /**
- * describe columns by query sql
- * @param connection
- * @param tableId
- * @return
- * @throws SQLException
- */
- Map<ColumnId, ColumnDefinition> describeColumnsByQuerying(Connection connection, TableId tableId) throws SQLException;
-
- /**
- * 添加字段到schema中
- * @param column
- * @param schema
- * @param index
- * @return
- */
- String addFieldToSchema(ColumnDefinition column, Schema schema,int index);
-
- /**
- * Apply the supplied DDL statements using the given connection. This gives the dialect the
- * opportunity to execute the statements with a different autocommit setting.
- */
- void applyDdlStatements(Connection connection, List<String> statements) throws SQLException;
-
- /**
- * build dml statement
- * @param table
- * @param keyColumns
- * @param nonKeyColumns
- * @return
- */
- String buildInsertStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
-
- /**
- * update statement
- * @param table
- * @param keyColumns
- * @param nonKeyColumns
- * @return
- */
- String buildUpdateStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
-
- /**
- * upsert statment
- * @param table
- * @param keyColumns
- * @param nonKeyColumns
- * @return
- */
- String buildUpsertQueryStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
-
- /**
- * delete statement
- * @param table
- * @param keyColumns
- * @return
- */
- default String buildDeleteStatement(TableId table, Collection<ColumnId> keyColumns) {
- throw new UnsupportedOperationException();
- }
-
-
- /**
- * build select table
- */
- String buildSelectTableMode();
- void buildSelectTable(ExpressionBuilder builder, TableId tableId);
-
-
- /** drop table
- * @param table
- * @param options
- * @return
- */
- String buildDropTableStatement(TableId table, DropOptions options);
-
- /**
- * create table
- * @param table
- * @param fields
- * @return
- */
- String buildCreateTableStatement(TableId table, Collection<SinkRecordField> fields);
- /**
- * build alter table
- * @param table
- * @param fields
- * @return
- */
- List<String> buildAlterTable(TableId table, Collection<SinkRecordField> fields);
-
- /**
- * Create a component that can bind record values into the supplied prepared statement.
- *
- * @param statement the prepared statement
- * @param pkMode the primary key mode; may not be null
- * @param schemaPair the key and value schemas; may not be null
- * @param fieldsMetadata the field metadata; may not be null
- * @param tableDefinition the table definition; may be null
- * @param insertMode the insert mode; may not be null
- * @return the statement binder; may not be null
- */
- StatementBinder statementBinder(
- PreparedStatement statement,
- JdbcSinkConfig.PrimaryKeyMode pkMode,
- SchemaPair schemaPair,
- FieldsMetadata fieldsMetadata,
- TableDefinition tableDefinition,
- JdbcSinkConfig.InsertMode insertMode
- );
-
- /**
- * value column types
- * @param rsMetadata
- * @param columns
- * @throws io.openmessaging.connector.api.errors.ConnectException
- */
- void validateColumnTypes(
- ResultSetMetaData rsMetadata,
- List<ColumnId> columns
- ) throws io.openmessaging.connector.api.errors.ConnectException;
-
-
- /**
- * bind field
- * @param statement
- * @param index
- * @param schema
- * @param value
- * @param colDef
- * @throws SQLException
- */
- void bindField(PreparedStatement statement, int index, Schema schema, Object value, ColumnDefinition colDef) throws SQLException ;
-
- /**
- * criteria for
- * @param incrementingColumn
- * @param timestampColumns
- * @return
- */
- TimestampIncrementingCriteria criteriaFor(
- ColumnId incrementingColumn,
- List<ColumnId> timestampColumns
- );
-
- /**
- * get min timestamp value
- * @param con
- * @param tableOrQuery
- * @param timestampColumns
- * @return
- * @throws SQLException
- */
- Long getMinTimestampValue(Connection con, String tableOrQuery, List<String> timestampColumns) throws SQLException;
-
- /**
- * A function to bind the values from a sink record into a prepared statement.
- */
- @FunctionalInterface
- interface StatementBinder {
/**
- * bind record
- * @param record
+ * dialect name
+ *
+ * @return
+ */
+ String name();
+
+ /**
+ * create jdbc prepared statement
+ *
+ * @param connection
+ * @param query
+ * @return
+ * @throws SQLException
+ */
+ PreparedStatement createPreparedStatement(Connection connection, String query) throws SQLException;
+
+ /**
+ * parse to Table Id
+ *
+ * @param fqn
+ * @return
+ */
+ TableId parseToTableId(String fqn);
+
+ /**
+ * Get the identifier rules for this database.
+ *
+ * @return the identifier rules
+ */
+ IdentifierRules identifierRules();
+
+ /**
+ * Get a new expression builder that can be used to build expressions with quoted identifiers.
+ *
+ * @return
+ */
+ ExpressionBuilder expressionBuilder();
+
+ Timestamp currentTimeOnDB(Connection connection, Calendar cal) throws SQLException;
+
+ /**
+ * Get a list of identifiers of the non-system tables in the database.
+ *
+ * @param connection
+ * @return
+ * @throws SQLException
+ */
+ List<TableId> tableIds(Connection connection) throws SQLException;
+
+ /**
+ * table exists
+ *
+ * @param connection
+ * @param tableId
+ * @return
+ * @throws SQLException
+ */
+ boolean tableExists(Connection connection, TableId tableId) throws SQLException;
+
+
+ /**
+ * Create the definition for the columns described by the database metadata.
+ */
+ Map<ColumnId, ColumnDefinition> describeColumns(Connection connection, String tablePattern, String columnPattern) throws SQLException;
+
+ /**
+ * Create the definition for the columns described by the database metadata.
+ */
+ Map<ColumnId, ColumnDefinition> describeColumns(Connection connection, String catalogPattern, String schemaPattern, String tablePattern, String columnPattern) throws SQLException;
+
+ /**
+ * Create the definition for the columns in the result set.
+ */
+ Map<ColumnId, ColumnDefinition> describeColumns(Connection conn, TableId tableId, ResultSetMetaData rsMetadata) throws SQLException;
+
+ /**
+ * describe table info
+ *
+ * @param connection
+ * @param tableId
+ * @return
* @throws SQLException
*/
- void bindRecord(ConnectRecord record) throws SQLException;
- }
-
-
- /**
- * Create a function that converts column values for the column defined by the specified mapping.
- * @param mapping
- * @return
- */
- ColumnConverter createColumnConverter(ColumnMapping mapping);
-
- /**
- * A function that obtains a column value from the current row of the specified result set.
- */
- @FunctionalInterface
- interface ColumnConverter {
+ TableDefinition describeTable(Connection connection, TableId tableId) throws SQLException;
+
/**
- * convert
- * @param resultSet
+ * describe columns by query sql
+ *
+ * @param connection
+ * @param tableId
* @return
* @throws SQLException
- * @throws IOException
*/
- Object convert(ResultSet resultSet) throws SQLException, IOException;
- }
+ Map<ColumnId, ColumnDefinition> describeColumnsByQuerying(Connection connection, TableId tableId) throws SQLException;
+
+ /**
+ * add field to schema
+ *
+ * @param column
+ * @param schema
+ * @param index
+ * @return
+ */
+ String addFieldToSchema(ColumnDefinition column, Schema schema, int index);
+
+ /**
+ * Apply the supplied DDL statements using the given connection. This gives the dialect the
+ * opportunity to execute the statements with a different autocommit setting.
+ */
+ void applyDdlStatements(Connection connection, List<String> statements) throws SQLException;
+
+ /**
+ * build dml statement
+ *
+ * @param table
+ * @param keyColumns
+ * @param nonKeyColumns
+ * @return
+ */
+ String buildInsertStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
+
+ /**
+ * update statement
+ *
+ * @param table
+ * @param keyColumns
+ * @param nonKeyColumns
+ * @return
+ */
+ String buildUpdateStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
+
+ /**
+ * upsert statment
+ *
+ * @param table
+ * @param keyColumns
+ * @param nonKeyColumns
+ * @return
+ */
+ String buildUpsertQueryStatement(TableId table, Collection<ColumnId> keyColumns, Collection<ColumnId> nonKeyColumns);
+
+ /**
+ * delete statement
+ *
+ * @param table
+ * @param keyColumns
+ * @return
+ */
+ default String buildDeleteStatement(TableId table, Collection<ColumnId> keyColumns) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ /**
+ * build select table
+ */
+ String buildSelectTableMode();
+
+ void buildSelectTable(ExpressionBuilder builder, TableId tableId);
+
+
+ /**
+ * drop table
+ *
+ * @param table
+ * @param options
+ * @return
+ */
+ String buildDropTableStatement(TableId table, DropOptions options);
+
+ /**
+ * create table
+ *
+ * @param table
+ * @param fields
+ * @return
+ */
+ String buildCreateTableStatement(TableId table, Collection<SinkRecordField> fields);
+
+ /**
+ * build alter table
+ *
+ * @param table
+ * @param fields
+ * @return
+ */
+ List<String> buildAlterTable(TableId table, Collection<SinkRecordField> fields);
+
+ /**
+ * Create a component that can bind record values into the supplied prepared statement.
+ *
+ * @param statement the prepared statement
+ * @param pkMode the primary key mode; may not be null
+ * @param schemaPair the key and value schemas; may not be null
+ * @param fieldsMetadata the field metadata; may not be null
+ * @param tableDefinition the table definition; may be null
+ * @param insertMode the insert mode; may not be null
+ * @return the statement binder; may not be null
+ */
+ StatementBinder statementBinder(
+ PreparedStatement statement,
+ JdbcSinkConfig.PrimaryKeyMode pkMode,
+ SchemaPair schemaPair,
+ FieldsMetadata fieldsMetadata,
+ TableDefinition tableDefinition,
+ JdbcSinkConfig.InsertMode insertMode
+ );
+
+ /**
+ * value column types
+ *
+ * @param rsMetadata
+ * @param columns
+ * @throws io.openmessaging.connector.api.errors.ConnectException
+ */
+ void validateColumnTypes(
+ ResultSetMetaData rsMetadata,
+ List<ColumnId> columns
+ ) throws io.openmessaging.connector.api.errors.ConnectException;
+
+
+ /**
+ * bind field
+ *
+ * @param statement
+ * @param index
+ * @param schema
+ * @param value
+ * @param colDef
+ * @throws SQLException
+ */
+ void bindField(PreparedStatement statement, int index, Schema schema, Object value, ColumnDefinition colDef) throws SQLException;
+
+ /**
+ * criteria for
+ *
+ * @param incrementingColumn
+ * @param timestampColumns
+ * @return
+ */
+ TimestampIncrementingCriteria criteriaFor(
+ ColumnId incrementingColumn,
+ List<ColumnId> timestampColumns
+ );
+
+ /**
+ * get min timestamp value
+ *
+ * @param con
+ * @param tableOrQuery
+ * @param timestampColumns
+ * @return
+ * @throws SQLException
+ */
+ Long getMinTimestampValue(Connection con, String tableOrQuery, List<String> timestampColumns) throws SQLException;
+
+ /**
+ * A function to bind the values from a sink record into a prepared statement.
+ */
+ @FunctionalInterface
+ interface StatementBinder {
+ /**
+ * bind record
+ *
+ * @param record
+ * @throws SQLException
+ */
+ void bindRecord(ConnectRecord record) throws SQLException;
+ }
+
+
+ /**
+ * Create a function that converts column values for the column defined by the specified mapping.
+ *
+ * @param mapping
+ * @return
+ */
+ ColumnConverter createColumnConverter(ColumnMapping mapping);
+
+ /**
+ * A function that obtains a column value from the current row of the specified result set.
+ */
+ @FunctionalInterface
+ interface ColumnConverter {
+ /**
+ * convert
+ *
+ * @param resultSet
+ * @return
+ * @throws SQLException
+ * @throws IOException
+ */
+ Object convert(ResultSet resultSet) throws SQLException, IOException;
+ }
}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialectFactory.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialectFactory.java
index 61d37b5..0938684 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialectFactory.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DatabaseDialectFactory.java
@@ -40,153 +40,154 @@ import java.util.regex.Pattern;
*/
public class DatabaseDialectFactory {
- /**
- * regex jdbc protocol
- */
- private static final Pattern PROTOCOL_PATTERN = Pattern.compile("jdbc:([^:]+):(.*)");
- private static final Logger LOG = LoggerFactory.getLogger(DatabaseDialectFactory.class);
- private static final ConcurrentMap<String, DatabaseDialectProvider> REGISTRY = new ConcurrentSkipListMap<>();
-
- static {
- loadAllDialects();
- }
-
- /**
- * load all database dialect
- */
- private static void loadAllDialects() {
- LOG.debug("Searching for and loading all JDBC source dialects on the classpath");
- final AtomicInteger count = new AtomicInteger();
- AccessController.doPrivileged(new PrivilegedAction<Void>() {
- @Override
- public Void run() {
- ServiceLoader<DatabaseDialectProvider> loadedDialects =
- ServiceLoader.load(
- DatabaseDialectProvider.class,
- this.getClass().getClassLoader()
- );
- Iterator<DatabaseDialectProvider> dialectIterator = loadedDialects.iterator();
- try {
- while (dialectIterator.hasNext()) {
- try {
- DatabaseDialectProvider provider = dialectIterator.next();
- REGISTRY.put(provider.getClass().getName(), provider);
- count.incrementAndGet();
- LOG.debug("Found '{}' provider {}", provider, provider.getClass());
- } catch (Throwable t) {
- LOG.debug("Skipping dialect provider after error while loading", t);
+ /**
+ * regex jdbc protocol
+ */
+ private static final Pattern PROTOCOL_PATTERN = Pattern.compile("jdbc:([^:]+):(.*)");
+ private static final Logger LOG = LoggerFactory.getLogger(DatabaseDialectFactory.class);
+ private static final ConcurrentMap<String, DatabaseDialectProvider> REGISTRY = new ConcurrentSkipListMap<>();
+
+ static {
+ loadAllDialects();
+ }
+
+ /**
+ * load all database dialect
+ */
+ private static void loadAllDialects() {
+ LOG.debug("Searching for and loading all JDBC source dialects on the classpath");
+ final AtomicInteger count = new AtomicInteger();
+ AccessController.doPrivileged(new PrivilegedAction<Void>() {
+ @Override
+ public Void run() {
+ ServiceLoader<DatabaseDialectProvider> loadedDialects =
+ ServiceLoader.load(
+ DatabaseDialectProvider.class,
+ this.getClass().getClassLoader()
+ );
+ Iterator<DatabaseDialectProvider> dialectIterator = loadedDialects.iterator();
+ try {
+ while (dialectIterator.hasNext()) {
+ try {
+ DatabaseDialectProvider provider = dialectIterator.next();
+ REGISTRY.put(provider.getClass().getName(), provider);
+ count.incrementAndGet();
+ LOG.debug("Found '{}' provider {}", provider, provider.getClass());
+ } catch (Throwable t) {
+ LOG.debug("Skipping dialect provider after error while loading", t);
+ }
+ }
+ } catch (Throwable t) {
+ LOG.debug("Error loading dialect providers", t);
+ }
+ return null;
}
- }
- } catch (Throwable t) {
- LOG.debug("Error loading dialect providers", t);
- }
- return null;
- }
- });
- LOG.debug("Registered {} source dialects", count.get());
- }
-
-
- /**
- * find dialect for
- * @param jdbcUrl
- * @param config
- * @return
- * @throws ConnectException
- */
- public static DatabaseDialect findDialectFor(
- String jdbcUrl,
- AbstractConfig config
- ) throws ConnectException {
- final JdbcUrlInfo info = extractJdbcUrlInfo(jdbcUrl);
- LOG.debug("Finding best dialect for {}", info);
- DatabaseDialectProvider bestMatch = null;
- for (DatabaseDialectProvider provider : REGISTRY.values()) {
- if (provider.protocolName().equals(info.subProtocol())){
- bestMatch=provider;
- break;
- }
+ });
+ LOG.debug("Registered {} source dialects", count.get());
}
- LOG.debug("Using dialect {} against {}", bestMatch, info);
- return bestMatch.create(config);
- }
-
- /**
- * create database dialect
- * @param dialectName
- * @param config
- * @return
- * @throws ConnectException
- */
- public static DatabaseDialect create(
- String dialectName,
- AbstractConfig config
- ) throws ConnectException {
- LOG.debug("Looking for named dialect '{}'", dialectName);
- Set<String> dialectNames = new HashSet<>();
- for (DatabaseDialectProvider provider : REGISTRY.values()) {
- dialectNames.add(provider.dialectName());
- if (provider.dialectName().equals(dialectName)) {
- return provider.create(config);
- }
+
+
+ /**
+ * find dialect for
+ *
+ * @param jdbcUrl
+ * @param config
+ * @return
+ * @throws ConnectException
+ */
+ public static DatabaseDialect findDialectFor(
+ String jdbcUrl,
+ AbstractConfig config
+ ) throws ConnectException {
+ final JdbcUrlInfo info = extractJdbcUrlInfo(jdbcUrl);
+ LOG.debug("Finding best dialect for {}", info);
+ DatabaseDialectProvider bestMatch = null;
+ for (DatabaseDialectProvider provider : REGISTRY.values()) {
+ if (provider.protocolName().equals(info.subProtocol())) {
+ bestMatch = provider;
+ break;
+ }
+ }
+ LOG.debug("Using dialect {} against {}", bestMatch, info);
+ return bestMatch.create(config);
}
- for (DatabaseDialectProvider provider : REGISTRY.values()) {
- if (provider.dialectName().equalsIgnoreCase(dialectName)) {
- return provider.create(config);
- }
+
+ /**
+ * create database dialect
+ *
+ * @param dialectName
+ * @param config
+ * @return
+ * @throws ConnectException
+ */
+ public static DatabaseDialect create(
+ String dialectName,
+ AbstractConfig config
+ ) throws ConnectException {
+ LOG.debug("Looking for named dialect '{}'", dialectName);
+ Set<String> dialectNames = new HashSet<>();
+ for (DatabaseDialectProvider provider : REGISTRY.values()) {
+ dialectNames.add(provider.dialectName());
+ if (provider.dialectName().equals(dialectName)) {
+ return provider.create(config);
+ }
+ }
+ for (DatabaseDialectProvider provider : REGISTRY.values()) {
+ if (provider.dialectName().equalsIgnoreCase(dialectName)) {
+ return provider.create(config);
+ }
+ }
+ throw new ConnectException(
+ "Unable to find dialect with name '" + dialectName + "' in the available dialects: "
+ + dialectNames
+ );
}
- throw new ConnectException(
- "Unable to find dialect with name '" + dialectName + "' in the available dialects: "
- + dialectNames
- );
- }
-
- public static JdbcUrlInfo extractJdbcUrlInfo(final String url) {
- Matcher matcher = PROTOCOL_PATTERN.matcher(url);
- if (matcher.matches()) {
- return new JdbcUrlDetails(matcher.group(1), matcher.group(2), url);
+
+ public static JdbcUrlInfo extractJdbcUrlInfo(final String url) {
+ Matcher matcher = PROTOCOL_PATTERN.matcher(url);
+ if (matcher.matches()) {
+ return new JdbcUrlDetails(matcher.group(1), matcher.group(2), url);
+ }
+ throw new ConnectException("Not a valid JDBC URL: " + url);
}
- throw new ConnectException("Not a valid JDBC URL: " + url);
- }
+ static class JdbcUrlDetails implements JdbcUrlInfo {
+ final String subprotocol;
+ final String subname;
+ final String url;
- static class JdbcUrlDetails implements JdbcUrlInfo {
- final String subprotocol;
- final String subname;
- final String url;
+ public JdbcUrlDetails(
+ String subprotocol,
+ String subname,
+ String url
+ ) {
+ this.subprotocol = subprotocol;
+ this.subname = subname;
+ this.url = url;
+ }
- public JdbcUrlDetails(
- String subprotocol,
- String subname,
- String url
- ) {
- this.subprotocol = subprotocol;
- this.subname = subname;
- this.url = url;
- }
+ @Override
+ public String subProtocol() {
+ return subprotocol;
+ }
- @Override
- public String subProtocol() {
- return subprotocol;
- }
+ @Override
+ public String subName() {
+ return subname;
+ }
- @Override
- public String subName() {
- return subname;
- }
+ @Override
+ public String url() {
+ return url;
+ }
- @Override
- public String url() {
- return url;
+ @Override
+ public String toString() {
+ return "JDBC subprotocol '" + subprotocol + "' and source '" + url + "'";
+ }
}
- @Override
- public String toString() {
- return "JDBC subprotocol '" + subprotocol + "' and source '" + url + "'";
+ private DatabaseDialectFactory() {
}
- }
-
- private DatabaseDialectFactory() {
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DropOptions.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DropOptions.java
index 4e8122f..11d6674 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DropOptions.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/DropOptions.java
@@ -20,81 +20,81 @@ import java.util.Objects;
public class DropOptions {
- private final boolean ifExists;
- private final boolean cascade;
+ private final boolean ifExists;
+ private final boolean cascade;
- /**
- * Create a new instance with the default settings.
- */
- public DropOptions() {
- this(false, false);
- }
-
- protected DropOptions(
- boolean ifExists,
- boolean cascade
- ) {
- this.ifExists = ifExists;
- this.cascade = cascade;
- }
+ /**
+ * Create a new instance with the default settings.
+ */
+ public DropOptions() {
+ this(false, false);
+ }
- /**
- * Get whether the 'IF EXISTS' clause should be used with the 'DROP' statement.
- *
- * @return true if the object should be dropped only if it already exists, or false otherwise
- */
- public boolean ifExists() {
- return ifExists;
- }
+ protected DropOptions(
+ boolean ifExists,
+ boolean cascade
+ ) {
+ this.ifExists = ifExists;
+ this.cascade = cascade;
+ }
- /**
- * Get whether the 'DROP' statement should cascade to dependent objects.
- *
- * @return true if dependent objects should also be dropped, or false otherwise
- */
- public boolean cascade() {
- return cascade;
- }
+ /**
+ * Get whether the 'IF EXISTS' clause should be used with the 'DROP' statement.
+ *
+ * @return true if the object should be dropped only if it already exists, or false otherwise
+ */
+ public boolean ifExists() {
+ return ifExists;
+ }
- /**
- * Set whether the 'IF EXISTS' clause should be used with the 'DROP' statement.
- *
- * @param ifExists true if the object should be dropped only if it already exists
- * @return a new options object with the current state plus the new if-exists state; never null
- */
- public DropOptions setIfExists(boolean ifExists) {
- return new DropOptions(ifExists, cascade);
- }
+ /**
+ * Get whether the 'DROP' statement should cascade to dependent objects.
+ *
+ * @return true if dependent objects should also be dropped, or false otherwise
+ */
+ public boolean cascade() {
+ return cascade;
+ }
- /**
- * Set whether the 'DROP' statement should cascade to dependent objects.
- *
- * @param cascade true if dependent objects should also be dropped, or false otherwise
- * @return a new options object with the current state plus the new cascade state; never null
- */
- public DropOptions setCascade(boolean cascade) {
- return new DropOptions(ifExists, cascade);
- }
+ /**
+ * Set whether the 'IF EXISTS' clause should be used with the 'DROP' statement.
+ *
+ * @param ifExists true if the object should be dropped only if it already exists
+ * @return a new options object with the current state plus the new if-exists state; never null
+ */
+ public DropOptions setIfExists(boolean ifExists) {
+ return new DropOptions(ifExists, cascade);
+ }
- @Override
- public String toString() {
- return "DropOptions{ifExists=" + ifExists + ", cascade=" + cascade + "}";
- }
+ /**
+ * Set whether the 'DROP' statement should cascade to dependent objects.
+ *
+ * @param cascade true if dependent objects should also be dropped, or false otherwise
+ * @return a new options object with the current state plus the new cascade state; never null
+ */
+ public DropOptions setCascade(boolean cascade) {
+ return new DropOptions(ifExists, cascade);
+ }
- @Override
- public int hashCode() {
- return Objects.hash(ifExists, cascade);
- }
+ @Override
+ public String toString() {
+ return "DropOptions{ifExists=" + ifExists + ", cascade=" + cascade + "}";
+ }
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
+ @Override
+ public int hashCode() {
+ return Objects.hash(ifExists, cascade);
}
- if (obj instanceof DropOptions) {
- DropOptions that = (DropOptions) obj;
- return this.ifExists() == that.ifExists() && this.cascade() == that.cascade();
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (obj instanceof DropOptions) {
+ DropOptions that = (DropOptions) obj;
+ return this.ifExists() == that.ifExists() && this.cascade() == that.cascade();
+ }
+ return false;
}
- return false;
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/PreparedStatementBinder.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/PreparedStatementBinder.java
index d74ec52..dc9c0ae 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/PreparedStatementBinder.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/PreparedStatementBinder.java
@@ -37,94 +37,94 @@ import java.util.Objects;
*/
public class PreparedStatementBinder implements DatabaseDialect.StatementBinder {
- private final JdbcSinkConfig.PrimaryKeyMode pkMode;
- private final PreparedStatement statement;
- private final SchemaPair schemaPair;
- private final FieldsMetadata fieldsMetadata;
- private final JdbcSinkConfig.InsertMode insertMode;
- private final DatabaseDialect dialect;
- private final TableDefinition tabDef;
+ private final JdbcSinkConfig.PrimaryKeyMode pkMode;
+ private final PreparedStatement statement;
+ private final SchemaPair schemaPair;
+ private final FieldsMetadata fieldsMetadata;
+ private final JdbcSinkConfig.InsertMode insertMode;
+ private final DatabaseDialect dialect;
+ private final TableDefinition tabDef;
- public PreparedStatementBinder(
- DatabaseDialect dialect,
- PreparedStatement statement,
- JdbcSinkConfig.PrimaryKeyMode pkMode,
- SchemaPair schemaPair,
- FieldsMetadata fieldsMetadata,
- TableDefinition tabDef,
- JdbcSinkConfig.InsertMode insertMode
- ) {
- this.dialect = dialect;
- this.pkMode = pkMode;
- this.statement = statement;
- this.schemaPair = schemaPair;
- this.fieldsMetadata = fieldsMetadata;
- this.insertMode = insertMode;
- this.tabDef = tabDef;
- }
+ public PreparedStatementBinder(
+ DatabaseDialect dialect,
+ PreparedStatement statement,
+ JdbcSinkConfig.PrimaryKeyMode pkMode,
+ SchemaPair schemaPair,
+ FieldsMetadata fieldsMetadata,
+ TableDefinition tabDef,
+ JdbcSinkConfig.InsertMode insertMode
+ ) {
+ this.dialect = dialect;
+ this.pkMode = pkMode;
+ this.statement = statement;
+ this.schemaPair = schemaPair;
+ this.fieldsMetadata = fieldsMetadata;
+ this.insertMode = insertMode;
+ this.tabDef = tabDef;
+ }
- @Override
- public void bindRecord(ConnectRecord record) throws SQLException {
- final boolean isDelete = Objects.isNull(record.getData());
+ @Override
+ public void bindRecord(ConnectRecord record) throws SQLException {
+ final boolean isDelete = Objects.isNull(record.getData());
- int index = 1;
- if (isDelete) {
- bindKeyFields(record, index);
- } else {
- switch (insertMode) {
- case INSERT:
- case UPSERT:
- index = bindKeyFields(record, index);
- bindNonKeyFields(record,index);
- break;
+ int index = 1;
+ if (isDelete) {
+ bindKeyFields(record, index);
+ } else {
+ switch (insertMode) {
+ case INSERT:
+ case UPSERT:
+ index = bindKeyFields(record, index);
+ bindNonKeyFields(record, index);
+ break;
- case UPDATE:
- index = bindNonKeyFields(record, index);
- bindKeyFields(record, index);
- break;
- default:
- throw new AssertionError();
+ case UPDATE:
+ index = bindNonKeyFields(record, index);
+ bindKeyFields(record, index);
+ break;
+ default:
+ throw new AssertionError();
- }
+ }
+ }
+ statement.addBatch();
}
- statement.addBatch();
- }
- protected int bindKeyFields(ConnectRecord record, int index) throws SQLException {
- switch (pkMode) {
- case NONE:
- if (!fieldsMetadata.keyFieldNames.isEmpty()) {
- throw new AssertionError();
- }
- break;
- case RECORD_VALUE: {
- Object[] data=JSONArray.parseArray(JSON.toJSONString(record.getData())).stream().toArray();
- for (String fieldName : fieldsMetadata.keyFieldNames) {
- final Field field = schemaPair.schema.getField(fieldName);
- bindField(index++, field.getSchema(), data[field.getIndex()] , fieldName);
+ protected int bindKeyFields(ConnectRecord record, int index) throws SQLException {
+ switch (pkMode) {
+ case NONE:
+ if (!fieldsMetadata.keyFieldNames.isEmpty()) {
+ throw new AssertionError();
+ }
+ break;
+ case RECORD_VALUE: {
+ Object[] data = JSONArray.parseArray(JSON.toJSONString(record.getData())).stream().toArray();
+ for (String fieldName : fieldsMetadata.keyFieldNames) {
+ final Field field = schemaPair.schema.getField(fieldName);
+ bindField(index++, field.getSchema(), data[field.getIndex()], fieldName);
+ }
+ }
+ break;
+ default:
+ throw new ConnectException("Unknown primary key mode: " + pkMode);
}
- }
- break;
- default:
- throw new ConnectException("Unknown primary key mode: " + pkMode);
+ return index;
}
- return index;
- }
- protected int bindNonKeyFields(
- ConnectRecord record,
- int index
- ) throws SQLException {
- for (final String fieldName : fieldsMetadata.nonKeyFieldNames) {
- final Field field = record.getSchema().getField(fieldName);
- bindField(index++, field.getSchema(), ((Object[])record.getData())[field.getIndex()] , fieldName);
+ protected int bindNonKeyFields(
+ ConnectRecord record,
+ int index
+ ) throws SQLException {
+ for (final String fieldName : fieldsMetadata.nonKeyFieldNames) {
+ final Field field = record.getSchema().getField(fieldName);
+ bindField(index++, field.getSchema(), ((Object[]) record.getData())[field.getIndex()], fieldName);
+ }
+ return index;
}
- return index;
- }
- protected void bindField(int index, Schema schema, Object value, String fieldName)
- throws SQLException {
- ColumnDefinition colDef = tabDef == null ? null : tabDef.definitionForColumn(fieldName);
- dialect.bindField(statement, index, schema, value, colDef);
- }
+ protected void bindField(int index, Schema schema, Object value, String fieldName)
+ throws SQLException {
+ ColumnDefinition colDef = tabDef == null ? null : tabDef.definitionForColumn(fieldName);
+ dialect.bindField(statement, index, schema, value, colDef);
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
index f177148..d9e8dee 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
@@ -43,7 +43,13 @@ import org.apache.rocketmq.connect.jdbc.schema.column.parser.DateColumnParser;
import org.apache.rocketmq.connect.jdbc.schema.column.parser.TimeColumnParser;
import org.apache.rocketmq.connect.jdbc.schema.column.parser.TimestampColumnParser;
import org.apache.rocketmq.connect.jdbc.source.TimestampIncrementingCriteria;
-import org.apache.rocketmq.connect.jdbc.util.*;
+import org.apache.rocketmq.connect.jdbc.util.DateTimeUtils;
+import org.apache.rocketmq.connect.jdbc.util.ExpressionBuilder;
+import org.apache.rocketmq.connect.jdbc.util.IdentifierRules;
+import org.apache.rocketmq.connect.jdbc.util.JdbcDriverInfo;
+import org.apache.rocketmq.connect.jdbc.util.NumericMapping;
+import org.apache.rocketmq.connect.jdbc.util.QuoteMethod;
+import org.apache.rocketmq.connect.jdbc.util.TableType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,790 +94,808 @@ import java.util.stream.Collectors;
*/
public class GenericDatabaseDialect implements DatabaseDialect {
- protected static final int NUMERIC_TYPE_SCALE_LOW = -84;
- protected static final int NUMERIC_TYPE_SCALE_HIGH = 127;
- protected static final int NUMERIC_TYPE_SCALE_UNSET = -127;
+ protected static final int NUMERIC_TYPE_SCALE_LOW = -84;
+ protected static final int NUMERIC_TYPE_SCALE_HIGH = 127;
+ protected static final int NUMERIC_TYPE_SCALE_UNSET = -127;
- private static final int MAX_INTEGER_TYPE_PRECISION = 18;
+ private static final int MAX_INTEGER_TYPE_PRECISION = 18;
- /**
- * The provider for GenericDatabaseDialect
- */
- public static class Provider extends DatabaseDialectProvider {
- public Provider() {
- super(GenericDatabaseDialect.class.getSimpleName(),"");
+ /**
+ * The provider for GenericDatabaseDialect
+ */
+ public static class Provider extends DatabaseDialectProvider {
+ public Provider() {
+ super(GenericDatabaseDialect.class.getSimpleName(), "");
+ }
+
+ @Override
+ public DatabaseDialect create(AbstractConfig config) {
+ return new GenericDatabaseDialect(config);
+ }
+ }
+
+ /**
+ * The provider for GenericDatabaseDialect
+ */
+ public static class DialectName {
+ public static String generateDialectName(Class clazz) {
+ return clazz.getSimpleName().replace("DatabaseDialect", "");
+ }
}
+
+ private static final Logger log = LoggerFactory.getLogger(GenericDatabaseDialect.class);
+
+// @Deprecated
+// protected final Logger log = LoggerFactory.getLogger(GenericDatabaseDialect.class);
+ protected AbstractConfig config;
+ /**
+ * Whether to map {@code NUMERIC} JDBC types by precision.
+ */
+ protected NumericMapping mapNumerics = NumericMapping.NONE;
+ protected String catalogPattern;
+ protected String schemaPattern;
+ protected Set<String> tableTypes;
+ protected String jdbcUrl;
+ protected JdbcUrlInfo jdbcUrlInfo;
+ private QuoteMethod quoteSqlIdentifiers = QuoteMethod.ALWAYS;
+ private IdentifierRules defaultIdentifierRules;
+ private AtomicReference<IdentifierRules> identifierRules = new AtomicReference<>();
+ private Queue<Connection> connections = new ConcurrentLinkedQueue<>();
+ private volatile JdbcDriverInfo jdbcDriverInfo;
+ private int batchMaxRows;
+ private TimeZone timeZone;
+
+ public GenericDatabaseDialect(AbstractConfig config) {
+ this(config, IdentifierRules.DEFAULT);
+ }
+
+ protected GenericDatabaseDialect(
+ AbstractConfig config, IdentifierRules defaultIdentifierRules
+ ) {
+ this.config = config;
+ this.defaultIdentifierRules = defaultIdentifierRules;
+ this.jdbcUrl = config.getConnectionDbUrl();
+ this.jdbcUrlInfo = DatabaseDialectFactory.extractJdbcUrlInfo(jdbcUrl);
+ if (config instanceof JdbcSinkConfig) {
+ JdbcSinkConfig sinkConfig = (JdbcSinkConfig) config;
+ catalogPattern = JdbcSourceConfig.CATALOG_PATTERN_DEFAULT;
+ schemaPattern = JdbcSourceConfig.SCHEMA_PATTERN_DEFAULT;
+ tableTypes = sinkConfig.tableTypeNames();
+ quoteSqlIdentifiers = QuoteMethod.get(config.getQuoteSqlIdentifiers());
+ mapNumerics = NumericMapping.NONE;
+ batchMaxRows = 0;
+ timeZone = sinkConfig.getTimeZone();
+ } else {
+ JdbcSourceConfig sourceConfig = (JdbcSourceConfig) config;
+ catalogPattern = sourceConfig.getCatalogPattern();
+ schemaPattern = sourceConfig.getSchemaPattern();
+ tableTypes = sourceConfig.getTableTypes().stream().map(TableType::toString).collect(Collectors.toSet());
+ quoteSqlIdentifiers = QuoteMethod.get(config.getQuoteSqlIdentifiers());
+ mapNumerics = sourceConfig.numericMapping();
+ batchMaxRows = sourceConfig.getBatchMaxRows();
+ timeZone = sourceConfig.getTimeZone();
+ }
+
+ }
+
+ @Override
+ public String name() {
+ return DialectName.generateDialectName(getClass());
+// return getClass().getSimpleName().replace("DatabaseDialect", "");
+ }
+
+ /**
+ * init jdbc connection
+ *
+ * @return
+ * @throws SQLException
+ */
+ @SneakyThrows
@Override
- public DatabaseDialect create(AbstractConfig config) {
- return new GenericDatabaseDialect(config);
- }
- }
-
- private static final Logger glog = LoggerFactory.getLogger(GenericDatabaseDialect.class);
-
- @Deprecated
- protected final Logger log = LoggerFactory.getLogger(GenericDatabaseDialect.class);
- protected AbstractConfig config;
- /**
- * Whether to map {@code NUMERIC} JDBC types by precision.
- */
- protected NumericMapping mapNumerics=NumericMapping.NONE;
- protected String catalogPattern;
- protected String schemaPattern;
- protected Set<String> tableTypes;
- protected String jdbcUrl;
- protected JdbcUrlInfo jdbcUrlInfo;
- private QuoteMethod quoteSqlIdentifiers=QuoteMethod.ALWAYS;
- private IdentifierRules defaultIdentifierRules;
- private AtomicReference<IdentifierRules> identifierRules = new AtomicReference<>();
- private Queue<Connection> connections = new ConcurrentLinkedQueue<>();
- private volatile JdbcDriverInfo jdbcDriverInfo;
- private int batchMaxRows;
- private TimeZone timeZone;
-
- public GenericDatabaseDialect(AbstractConfig config) {
- this(config, IdentifierRules.DEFAULT);
- }
-
- protected GenericDatabaseDialect(
- AbstractConfig config, IdentifierRules defaultIdentifierRules
- ) {
- this.config = config;
- this.defaultIdentifierRules = defaultIdentifierRules;
- this.jdbcUrl = config.getConnectionDbUrl();
- this.jdbcUrlInfo = DatabaseDialectFactory.extractJdbcUrlInfo(jdbcUrl);
- if (config instanceof JdbcSinkConfig) {
- JdbcSinkConfig sinkConfig = (JdbcSinkConfig) config;
- catalogPattern = JdbcSourceConfig.CATALOG_PATTERN_DEFAULT;
- schemaPattern = JdbcSourceConfig.SCHEMA_PATTERN_DEFAULT;
- tableTypes = sinkConfig.tableTypeNames();
- quoteSqlIdentifiers = QuoteMethod.get(config.getQuoteSqlIdentifiers());
- mapNumerics = NumericMapping.NONE;
- batchMaxRows = 0;
- timeZone = sinkConfig.getTimeZone();
- } else {
- JdbcSourceConfig sourceConfig = (JdbcSourceConfig) config;
- catalogPattern = sourceConfig.getCatalogPattern();
- schemaPattern = sourceConfig.getSchemaPattern();
- tableTypes = sourceConfig.getTableTypes().stream().map(TableType::toString).collect(Collectors.toSet());
- quoteSqlIdentifiers = QuoteMethod.get(config.getQuoteSqlIdentifiers());
- mapNumerics = sourceConfig.numericMapping();
- batchMaxRows = sourceConfig.getBatchMaxRows();
- timeZone = sourceConfig.getTimeZone();
- }
-
- }
-
- @Override
- public String name() {
- return getClass().getSimpleName().replace("DatabaseDialect", "");
- }
-
- /**
- * init jdbc connection
- * @return
- * @throws SQLException
- */
- @SneakyThrows
- @Override
- public Connection getConnection() throws SQLException {
- // These config names are the same for both source and sink configs ...
- String username = config.getConnectionDbUser();
- String dbPassword = config.getConnectionDbPassword();
- Properties properties = new Properties();
- if (username != null) {
- properties.setProperty("user", username);
- }
- if (dbPassword != null) {
- properties.setProperty("password", dbPassword);
- }
- properties = addConnectionProperties(properties);
- DriverManager.setLoginTimeout(40);
- Connection connection = DriverManager.getConnection(jdbcUrl, properties);
- if (jdbcDriverInfo == null) {
- jdbcDriverInfo = createJdbcDriverInfo(connection);
- }
- connections.add(connection);
- return connection;
- }
-
- @Override
- public void close() {
- Connection conn;
- while ((conn = connections.poll()) != null) {
- try {
- conn.close();
- } catch (Throwable e) {
- glog.warn("Error while closing connection to {}", jdbcDriverInfo, e);
- }
- }
- }
-
- @Override
- public boolean isConnectionValid(
- Connection connection,
- int timeout
- ) throws SQLException {
- if (jdbcDriverInfo().jdbcMajorVersion() >= 4) {
- return connection.isValid(timeout);
- }
- // issue a test query ...
- String query = checkConnectionQuery();
- if (query != null) {
- try (Statement statement = connection.createStatement()) {
- if (statement.execute(query)) {
- ResultSet rs = null;
- try {
- // do nothing with the result set
- rs = statement.getResultSet();
- } finally {
- if (rs != null) {
- rs.close();
+ public Connection getConnection() throws SQLException {
+ // These config names are the same for both source and sink configs ...
+ String username = config.getConnectionDbUser();
+ String dbPassword = config.getConnectionDbPassword();
+ Properties properties = new Properties();
+ if (username != null) {
+ properties.setProperty("user", username);
+ }
+ if (dbPassword != null) {
+ properties.setProperty("password", dbPassword);
+ }
+ properties = addConnectionProperties(properties);
+ DriverManager.setLoginTimeout(40);
+ Connection connection = DriverManager.getConnection(jdbcUrl, properties);
+ if (jdbcDriverInfo == null) {
+ jdbcDriverInfo = createJdbcDriverInfo(connection);
+ }
+ connections.add(connection);
+ return connection;
+ }
+
+ @Override
+ public void close() {
+ Connection conn;
+ while ((conn = connections.poll()) != null) {
+ try {
+ conn.close();
+ } catch (Throwable e) {
+ log.warn("Error while closing connection to {}", jdbcDriverInfo, e);
+ }
+ }
+ }
+
+ @Override
+ public boolean isConnectionValid(
+ Connection connection,
+ int timeout
+ ) throws SQLException {
+ if (jdbcDriverInfo().jdbcMajorVersion() >= 4) {
+ return connection.isValid(timeout);
+ }
+ // issue a test query ...
+ String query = checkConnectionQuery();
+ if (query != null) {
+ try (Statement statement = connection.createStatement()) {
+ if (statement.execute(query)) {
+ ResultSet rs = null;
+ try {
+ // do nothing with the result set
+ rs = statement.getResultSet();
+ } finally {
+ if (rs != null) {
+ rs.close();
+ }
+ }
+ }
+ }
+ }
+ return true;
+ }
+
+ /**
+ * check connection query
+ *
+ * @return the check connection query; may be null if the connection should not be queried
+ */
+ protected String checkConnectionQuery() {
+ return "SELECT 1";
+ }
+
+ protected JdbcDriverInfo jdbcDriverInfo() {
+ if (jdbcDriverInfo == null) {
+ try (Connection connection = getConnection()) {
+ jdbcDriverInfo = createJdbcDriverInfo(connection);
+ } catch (SQLException e) {
+ throw new io.openmessaging.connector.api.errors.ConnectException("Unable to get JDBC driver information", e);
}
- }
}
- }
- }
- return true;
- }
-
- /**
- * check connection query
- * @return the check connection query; may be null if the connection should not be queried
- */
- protected String checkConnectionQuery() {
- return "SELECT 1";
- }
-
- protected JdbcDriverInfo jdbcDriverInfo() {
- if (jdbcDriverInfo == null) {
- try (Connection connection = getConnection()) {
- jdbcDriverInfo = createJdbcDriverInfo(connection);
- } catch (SQLException e) {
- throw new io.openmessaging.connector.api.errors.ConnectException("Unable to get JDBC driver information", e);
- }
- }
- return jdbcDriverInfo;
- }
-
- protected JdbcDriverInfo createJdbcDriverInfo(Connection connection) throws SQLException {
- DatabaseMetaData metadata = connection.getMetaData();
- return new JdbcDriverInfo(
- metadata.getJDBCMajorVersion(),
- metadata.getJDBCMinorVersion(),
- metadata.getDriverName(),
- metadata.getDatabaseProductName(),
- metadata.getDatabaseProductVersion()
- );
- }
-
- /**
- * add connect properties
- * @param properties
- * @return
- */
- protected Properties addConnectionProperties(Properties properties) {
- // todo 设置 config中的属性
- return properties;
- }
-
- @Override
- public PreparedStatement createPreparedStatement(
- Connection db,
- String query
- ) throws SQLException {
- glog.trace("Creating a PreparedStatement '{}'", query);
- PreparedStatement stmt = db.prepareStatement(query);
- initializePreparedStatement(stmt);
- return stmt;
- }
-
- /**
- * init PreparedStatement
- * @param stmt
- * @throws SQLException
- */
- protected void initializePreparedStatement(PreparedStatement stmt) throws SQLException {
- if (batchMaxRows > 0) {
- stmt.setFetchSize(batchMaxRows);
- }
- }
-
- @Override
- public TableId parseToTableId(String fqn) {
- List<String> parts = identifierRules().parseQualifiedIdentifier(fqn);
- if (parts.isEmpty()) {
- throw new IllegalArgumentException("Invalid fully qualified name: '" + fqn + "'");
- }
- if (parts.size() == 1) {
- return new TableId(null, null, parts.get(0));
- }
- if (parts.size() == 3) {
- return new TableId(parts.get(0), parts.get(1), parts.get(2));
- }
- assert parts.size() >= 2;
- if (useCatalog()) {
- return new TableId(parts.get(0), null, parts.get(1));
- }
- return new TableId(null, parts.get(0), parts.get(1));
- }
-
- /**
- * Return whether the database uses JDBC catalogs.
- * @return true if catalogs are used, or false otherwise
- */
- protected boolean useCatalog() {
- return false;
- }
-
- @Override
- public List<TableId> tableIds(Connection conn) throws SQLException {
- DatabaseMetaData metadata = conn.getMetaData();
- String[] tableTypes = tableTypes(metadata, this.tableTypes);
- String tableTypeDisplay = displayableTableTypes(tableTypes, ", ");
- glog.debug("Using {} dialect to get {}", this, tableTypeDisplay);
- try (ResultSet rs = metadata.getTables(catalogPattern(), schemaPattern(), "%", tableTypes)) {
- List<TableId> tableIds = new ArrayList<>();
- while (rs.next()) {
- String catalogName = rs.getString(1);
- String schemaName = rs.getString(2);
- String tableName = rs.getString(3);
- TableId tableId = new TableId(catalogName, schemaName, tableName);
- if (includeTable(tableId)) {
- tableIds.add(tableId);
+ return jdbcDriverInfo;
+ }
+
+ protected JdbcDriverInfo createJdbcDriverInfo(Connection connection) throws SQLException {
+ DatabaseMetaData metadata = connection.getMetaData();
+ return new JdbcDriverInfo(
+ metadata.getJDBCMajorVersion(),
+ metadata.getJDBCMinorVersion(),
+ metadata.getDriverName(),
+ metadata.getDatabaseProductName(),
+ metadata.getDatabaseProductVersion()
+ );
+ }
+
+ /**
+ * add connect properties
+ *
+ * @param properties
+ * @return
+ */
+ protected Properties addConnectionProperties(Properties properties) {
+ // do nothiong
+ return properties;
+ }
+
+ @Override
+ public PreparedStatement createPreparedStatement(
+ Connection db,
+ String query
+ ) throws SQLException {
+ log.trace("Creating a PreparedStatement '{}'", query);
+ PreparedStatement stmt = db.prepareStatement(query);
+ initializePreparedStatement(stmt);
+ return stmt;
+ }
+
+ /**
+ * init PreparedStatement
+ *
+ * @param stmt
+ * @throws SQLException
+ */
+ protected void initializePreparedStatement(PreparedStatement stmt) throws SQLException {
+ if (batchMaxRows > 0) {
+ stmt.setFetchSize(batchMaxRows);
+ }
+ }
+
+ @Override
+ public TableId parseToTableId(String fqn) {
+ List<String> parts = identifierRules().parseQualifiedIdentifier(fqn);
+ if (parts.isEmpty()) {
+ throw new IllegalArgumentException("Invalid fully qualified name: '" + fqn + "'");
}
- }
- glog.debug("Used {} dialect to find {} {}", this, tableIds.size(), tableTypeDisplay);
- return tableIds;
- }
- }
-
- protected String catalogPattern() {
- return catalogPattern;
- }
-
- protected String schemaPattern() {
- return schemaPattern;
- }
-
- /**
- * Determine whether the table with the specific name is to be included in the tables.
- *
- * <p>This method can be overridden to exclude certain database tables.
- *
- * @param table the identifier of the table; may be null
- * @return true if the table should be included; false otherwise
- */
- protected boolean includeTable(TableId table) {
- return true;
- }
-
- /**
- * Find the available table types that are returned by the JDBC driver that case insensitively
- * match the specified types.
- *
- * @param metadata the database metadata; may not be null but may be empty if no table types
- * @param types the case-independent table types that are desired
- * @return the array of table types take directly from the list of available types returned by the
- * JDBC driver; never null
- * @throws SQLException if there is an error with the database connection
- */
- protected String[] tableTypes(
- DatabaseMetaData metadata,
- Set<String> types
- ) throws SQLException {
- glog.debug("Using {} dialect to check support for {}", this, types);
- // Compute the uppercase form of the desired types ...
- Set<String> uppercaseTypes = new HashSet<>();
- for (String type : types) {
- if (type != null) {
- uppercaseTypes.add(type.toUpperCase(Locale.ROOT));
- }
- }
- // Now find out the available table types ...
- Set<String> matchingTableTypes = new HashSet<>();
- try (ResultSet rs = metadata.getTableTypes()) {
- while (rs.next()) {
- String tableType = rs.getString(1);
- if (tableType != null && uppercaseTypes.contains(tableType.toUpperCase(Locale.ROOT))) {
- matchingTableTypes.add(tableType);
+ if (parts.size() == 1) {
+ return new TableId(null, null, parts.get(0));
}
- }
- }
- String[] result = matchingTableTypes.toArray(new String[matchingTableTypes.size()]);
- glog.debug("Used {} dialect to find table types: {}", this, result);
- return result;
- }
-
- @Override
- public IdentifierRules identifierRules() {
- if (identifierRules.get() == null) {
- // Otherwise try to get the actual quote string and separator from the database, since
- // many databases allow them to be changed
- try (Connection connection = getConnection()) {
- DatabaseMetaData metaData = connection.getMetaData();
- String leadingQuoteStr = metaData.getIdentifierQuoteString();
- String trailingQuoteStr = leadingQuoteStr; // JDBC does not distinguish
- String separator = metaData.getCatalogSeparator();
- if (leadingQuoteStr == null || leadingQuoteStr.isEmpty()) {
- leadingQuoteStr = defaultIdentifierRules.leadingQuoteString();
- trailingQuoteStr = defaultIdentifierRules.trailingQuoteString();
+ if (parts.size() == 3) {
+ return new TableId(parts.get(0), parts.get(1), parts.get(2));
}
- if (separator == null || separator.isEmpty()) {
- separator = defaultIdentifierRules.identifierDelimiter();
+ assert parts.size() >= 2;
+ if (useCatalog()) {
+ return new TableId(parts.get(0), null, parts.get(1));
}
- identifierRules.set(new IdentifierRules(separator, leadingQuoteStr, trailingQuoteStr));
- } catch (SQLException e) {
- if (defaultIdentifierRules != null) {
- identifierRules.set(defaultIdentifierRules);
- glog.warn("Unable to get identifier metadata; using default rules", e);
- } else {
- throw new io.openmessaging.connector.api.errors.ConnectException("Unable to get identifier metadata", e);
+ return new TableId(null, parts.get(0), parts.get(1));
+ }
+
+ /**
+ * Return whether the database uses JDBC catalogs.
+ *
+ * @return true if catalogs are used, or false otherwise
+ */
+ protected boolean useCatalog() {
+ return false;
+ }
+
+ @Override
+ public List<TableId> tableIds(Connection conn) throws SQLException {
+ DatabaseMetaData metadata = conn.getMetaData();
+ String[] tableTypes = tableTypes(metadata, this.tableTypes);
+ String tableTypeDisplay = displayableTableTypes(tableTypes, ", ");
+ log.debug("Using {} dialect to get {}", this, tableTypeDisplay);
+ try (ResultSet rs = metadata.getTables(catalogPattern(), schemaPattern(), "%", tableTypes)) {
+ List<TableId> tableIds = new ArrayList<>();
+ while (rs.next()) {
+ String catalogName = rs.getString(1);
+ String schemaName = rs.getString(2);
+ String tableName = rs.getString(3);
+ TableId tableId = new TableId(catalogName, schemaName, tableName);
+ if (includeTable(tableId)) {
+ tableIds.add(tableId);
+ }
+ }
+ log.debug("Used {} dialect to find {} {}", this, tableIds.size(), tableTypeDisplay);
+ return tableIds;
+ }
+ }
+
+ protected String catalogPattern() {
+ return catalogPattern;
+ }
+
+ protected String schemaPattern() {
+ return schemaPattern;
+ }
+
+ /**
+ * Determine whether the table with the specific name is to be included in the tables.
+ *
+ * <p>This method can be overridden to exclude certain database tables.
+ *
+ * @param table the identifier of the table; may be null
+ * @return true if the table should be included; false otherwise
+ */
+ protected boolean includeTable(TableId table) {
+ return true;
+ }
+
+ /**
+ * Find the available table types that are returned by the JDBC driver that case insensitively
+ * match the specified types.
+ *
+ * @param metadata the database metadata; may not be null but may be empty if no table types
+ * @param types the case-independent table types that are desired
+ * @return the array of table types take directly from the list of available types returned by the
+ * JDBC driver; never null
+ * @throws SQLException if there is an error with the database connection
+ */
+ protected String[] tableTypes(
+ DatabaseMetaData metadata,
+ Set<String> types
+ ) throws SQLException {
+ log.debug("Using {} dialect to check support for {}", this, types);
+ // Compute the uppercase form of the desired types ...
+ Set<String> uppercaseTypes = new HashSet<>();
+ for (String type : types) {
+ if (type != null) {
+ uppercaseTypes.add(type.toUpperCase(Locale.ROOT));
+ }
+ }
+ // Now find out the available table types ...
+ Set<String> matchingTableTypes = new HashSet<>();
+ try (ResultSet rs = metadata.getTableTypes()) {
+ while (rs.next()) {
+ String tableType = rs.getString(1);
+ if (tableType != null && uppercaseTypes.contains(tableType.toUpperCase(Locale.ROOT))) {
+ matchingTableTypes.add(tableType);
+ }
+ }
+ }
+ String[] result = matchingTableTypes.toArray(new String[matchingTableTypes.size()]);
+ log.debug("Used {} dialect to find table types: {}", this, result);
+ return result;
+ }
+
+ @Override
+ public IdentifierRules identifierRules() {
+ if (identifierRules.get() == null) {
+ // Otherwise try to get the actual quote string and separator from the database, since
+ // many databases allow them to be changed
+ try (Connection connection = getConnection()) {
+ DatabaseMetaData metaData = connection.getMetaData();
+ String leadingQuoteStr = metaData.getIdentifierQuoteString();
+ String trailingQuoteStr = leadingQuoteStr; // JDBC does not distinguish
+ String separator = metaData.getCatalogSeparator();
+ if (leadingQuoteStr == null || leadingQuoteStr.isEmpty()) {
+ leadingQuoteStr = defaultIdentifierRules.leadingQuoteString();
+ trailingQuoteStr = defaultIdentifierRules.trailingQuoteString();
+ }
+ if (separator == null || separator.isEmpty()) {
+ separator = defaultIdentifierRules.identifierDelimiter();
+ }
+ identifierRules.set(new IdentifierRules(separator, leadingQuoteStr, trailingQuoteStr));
+ } catch (SQLException e) {
+ if (defaultIdentifierRules != null) {
+ identifierRules.set(defaultIdentifierRules);
+ log.warn("Unable to get identifier metadata; using default rules", e);
+ } else {
+ throw new io.openmessaging.connector.api.errors.ConnectException("Unable to get identifier metadata", e);
+ }
+ }
+ }
+ return identifierRules.get();
+ }
+
+ @Override
+ public ExpressionBuilder expressionBuilder() {
+ return identifierRules().expressionBuilder()
+ .setQuoteIdentifiers(quoteSqlIdentifiers);
+ }
+
+ /**
+ * Return current time at the database
+ *
+ * @param conn database connection
+ * @param cal calendar
+ * @return the current time at the database
+ */
+ @Override
+ public Timestamp currentTimeOnDB(
+ Connection conn,
+ Calendar cal
+ ) throws SQLException {
+ String query = currentTimestampDatabaseQuery();
+ assert query != null;
+ assert !query.isEmpty();
+ try (Statement stmt = conn.createStatement()) {
+ log.debug("executing query " + query + " to get current time from database");
+ try (ResultSet rs = stmt.executeQuery(query)) {
+ if (rs.next()) {
+ return rs.getTimestamp(1, cal);
+ } else {
+ throw new io.openmessaging.connector.api.errors.ConnectException(
+ "Unable to get current time from DB using " + this + " and query '" + query + "'"
+ );
+ }
+ }
+ } catch (SQLException e) {
+ log.error("Failed to get current time from DB using {} and query '{}'", this, query, e);
+ throw e;
}
- }
- }
- return identifierRules.get();
- }
-
- @Override
- public ExpressionBuilder expressionBuilder() {
- return identifierRules().expressionBuilder()
- .setQuoteIdentifiers(quoteSqlIdentifiers);
- }
-
- /**
- * Return current time at the database
- * @param conn database connection
- * @param cal calendar
- * @return the current time at the database
- */
- @Override
- public Timestamp currentTimeOnDB(
- Connection conn,
- Calendar cal
- ) throws SQLException {
- String query = currentTimestampDatabaseQuery();
- assert query != null;
- assert !query.isEmpty();
- try (Statement stmt = conn.createStatement()) {
- glog.debug("executing query " + query + " to get current time from database");
- try (ResultSet rs = stmt.executeQuery(query)) {
- if (rs.next()) {
- return rs.getTimestamp(1, cal);
+ }
+
+ /**
+ * Get the query string to determine the current timestamp in the database.
+ *
+ * @return the query string; never null or empty
+ */
+ protected String currentTimestampDatabaseQuery() {
+ return "SELECT CURRENT_TIMESTAMP";
+ }
+
+ @Override
+ public boolean tableExists(
+ Connection connection,
+ TableId tableId
+ ) throws SQLException {
+ DatabaseMetaData metadata = connection.getMetaData();
+ String[] tableTypes = tableTypes(metadata, this.tableTypes);
+ String tableTypeDisplay = displayableTableTypes(tableTypes, "/");
+ log.info("Checking {} dialect for existence of {} {}", this, tableTypeDisplay, tableId);
+ try (ResultSet rs = connection.getMetaData().getTables(
+ tableId.catalogName(),
+ tableId.schemaName(),
+ tableId.tableName(),
+ tableTypes
+ )) {
+ final boolean exists = rs.next();
+ log.info(
+ "Using {} dialect {} {} {}",
+ this,
+ tableTypeDisplay,
+ tableId,
+ exists ? "present" : "absent"
+ );
+ return exists;
+ }
+ }
+
+ protected String displayableTableTypes(String[] types, String delim) {
+ return Arrays.stream(types).sorted().collect(Collectors.joining(delim));
+ }
+
+ @Override
+ public Map<ColumnId, ColumnDefinition> describeColumns(
+ Connection connection,
+ String tablePattern,
+ String columnPattern
+ ) throws SQLException {
+ //if the table pattern is fqn, then just use the actual table name
+ TableId tableId = parseToTableId(tablePattern);
+ String catalog = tableId.catalogName() != null ? tableId.catalogName() : catalogPattern;
+ String schema = tableId.schemaName() != null ? tableId.schemaName() : schemaPattern;
+ return describeColumns(connection, catalog, schema, tableId.tableName(), columnPattern);
+ }
+
+ @Override
+ public Map<ColumnId, ColumnDefinition> describeColumns(
+ Connection connection,
+ String catalogPattern,
+ String schemaPattern,
+ String tablePattern,
+ String columnPattern
+ ) throws SQLException {
+ log.debug(
+ "Querying {} dialect column metadata for catalog:{} schema:{} table:{}",
+ this,
+ catalogPattern,
+ schemaPattern,
+ tablePattern);
+
+ // Get the primary keys of the table(s) ...
+ final Set<ColumnId> pkColumns = primaryKeyColumns(
+ connection,
+ catalogPattern,
+ schemaPattern,
+ tablePattern
+ );
+ Map<ColumnId, ColumnDefinition> results = new HashMap<>();
+ try (ResultSet rs = connection.getMetaData().getColumns(
+ catalogPattern,
+ schemaPattern,
+ tablePattern,
+ columnPattern
+ )) {
+ final int rsColumnCount = rs.getMetaData().getColumnCount();
+ while (rs.next()) {
+ final String catalogName = rs.getString(1);
+ final String schemaName = rs.getString(2);
+ final String tableName = rs.getString(3);
+ final TableId tableId = new TableId(catalogName, schemaName, tableName);
+ final String columnName = rs.getString(4);
+ final ColumnId columnId = new ColumnId(tableId, columnName, null);
+ final int jdbcType = rs.getInt(5);
+ final String typeName = rs.getString(6);
+ final int precision = rs.getInt(7);
+ final int scale = rs.getInt(9);
+ final String typeClassName = null;
+ ColumnDefinition.Nullability nullability;
+ final int nullableValue = rs.getInt(11);
+ switch (nullableValue) {
+ case DatabaseMetaData.columnNoNulls:
+ nullability = ColumnDefinition.Nullability.NOT_NULL;
+ break;
+ case DatabaseMetaData.columnNullable:
+ nullability = ColumnDefinition.Nullability.NULL;
+ break;
+ case DatabaseMetaData.columnNullableUnknown:
+ default:
+ nullability = ColumnDefinition.Nullability.UNKNOWN;
+ break;
+ }
+ Boolean autoIncremented = null;
+ if (rsColumnCount >= 23) {
+ // Not all drivers include all columns ...
+ String isAutoIncremented = rs.getString(23);
+ if ("yes".equalsIgnoreCase(isAutoIncremented)) {
+ autoIncremented = Boolean.TRUE;
+ } else if ("no".equalsIgnoreCase(isAutoIncremented)) {
+ autoIncremented = Boolean.FALSE;
+ }
+ }
+ Boolean signed = null;
+ Boolean caseSensitive = null;
+ Boolean searchable = null;
+ Boolean currency = null;
+ Integer displaySize = null;
+ boolean isPrimaryKey = pkColumns.contains(columnId);
+ if (isPrimaryKey) {
+ // Some DBMSes report pks as null
+ nullability = ColumnDefinition.Nullability.NOT_NULL;
+ }
+ ColumnDefinition defn = columnDefinition(
+ rs,
+ columnId,
+ jdbcType,
+ typeName,
+ typeClassName,
+ nullability,
+ ColumnDefinition.Mutability.UNKNOWN,
+ precision,
+ scale,
+ signed,
+ displaySize,
+ autoIncremented,
+ caseSensitive,
+ searchable,
+ currency,
+ isPrimaryKey
+ );
+ results.put(columnId, defn);
+ }
+ return results;
+ }
+ }
+
+ @Override
+ public Map<ColumnId, ColumnDefinition> describeColumns(Connection conn,
+ TableId tableId, ResultSetMetaData rsMetadata) throws SQLException {
+ ColumnDefAdjuster adjuster;
+ if (tableId == null) {
+ adjuster = new ColumnDefAdjuster();
} else {
- throw new io.openmessaging.connector.api.errors.ConnectException(
- "Unable to get current time from DB using " + this + " and query '" + query + "'"
- );
+ String catalog = tableId.catalogName() != null ? tableId.catalogName() : catalogPattern;
+ String schema = tableId.schemaName() != null ? tableId.schemaName() : schemaPattern;
+ adjuster = ColumnDefAdjuster.create(
+ conn, catalog, schema, tableId.tableName(), null);
}
- }
- } catch (SQLException e) {
- glog.error("Failed to get current time from DB using {} and query '{}'", this, query, e);
- throw e;
- }
- }
-
- /**
- * Get the query string to determine the current timestamp in the database.
- * @return the query string; never null or empty
- */
- protected String currentTimestampDatabaseQuery() {
- return "SELECT CURRENT_TIMESTAMP";
- }
-
- @Override
- public boolean tableExists(
- Connection connection,
- TableId tableId
- ) throws SQLException {
- DatabaseMetaData metadata = connection.getMetaData();
- String[] tableTypes = tableTypes(metadata, this.tableTypes);
- String tableTypeDisplay = displayableTableTypes(tableTypes, "/");
- glog.info("Checking {} dialect for existence of {} {}", this, tableTypeDisplay, tableId);
- try (ResultSet rs = connection.getMetaData().getTables(
- tableId.catalogName(),
- tableId.schemaName(),
- tableId.tableName(),
- tableTypes
- )) {
- final boolean exists = rs.next();
- glog.info(
- "Using {} dialect {} {} {}",
- this,
- tableTypeDisplay,
- tableId,
- exists ? "present" : "absent"
- );
- return exists;
- }
- }
-
- protected String displayableTableTypes(String[] types, String delim) {
- return Arrays.stream(types).sorted().collect(Collectors.joining(delim));
- }
-
- @Override
- public Map<ColumnId, ColumnDefinition> describeColumns(
- Connection connection,
- String tablePattern,
- String columnPattern
- ) throws SQLException {
- //if the table pattern is fqn, then just use the actual table name
- TableId tableId = parseToTableId(tablePattern);
- String catalog = tableId.catalogName() != null ? tableId.catalogName() : catalogPattern;
- String schema = tableId.schemaName() != null ? tableId.schemaName() : schemaPattern;
- return describeColumns(connection, catalog , schema, tableId.tableName(), columnPattern);
- }
-
- @Override
- public Map<ColumnId, ColumnDefinition> describeColumns(
- Connection connection,
- String catalogPattern,
- String schemaPattern,
- String tablePattern,
- String columnPattern
- ) throws SQLException {
- glog.debug(
- "Querying {} dialect column metadata for catalog:{} schema:{} table:{}",
- this,
- catalogPattern,
- schemaPattern,
- tablePattern);
-
- // Get the primary keys of the table(s) ...
- final Set<ColumnId> pkColumns = primaryKeyColumns(
- connection,
- catalogPattern,
- schemaPattern,
- tablePattern
- );
- Map<ColumnId, ColumnDefinition> results = new HashMap<>();
- try (ResultSet rs = connection.getMetaData().getColumns(
- catalogPattern,
- schemaPattern,
- tablePattern,
- columnPattern
- )) {
- final int rsColumnCount = rs.getMetaData().getColumnCount();
- while (rs.next()) {
- final String catalogName = rs.getString(1);
- final String schemaName = rs.getString(2);
- final String tableName = rs.getString(3);
- final TableId tableId = new TableId(catalogName, schemaName, tableName);
- final String columnName = rs.getString(4);
- final ColumnId columnId = new ColumnId(tableId, columnName, null);
- final int jdbcType = rs.getInt(5);
- final String typeName = rs.getString(6);
- final int precision = rs.getInt(7);
- final int scale = rs.getInt(9);
- final String typeClassName = null;
- ColumnDefinition.Nullability nullability;
- final int nullableValue = rs.getInt(11);
- switch (nullableValue) {
- case DatabaseMetaData.columnNoNulls:
- nullability = ColumnDefinition.Nullability.NOT_NULL;
- break;
- case DatabaseMetaData.columnNullable:
- nullability = ColumnDefinition.Nullability.NULL;
- break;
- case DatabaseMetaData.columnNullableUnknown:
- default:
- nullability = ColumnDefinition.Nullability.UNKNOWN;
- break;
+ Map<ColumnId, ColumnDefinition> result = new LinkedHashMap<>();
+ for (int i = 1; i <= rsMetadata.getColumnCount(); ++i) {
+ ColumnDefinition defn = describeColumn(rsMetadata, adjuster, i);
+ result.put(defn.id(), defn);
}
- Boolean autoIncremented = null;
- if (rsColumnCount >= 23) {
- // Not all drivers include all columns ...
- String isAutoIncremented = rs.getString(23);
- if ("yes".equalsIgnoreCase(isAutoIncremented)) {
- autoIncremented = Boolean.TRUE;
- } else if ("no".equalsIgnoreCase(isAutoIncremented)) {
- autoIncremented = Boolean.FALSE;
- }
+ return result;
+ }
+
+ /**
+ * Create a definition for the specified column in the result set.
+ *
+ * @param rsMetadata the result set metadata; may not be null
+ * @param column the column number, starting at 1 for the first column
+ * @return the column definition; never null
+ * @throws SQLException if there is an error accessing the result set metadata
+ */
+ protected ColumnDefinition describeColumn(
+ ResultSetMetaData rsMetadata,
+ ColumnDefAdjuster adjuster,
+ int column
+ ) throws SQLException {
+ String catalog = rsMetadata.getCatalogName(column);
+ String schema = rsMetadata.getSchemaName(column);
+ String tableName = rsMetadata.getTableName(column);
+ TableId tableId = new TableId(catalog, schema, tableName);
+ String name = rsMetadata.getColumnName(column);
+ String alias = rsMetadata.getColumnLabel(column);
+ ColumnId id = new ColumnId(tableId, name, alias);
+
+ ColumnDefinition.Nullability nullability = adjuster.nullable(name);
+ if (nullability == null) {
+ nullability = columnIsNullAble(rsMetadata, column);
+ }
+
+ ColumnDefinition.Mutability mutability = ColumnDefinition.Mutability.MAYBE_WRITABLE;
+ if (rsMetadata.isReadOnly(column)) {
+ mutability = ColumnDefinition.Mutability.READ_ONLY;
+ } else if (rsMetadata.isWritable(column)) {
+ mutability = ColumnDefinition.Mutability.MAYBE_WRITABLE;
+ } else if (rsMetadata.isDefinitelyWritable(column)) {
+ mutability = ColumnDefinition.Mutability.WRITABLE;
+ }
+ return new ColumnDefinition(
+ id,
+ rsMetadata.getColumnType(column),
+ rsMetadata.getColumnTypeName(column),
+ rsMetadata.getColumnClassName(column),
+ nullability,
+ mutability,
+ rsMetadata.getPrecision(column),
+ rsMetadata.getScale(column),
+ rsMetadata.isSigned(column),
+ rsMetadata.getColumnDisplaySize(column),
+ rsMetadata.isAutoIncrement(column),
+ false,
+ rsMetadata.isSearchable(column),
+ rsMetadata.isCurrency(column),
+ false
+ );
+ }
+
+ protected ColumnDefinition.Nullability columnIsNullAble(ResultSetMetaData rsMetadata, int column) throws SQLException {
+ switch (rsMetadata.isNullable(column)) {
+ case ResultSetMetaData.columnNullable:
+ return ColumnDefinition.Nullability.NULL;
+ case ResultSetMetaData.columnNoNulls:
+ return ColumnDefinition.Nullability.NOT_NULL;
+ case ResultSetMetaData.columnNullableUnknown:
+ default:
+ return ColumnDefinition.Nullability.UNKNOWN;
+ }
+ }
+
+ protected Set<ColumnId> primaryKeyColumns(
+ Connection connection,
+ String catalogPattern,
+ String schemaPattern,
+ String tablePattern
+ ) throws SQLException {
+
+ // Get the primary keys of the table(s) ...
+ final Set<ColumnId> pkColumns = new HashSet<>();
+ try (ResultSet rs = connection.getMetaData().getPrimaryKeys(
+ catalogPattern, schemaPattern, tablePattern)) {
+ while (rs.next()) {
+ String catalogName = rs.getString(1);
+ String schemaName = rs.getString(2);
+ String tableName = rs.getString(3);
+ TableId tableId = new TableId(catalogName, schemaName, tableName);
+ final String colName = rs.getString(4);
+ ColumnId columnId = new ColumnId(tableId, colName);
+ pkColumns.add(columnId);
+ }
+ }
+ return pkColumns;
+ }
+
+ @Override
+ public Map<ColumnId, ColumnDefinition> describeColumnsByQuerying(
+ Connection db,
+ TableId tableId
+ ) throws SQLException {
+ String queryStr = "SELECT * FROM {} LIMIT 1";
+ String quotedName = expressionBuilder().append(tableId).toString();
+ try (PreparedStatement stmt = db.prepareStatement(queryStr)) {
+ stmt.setString(1, quotedName);
+ try (ResultSet rs = stmt.executeQuery()) {
+ ResultSetMetaData rsmd = rs.getMetaData();
+ return describeColumns(db, tableId, rsmd);
+ }
+ }
+ }
+
+
+ @Override
+ public TableDefinition describeTable(
+ Connection connection,
+ TableId tableId
+ ) throws SQLException {
+ Map<ColumnId, ColumnDefinition> columnDefns = describeColumns(connection, tableId.catalogName(),
+ tableId.schemaName(),
+ tableId.tableName(), null
+ );
+ if (columnDefns.isEmpty()) {
+ return null;
}
- Boolean signed = null;
- Boolean caseSensitive = null;
- Boolean searchable = null;
- Boolean currency = null;
- Integer displaySize = null;
- boolean isPrimaryKey = pkColumns.contains(columnId);
- if (isPrimaryKey) {
- // Some DBMSes report pks as null
- nullability = ColumnDefinition.Nullability.NOT_NULL;
+ TableType tableType = tableTypeFor(connection, tableId);
+ return new TableDefinition(tableId, columnDefns.values(), tableType);
+ }
+
+ protected TableType tableTypeFor(
+ Connection connection,
+ TableId tableId
+ ) throws SQLException {
+ DatabaseMetaData metadata = connection.getMetaData();
+ String[] tableTypes = tableTypes(metadata, this.tableTypes);
+ String tableTypeDisplay = displayableTableTypes(tableTypes, "/");
+ log.info("Checking {} dialect for type of {} {}", this, tableTypeDisplay, tableId);
+ try (ResultSet rs = connection.getMetaData().getTables(
+ tableId.catalogName(),
+ tableId.schemaName(),
+ tableId.tableName(),
+ tableTypes
+ )) {
+ if (rs.next()) {
+ //final String catalogName = rs.getString(1);
+ //final String schemaName = rs.getString(2);
+ //final String tableName = rs.getString(3);
+ final String tableType = rs.getString(4);
+ try {
+ return TableType.get(tableType);
+ } catch (IllegalArgumentException e) {
+ log.warn(
+ "{} dialect found unknown type '{}' for {} {}; using TABLE",
+ this,
+ tableType,
+ tableTypeDisplay,
+ tableId
+ );
+ return TableType.TABLE;
+ }
+ }
}
- ColumnDefinition defn = columnDefinition(
- rs,
- columnId,
+ log.warn(
+ "{} dialect did not find type for {} {}; using TABLE",
+ this,
+ tableTypeDisplay,
+ tableId
+ );
+ return TableType.TABLE;
+ }
+
+ /**
+ * Create a ColumnDefinition with supplied values and the result set from the {@link
+ * DatabaseMetaData#getColumns(String, String, String, String)} call. By default that method does
+ * not describe whether the column is signed, case sensitive, searchable, currency, or the
+ * preferred display size.
+ *
+ * <p>Subclasses can override this method to extract additional non-standard characteristics from
+ * the result set, and override the characteristics determined using the standard JDBC metadata
+ * columns and supplied as parameters.
+ *
+ * @param resultSet the result set
+ * @param id the column identifier
+ * @param jdbcType the JDBC type of the column
+ * @param typeName the name of the column's type
+ * @param classNameForType the name of the class used as instances of the value when {@link
+ * ResultSet#getObject(int)} is called
+ * @param nullability the nullability of the column
+ * @param mutability the mutability of the column
+ * @param precision the precision of the column for numeric values, or the length for
+ * non-numeric values
+ * @param scale the scale of the column for numeric values; ignored for other values
+ * @param signedNumbers true if the column holds signed numeric values; null if not known
+ * @param displaySize the preferred display size for the column values; null if not known
+ * @param autoIncremented true if the column is auto-incremented; null if not known
+ * @param caseSensitive true if the column values are case-sensitive; null if not known
+ * @param searchable true if the column is searchable; null if no; null if not known known
+ * @param currency true if the column is a currency value
+ * @param isPrimaryKey true if the column is part of the primary key; null if not known known
+ * @return the column definition; never null
+ */
+ protected ColumnDefinition columnDefinition(
+ ResultSet resultSet,
+ ColumnId id,
+ int jdbcType,
+ String typeName,
+ String classNameForType,
+ ColumnDefinition.Nullability nullability,
+ ColumnDefinition.Mutability mutability,
+ int precision,
+ int scale,
+ Boolean signedNumbers,
+ Integer displaySize,
+ Boolean autoIncremented,
+ Boolean caseSensitive,
+ Boolean searchable,
+ Boolean currency,
+ Boolean isPrimaryKey
+ ) {
+ return new ColumnDefinition(
+ id,
jdbcType,
typeName,
- typeClassName,
+ classNameForType,
nullability,
- ColumnDefinition.Mutability.UNKNOWN,
+ mutability,
precision,
scale,
- signed,
- displaySize,
- autoIncremented,
- caseSensitive,
- searchable,
- currency,
- isPrimaryKey
+ signedNumbers != null ? signedNumbers.booleanValue() : false,
+ displaySize != null ? displaySize.intValue() : 0,
+ autoIncremented != null ? autoIncremented.booleanValue() : false,
+ caseSensitive != null ? caseSensitive.booleanValue() : false,
+ searchable != null ? searchable.booleanValue() : false,
+ currency != null ? currency.booleanValue() : false,
+ isPrimaryKey != null ? isPrimaryKey.booleanValue() : false
);
- results.put(columnId, defn);
- }
- return results;
- }
- }
-
- @Override
- public Map<ColumnId, ColumnDefinition> describeColumns(Connection conn,
- TableId tableId,ResultSetMetaData rsMetadata) throws SQLException {
- ColumnDefAdjuster adjuster;
- //querySql模式tableId会是空的
- if(tableId == null){
- adjuster = new ColumnDefAdjuster();
- }else{
- String catalog = tableId.catalogName() != null ? tableId.catalogName() : catalogPattern;
- String schema = tableId.schemaName() != null ? tableId.schemaName() : schemaPattern;
- adjuster = ColumnDefAdjuster.create(
- conn, catalog, schema, tableId.tableName(), null);
- }
- Map<ColumnId, ColumnDefinition> result = new LinkedHashMap<>();
- for (int i = 1; i <= rsMetadata.getColumnCount(); ++i) {
- ColumnDefinition defn = describeColumn(rsMetadata, adjuster, i);
- result.put(defn.id(), defn);
- }
- return result;
- }
-
- /**
- * Create a definition for the specified column in the result set.
- * @param rsMetadata the result set metadata; may not be null
- * @param column the column number, starting at 1 for the first column
- * @return the column definition; never null
- * @throws SQLException if there is an error accessing the result set metadata
- */
- protected ColumnDefinition describeColumn(
- ResultSetMetaData rsMetadata,
- ColumnDefAdjuster adjuster,
- int column
- ) throws SQLException {
- String catalog = rsMetadata.getCatalogName(column);
- String schema = rsMetadata.getSchemaName(column);
- String tableName = rsMetadata.getTableName(column);
- TableId tableId = new TableId(catalog, schema, tableName);
- String name = rsMetadata.getColumnName(column);
- String alias = rsMetadata.getColumnLabel(column);
- ColumnId id = new ColumnId(tableId, name, alias);
-
- ColumnDefinition.Nullability nullability = adjuster.nullable(name);
- if(nullability == null){
- nullability = columnIsNullAble(rsMetadata, column);
- }
-
- ColumnDefinition.Mutability mutability = ColumnDefinition.Mutability.MAYBE_WRITABLE;
- if (rsMetadata.isReadOnly(column)) {
- mutability = ColumnDefinition.Mutability.READ_ONLY;
- } else if (rsMetadata.isWritable(column)) {
- mutability = ColumnDefinition.Mutability.MAYBE_WRITABLE;
- } else if (rsMetadata.isDefinitelyWritable(column)) {
- mutability = ColumnDefinition.Mutability.WRITABLE;
- }
- return new ColumnDefinition(
- id,
- rsMetadata.getColumnType(column),
- rsMetadata.getColumnTypeName(column),
- rsMetadata.getColumnClassName(column),
- nullability,
- mutability,
- rsMetadata.getPrecision(column),
- rsMetadata.getScale(column),
- rsMetadata.isSigned(column),
- rsMetadata.getColumnDisplaySize(column),
- rsMetadata.isAutoIncrement(column),
- false,
- rsMetadata.isSearchable(column),
- rsMetadata.isCurrency(column),
- false
- );
- }
-
- protected ColumnDefinition.Nullability columnIsNullAble(ResultSetMetaData rsMetadata, int column) throws SQLException {
- switch (rsMetadata.isNullable(column)) {
- case ResultSetMetaData.columnNullable:
- return ColumnDefinition.Nullability.NULL;
- case ResultSetMetaData.columnNoNulls:
- return ColumnDefinition.Nullability.NOT_NULL;
- case ResultSetMetaData.columnNullableUnknown:
- default:
- return ColumnDefinition.Nullability.UNKNOWN;
- }
- }
-
- protected Set<ColumnId> primaryKeyColumns(
- Connection connection,
- String catalogPattern,
- String schemaPattern,
- String tablePattern
- ) throws SQLException {
-
- // Get the primary keys of the table(s) ...
- final Set<ColumnId> pkColumns = new HashSet<>();
- try (ResultSet rs = connection.getMetaData().getPrimaryKeys(
- catalogPattern, schemaPattern, tablePattern)) {
- while (rs.next()) {
- String catalogName = rs.getString(1);
- String schemaName = rs.getString(2);
- String tableName = rs.getString(3);
- TableId tableId = new TableId(catalogName, schemaName, tableName);
- final String colName = rs.getString(4);
- ColumnId columnId = new ColumnId(tableId, colName);
- pkColumns.add(columnId);
- }
- }
- return pkColumns;
- }
-
- @Override
- public Map<ColumnId, ColumnDefinition> describeColumnsByQuerying(
- Connection db,
- TableId tableId
- ) throws SQLException {
- String queryStr = "SELECT * FROM {} LIMIT 1";
- String quotedName = expressionBuilder().append(tableId).toString();
- try (PreparedStatement stmt = db.prepareStatement(queryStr)) {
- stmt.setString(1, quotedName);
- try (ResultSet rs = stmt.executeQuery()) {
- ResultSetMetaData rsmd = rs.getMetaData();
- return describeColumns(db, tableId, rsmd);
- }
- }
- }
-
-
- @Override
- public TableDefinition describeTable(
- Connection connection,
- TableId tableId
- ) throws SQLException {
- Map<ColumnId, ColumnDefinition> columnDefns = describeColumns(connection, tableId.catalogName(),
- tableId.schemaName(),
- tableId.tableName(), null
- );
- if (columnDefns.isEmpty()) {
- return null;
- }
- TableType tableType = tableTypeFor(connection, tableId);
- return new TableDefinition(tableId, columnDefns.values(), tableType);
- }
-
- protected TableType tableTypeFor(
- Connection connection,
- TableId tableId
- ) throws SQLException {
- DatabaseMetaData metadata = connection.getMetaData();
- String[] tableTypes = tableTypes(metadata, this.tableTypes);
- String tableTypeDisplay = displayableTableTypes(tableTypes, "/");
- glog.info("Checking {} dialect for type of {} {}", this, tableTypeDisplay, tableId);
- try (ResultSet rs = connection.getMetaData().getTables(
- tableId.catalogName(),
- tableId.schemaName(),
- tableId.tableName(),
- tableTypes
- )) {
- if (rs.next()) {
- //final String catalogName = rs.getString(1);
- //final String schemaName = rs.getString(2);
- //final String tableName = rs.getString(3);
- final String tableType = rs.getString(4);
- try {
- return TableType.get(tableType);
- } catch (IllegalArgumentException e) {
- glog.warn(
- "{} dialect found unknown type '{}' for {} {}; using TABLE",
- this,
- tableType,
- tableTypeDisplay,
- tableId
- );
- return TableType.TABLE;
- }
- }
- }
- glog.warn(
- "{} dialect did not find type for {} {}; using TABLE",
- this,
- tableTypeDisplay,
- tableId
- );
- return TableType.TABLE;
- }
-
- /**
- * Create a ColumnDefinition with supplied values and the result set from the {@link
- * DatabaseMetaData#getColumns(String, String, String, String)} call. By default that method does
- * not describe whether the column is signed, case sensitive, searchable, currency, or the
- * preferred display size.
- *
- * <p>Subclasses can override this method to extract additional non-standard characteristics from
- * the result set, and override the characteristics determined using the standard JDBC metadata
- * columns and supplied as parameters.
- *
- * @param resultSet the result set
- * @param id the column identifier
- * @param jdbcType the JDBC type of the column
- * @param typeName the name of the column's type
- * @param classNameForType the name of the class used as instances of the value when {@link
- * ResultSet#getObject(int)} is called
- * @param nullability the nullability of the column
- * @param mutability the mutability of the column
- * @param precision the precision of the column for numeric values, or the length for
- * non-numeric values
- * @param scale the scale of the column for numeric values; ignored for other values
- * @param signedNumbers true if the column holds signed numeric values; null if not known
- * @param displaySize the preferred display size for the column values; null if not known
- * @param autoIncremented true if the column is auto-incremented; null if not known
- * @param caseSensitive true if the column values are case-sensitive; null if not known
- * @param searchable true if the column is searchable; null if no; null if not known known
- * @param currency true if the column is a currency value
- * @param isPrimaryKey true if the column is part of the primary key; null if not known known
- * @return the column definition; never null
- */
- protected ColumnDefinition columnDefinition(
- ResultSet resultSet,
- ColumnId id,
- int jdbcType,
- String typeName,
- String classNameForType,
- ColumnDefinition.Nullability nullability,
- ColumnDefinition.Mutability mutability,
- int precision,
- int scale,
- Boolean signedNumbers,
- Integer displaySize,
- Boolean autoIncremented,
- Boolean caseSensitive,
- Boolean searchable,
- Boolean currency,
- Boolean isPrimaryKey
- ) {
- return new ColumnDefinition(
- id,
- jdbcType,
- typeName,
- classNameForType,
- nullability,
- mutability,
- precision,
- scale,
- signedNumbers != null ? signedNumbers.booleanValue() : false,
- displaySize != null ? displaySize.intValue() : 0,
- autoIncremented != null ? autoIncremented.booleanValue() : false,
- caseSensitive != null ? caseSensitive.booleanValue() : false,
- searchable != null ? searchable.booleanValue() : false,
- currency != null ? currency.booleanValue() : false,
- isPrimaryKey != null ? isPrimaryKey.booleanValue() : false
- );
- }
+ }
// @Override
// public TimestampIncrementingCriteria criteriaFor(
@@ -884,1005 +908,1006 @@ public class GenericDatabaseDialect implements DatabaseDialect {
// incrementingColumn, incrementingColumnStep, timestampColumns,timestampColumnStep, timeZone);
// }
- /**
- * Determine the name of the field. By default this is the column alias or name.
- * @param columnDefinition the column definition; never null
- * @return the field name; never null
- */
- protected String fieldNameFor(ColumnDefinition columnDefinition) {
- return columnDefinition.id().aliasOrName();
- }
-
-
- @Override
- public String addFieldToSchema(
- ColumnDefinition columnDefn,
- Schema schema,
- int index
- ) {
- return addFieldToSchema(columnDefn, schema, fieldNameFor(columnDefn), index,columnDefn.type(),
- columnDefn.isOptional()
- );
- }
-
- /**
- * add field to schema
- * @param columnDefn
- * @param builder
- * @param fieldName
- * @param index
- * @param sqlType
- * @param optional
- * @return
- */
- @SuppressWarnings("fallthrough")
- protected String addFieldToSchema(
- final ColumnDefinition columnDefn,
- final Schema builder,
- final String fieldName,
- final int index,
- final int sqlType,
- final boolean optional
- ) {
- int precision = columnDefn.precision();
- int scale = columnDefn.scale();
- switch (sqlType) {
- case Types.NULL: {
- glog.debug("JDBC type 'NULL' not currently supported for column '{}'", fieldName);
- return null;
- }
- case Types.BOOLEAN: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.bool().build()));
- break;
- }
-
- // ints <= 8 bits
- case Types.BIT: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int8().build()));
- break;
- }
-
- case Types.TINYINT: {
- if (columnDefn.isSignedNumber()) {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int8().build()));
- } else {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
+ /**
+ * Determine the name of the field. By default this is the column alias or name.
+ *
+ * @param columnDefinition the column definition; never null
+ * @return the field name; never null
+ */
+ protected String fieldNameFor(ColumnDefinition columnDefinition) {
+ return columnDefinition.id().aliasOrName();
+ }
+
+
+ @Override
+ public String addFieldToSchema(
+ ColumnDefinition columnDefn,
+ Schema schema,
+ int index
+ ) {
+ return addFieldToSchema(columnDefn, schema, fieldNameFor(columnDefn), index, columnDefn.type(),
+ columnDefn.isOptional()
+ );
+ }
+
+ /**
+ * add field to schema
+ *
+ * @param columnDefn
+ * @param builder
+ * @param fieldName
+ * @param index
+ * @param sqlType
+ * @param optional
+ * @return
+ */
+ @SuppressWarnings("fallthrough")
+ protected String addFieldToSchema(
+ final ColumnDefinition columnDefn,
+ final Schema builder,
+ final String fieldName,
+ final int index,
+ final int sqlType,
+ final boolean optional
+ ) {
+ int precision = columnDefn.precision();
+ int scale = columnDefn.scale();
+ switch (sqlType) {
+ case Types.NULL: {
+ log.debug("JDBC type 'NULL' not currently supported for column '{}'", fieldName);
+ return null;
+ }
+ case Types.BOOLEAN: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.bool().build()));
+ break;
+ }
+
+ // ints <= 8 bits
+ case Types.BIT: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int8().build()));
+ break;
+ }
+
+ case Types.TINYINT: {
+ if (columnDefn.isSignedNumber()) {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int8().build()));
+ } else {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
+ }
+ break;
+ }
+
+ // 16 bit ints
+ case Types.SMALLINT: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
+ break;
+ }
+
+ // 32 bit ints
+ case Types.INTEGER: {
+ if (columnDefn.isSignedNumber()) {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
+ } else {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int64().build()));
+ }
+ break;
+ }
+
+ // 64 bit ints
+ case Types.BIGINT: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.int64().build()));
+ break;
+ }
+
+ // REAL is a single precision floating point value, i.e. a Java float
+ case Types.REAL: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.float32().build()));
+ break;
+ }
+
+ // FLOAT is, confusingly, double precision and effectively the same as DOUBLE. See REAL
+ // for single precision
+ case Types.FLOAT:
+ case Types.DOUBLE:
+ case Types.DECIMAL:
+ builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
+ break;
+
+ /**
+ * numeric
+ */
+ case Types.NUMERIC:
+ if (mapNumerics == NumericMapping.PRECISION_ONLY) {
+ log.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (scale == 0 && precision <= MAX_INTEGER_TYPE_PRECISION) { // integer
+ builder.addField(new Field(index, fieldName, integerSchema(optional, precision)));
+ break;
+ }
+ } else if (mapNumerics == NumericMapping.BEST_FIT) {
+ log.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
+ if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
+ builder.addField(new Field(index, fieldName, integerSchema(optional, precision)));
+ break;
+ } else if (scale > 0) { // floating point - use double in all cases
+ builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
+ break;
+ }
+ }
+ } else if (mapNumerics == NumericMapping.BEST_FIT_EAGER_DOUBLE) {
+ log.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
+ if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
+ builder.addField(new Field(index, fieldName, integerSchema(optional, precision)));
+ break;
+ }
+ } else if (scale > 0) { // floating point - use double in all cases
+ builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
+ break;
+ }
+ }
+
+ case Types.CHAR:
+ case Types.VARCHAR:
+ case Types.LONGVARCHAR:
+ case Types.NCHAR:
+ case Types.NVARCHAR:
+ case Types.LONGNVARCHAR:
+ case Types.CLOB:
+ case Types.NCLOB:
+ case Types.DATALINK:
+ case Types.SQLXML: {
+ // Some of these types will have fixed size, but we drop this from the schema conversion
+ // since only fixed byte arrays can have a fixed size
+ builder.addField(new Field(index, fieldName, SchemaBuilder.string().build()));
+ break;
+ }
+
+ // Binary == fixed bytes
+ // BLOB, VARBINARY, LONGVARBINARY == bytes
+ case Types.BINARY:
+ case Types.BLOB:
+ case Types.VARBINARY:
+ case Types.LONGVARBINARY: {
+ builder.addField(new Field(index, fieldName, SchemaBuilder.bytes().build()));
+ break;
+ }
+
+ // Date is day + moth + year
+ case Types.DATE: {
+ SchemaBuilder dateSchemaBuilder = DateColumnParser.builder();
+ builder.addField(new Field(index, fieldName, dateSchemaBuilder.build()));
+ break;
+ }
+
+ // Time is a time of day -- hour, minute, seconds, nanoseconds
+ case Types.TIME: {
+ SchemaBuilder timeSchemaBuilder = TimestampColumnParser.builder();
+ builder.addField(new Field(index, fieldName, timeSchemaBuilder.build()));
+ break;
+ }
+
+ // Timestamp is a date + time
+ case Types.TIMESTAMP: {
+ SchemaBuilder tsSchemaBuilder = TimestampColumnParser.builder();
+ builder.addField(new Field(index, fieldName, tsSchemaBuilder.build()));
+ break;
+ }
+
+ case Types.ARRAY:
+ case Types.JAVA_OBJECT:
+ case Types.OTHER:
+ case Types.DISTINCT:
+ case Types.STRUCT:
+ case Types.REF:
+ case Types.ROWID:
+ default: {
+ log.warn("JDBC type {} ({}) not currently supported", sqlType, columnDefn.typeName());
+ return null;
+ }
}
- break;
- }
-
- // 16 bit ints
- case Types.SMALLINT: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
- break;
- }
-
- // 32 bit ints
- case Types.INTEGER: {
- if (columnDefn.isSignedNumber()) {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int32().build()));
+ return fieldName;
+ }
+
+ private Schema integerSchema(boolean optional, int precision) {
+ Schema schema;
+ if (precision > 9) {
+ schema = SchemaBuilder.int64().build();
+ } else if (precision > 2) {
+ schema = SchemaBuilder.int32().build();
} else {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int64().build()));
+ schema = SchemaBuilder.int8().build();
}
- break;
- }
-
- // 64 bit ints
- case Types.BIGINT: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.int64().build()));
- break;
- }
-
- // REAL is a single precision floating point value, i.e. a Java float
- case Types.REAL: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.float32().build()));
- break;
- }
-
- // FLOAT is, confusingly, double precision and effectively the same as DOUBLE. See REAL
- // for single precision
- case Types.FLOAT:
- case Types.DOUBLE:
- case Types.DECIMAL:
- builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
- break;
-
- /**
- * numeric
- */
- case Types.NUMERIC:
- if (mapNumerics == NumericMapping.PRECISION_ONLY) {
- glog.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (scale == 0 && precision <= MAX_INTEGER_TYPE_PRECISION) { // integer
- builder.addField(new Field(index, fieldName, integerSchema(optional, precision)));
- break;
- }
- } else if (mapNumerics == NumericMapping.BEST_FIT) {
- glog.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
- if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
- builder.addField(new Field(index,fieldName, integerSchema(optional, precision)));
- break;
- } else if (scale > 0) { // floating point - use double in all cases
- builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
- break;
- }
- }
- } else if (mapNumerics == NumericMapping.BEST_FIT_EAGER_DOUBLE) {
- glog.debug("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
- if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
- builder.addField(new Field(index, fieldName, integerSchema(optional, precision)));
- break;
+ return schema;
+ }
+
+ /**
+ * execute ddl
+ *
+ * @param connection the connection to use
+ * @param statements the list of DDL statements to execute
+ * @throws SQLException
+ */
+ @Override
+ public void applyDdlStatements(
+ Connection connection,
+ List<String> statements
+ ) throws SQLException {
+ try (Statement statement = connection.createStatement()) {
+ for (String ddlStatement : statements) {
+ statement.executeUpdate(ddlStatement);
}
- } else if (scale > 0) { // floating point - use double in all cases
- builder.addField(new Field(index, fieldName, SchemaBuilder.float64().build()));
- break;
- }
}
+ }
- case Types.CHAR:
- case Types.VARCHAR:
- case Types.LONGVARCHAR:
- case Types.NCHAR:
- case Types.NVARCHAR:
- case Types.LONGNVARCHAR:
- case Types.CLOB:
- case Types.NCLOB:
- case Types.DATALINK:
- case Types.SQLXML: {
- // Some of these types will have fixed size, but we drop this from the schema conversion
- // since only fixed byte arrays can have a fixed size
- builder.addField(new Field(index, fieldName, SchemaBuilder.string().build()));
- break;
- }
-
- // Binary == fixed bytes
- // BLOB, VARBINARY, LONGVARBINARY == bytes
- case Types.BINARY:
- case Types.BLOB:
- case Types.VARBINARY:
- case Types.LONGVARBINARY: {
- builder.addField(new Field(index, fieldName, SchemaBuilder.bytes().build()));
- break;
- }
-
- // Date is day + moth + year
- case Types.DATE: {
- SchemaBuilder dateSchemaBuilder = DateColumnParser.builder();
- builder.addField(new Field(index, fieldName,dateSchemaBuilder.build()));
- break;
- }
-
- // Time is a time of day -- hour, minute, seconds, nanoseconds
- case Types.TIME: {
- SchemaBuilder timeSchemaBuilder = TimestampColumnParser.builder();
- builder.addField(new Field(index, fieldName, timeSchemaBuilder.build()));
- break;
- }
-
- // Timestamp is a date + time
- case Types.TIMESTAMP: {
- SchemaBuilder tsSchemaBuilder = TimestampColumnParser.builder();
- builder.addField(new Field(index, fieldName,tsSchemaBuilder.build()));
- break;
- }
-
- case Types.ARRAY:
- case Types.JAVA_OBJECT:
- case Types.OTHER:
- case Types.DISTINCT:
- case Types.STRUCT:
- case Types.REF:
- case Types.ROWID:
- default: {
- glog.warn("JDBC type {} ({}) not currently supported", sqlType, columnDefn.typeName());
- return null;
- }
- }
- return fieldName;
- }
-
- private Schema integerSchema(boolean optional, int precision) {
- Schema schema;
- if (precision > 9) {
- schema =SchemaBuilder.int64().build();
- } else if (precision > 2) {
- schema =SchemaBuilder.int32().build();
- } else {
- schema =SchemaBuilder.int8().build();
- }
- return schema;
- }
-
- /**
- * execute ddl
- * @param connection the connection to use
- * @param statements the list of DDL statements to execute
- * @throws SQLException
- */
- @Override
- public void applyDdlStatements(
- Connection connection,
- List<String> statements
- ) throws SQLException {
- try (Statement statement = connection.createStatement()) {
- for (String ddlStatement : statements) {
- statement.executeUpdate(ddlStatement);
- }
- }
- }
-
- @Override
- public ColumnConverter createColumnConverter(
- ColumnMapping mapping
- ) {
- return columnConverterFor(
- mapping,
- mapping.columnDefn(),
- mapping.columnNumber(),
- jdbcDriverInfo().jdbcVersionAtLeast(4, 0)
- );
- }
-
- @SuppressWarnings({"deprecation", "fallthrough"})
- protected ColumnConverter columnConverterFor(
- final ColumnMapping mapping,
- final ColumnDefinition defn,
- final int col,
- final boolean isJdbc4
- ) {
- switch (mapping.columnDefn().type()) {
-
- case Types.BOOLEAN: {
- return rs -> rs.getBoolean(col);
- }
-
- case Types.BIT: {
- /**
- * BIT should be either 0 or 1.
- * TODO: Postgres handles this differently, returning a string "t" or "f". See the
- * elasticsearch-jdbc plugin for an example of how this is handled
- */
- return rs -> rs.getByte(col);
- }
-
- // 8 bits int
- case Types.TINYINT: {
- if (defn.isSignedNumber()) {
- return rs -> rs.getByte(col);
- } else {
- return rs -> rs.getShort(col);
+ @Override
+ public ColumnConverter createColumnConverter(
+ ColumnMapping mapping
+ ) {
+ return columnConverterFor(
+ mapping,
+ mapping.columnDefn(),
+ mapping.columnNumber(),
+ jdbcDriverInfo().jdbcVersionAtLeast(4, 0)
+ );
+ }
+
+ @SuppressWarnings({"deprecation", "fallthrough"})
+ protected ColumnConverter columnConverterFor(
+ final ColumnMapping mapping,
+ final ColumnDefinition defn,
+ final int col,
+ final boolean isJdbc4
+ ) {
+ switch (mapping.columnDefn().type()) {
+
+ case Types.BOOLEAN: {
+ return rs -> rs.getBoolean(col);
+ }
+
+ case Types.BIT: {
+ /**
+ * BIT should be either 0 or 1.
+ * TODO: Postgres handles this differently, returning a string "t" or "f". See the
+ * elasticsearch-jdbc plugin for an example of how this is handled
+ */
+ return rs -> rs.getByte(col);
+ }
+
+ // 8 bits int
+ case Types.TINYINT: {
+ if (defn.isSignedNumber()) {
+ return rs -> rs.getByte(col);
+ } else {
+ return rs -> rs.getShort(col);
+ }
+ }
+
+ // 16 bits int
+ case Types.SMALLINT: {
+ if (defn.isSignedNumber()) {
+ return rs -> rs.getShort(col);
+ } else {
+ return rs -> rs.getInt(col);
+ }
+ }
+
+ // 32 bits int
+ case Types.INTEGER: {
+ if (defn.isSignedNumber()) {
+ return rs -> rs.getInt(col);
+ } else {
+ return rs -> rs.getLong(col);
+ }
+ }
+
+ // 64 bits int
+ case Types.BIGINT: {
+ return rs -> rs.getLong(col);
+ }
+
+ // REAL is a single precision floating point value, i.e. a Java float
+ case Types.REAL: {
+ return rs -> rs.getFloat(col);
+ }
+
+ // FLOAT is, confusingly, double precision and effectively the same as DOUBLE. See REAL
+ // for single precision
+ case Types.FLOAT:
+ case Types.DOUBLE: {
+ return rs -> rs.getDouble(col);
+ }
+
+ case Types.NUMERIC:
+ if (mapNumerics == NumericMapping.PRECISION_ONLY) {
+ int precision = defn.precision();
+ int scale = defn.scale();
+ log.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (scale == 0 && precision <= MAX_INTEGER_TYPE_PRECISION) { // integer
+ if (precision > 9) {
+ return rs -> rs.getLong(col);
+ } else if (precision > 4) {
+ return rs -> rs.getInt(col);
+ } else if (precision > 2) {
+ return rs -> rs.getShort(col);
+ } else {
+ return rs -> rs.getByte(col);
+ }
+ }
+ } else if (mapNumerics == NumericMapping.BEST_FIT) {
+ int precision = defn.precision();
+ int scale = defn.scale();
+ log.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
+ if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
+ if (precision > 9) {
+ return rs -> rs.getLong(col);
+ } else if (precision > 4) {
+ return rs -> rs.getInt(col);
+ } else if (precision > 2) {
+ return rs -> rs.getShort(col);
+ } else {
+ return rs -> rs.getByte(col);
+ }
+ } else if (scale > 0) { // floating point - use double in all cases
+ return rs -> rs.getDouble(col);
+ }
+ }
+ } else if (mapNumerics == NumericMapping.BEST_FIT_EAGER_DOUBLE) {
+ int precision = defn.precision();
+ int scale = defn.scale();
+ log.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
+ if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
+ if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
+ if (precision > 9) {
+ return rs -> rs.getLong(col);
+ } else if (precision > 4) {
+ return rs -> rs.getInt(col);
+ } else if (precision > 2) {
+ return rs -> rs.getShort(col);
+ } else {
+ return rs -> rs.getByte(col);
+ }
+ }
+ } else if (scale > 0) { // floating point - use double in all cases
+ return rs -> rs.getDouble(col);
+ }
+ }
+ // fallthrough
+
+ case Types.DECIMAL: {
+ final int precision = defn.precision();
+ log.debug("DECIMAL with precision: '{}' and scale: '{}'", precision, defn.scale());
+ final int scale = decimalScale(defn);
+ return rs -> rs.getBigDecimal(col, scale);
+ }
+
+ case Types.CHAR:
+ case Types.VARCHAR:
+ case Types.LONGVARCHAR: {
+ return rs -> rs.getString(col);
+ }
+
+ case Types.NCHAR:
+ case Types.NVARCHAR:
+ case Types.LONGNVARCHAR: {
+ return rs -> rs.getNString(col);
+ }
+
+ // Binary == fixed, VARBINARY and LONGVARBINARY == bytes
+ case Types.BINARY:
+ case Types.VARBINARY:
+ case Types.LONGVARBINARY: {
+ return rs -> rs.getBytes(col);
+ }
+
+ // Date is day + month + year
+ case Types.DATE: {
+ return rs -> rs.getDate(col,
+ DateTimeUtils.getTimeZoneCalendar(TimeZone.getTimeZone(ZoneOffset.UTC)));
+ }
+
+ // Time is a time of day -- hour, minute, seconds, nanoseconds
+ case Types.TIME: {
+ return rs -> rs.getTime(col, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ }
+
+ // Timestamp is a date + time
+ case Types.TIMESTAMP: {
+ return rs -> rs.getTimestamp(col, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ }
+
+ // Datalink is basically a URL -> string
+ case Types.DATALINK: {
+ return rs -> {
+ URL url = rs.getURL(col);
+ return url != null ? url.toString() : null;
+ };
+ }
+
+ // BLOB == fixed
+ case Types.BLOB: {
+ return rs -> {
+ Blob blob = rs.getBlob(col);
+ if (blob == null) {
+ return null;
+ } else {
+ try {
+ if (blob.length() > Integer.MAX_VALUE) {
+ throw new IOException("Can't process BLOBs longer than " + Integer.MAX_VALUE);
+ }
+ return blob.getBytes(1, (int) blob.length());
+ } finally {
+ if (isJdbc4) {
+ free(blob);
+ }
+ }
+ }
+ };
+ }
+ case Types.CLOB:
+ return rs -> {
+ Clob clob = rs.getClob(col);
+ if (clob == null) {
+ return null;
+ } else {
+ try {
+ if (clob.length() > Integer.MAX_VALUE) {
+ throw new IOException("Can't process CLOBs longer than " + Integer.MAX_VALUE);
+ }
+ return clob.getSubString(1, (int) clob.length());
+ } finally {
+ if (isJdbc4) {
+ free(clob);
+ }
+ }
+ }
+ };
+ case Types.NCLOB: {
+ return rs -> {
+ Clob clob = rs.getNClob(col);
+ if (clob == null) {
+ return null;
+ } else {
+ try {
+ if (clob.length() > Integer.MAX_VALUE) {
+ throw new IOException("Can't process NCLOBs longer than " + Integer.MAX_VALUE);
+ }
+ return clob.getSubString(1, (int) clob.length());
+ } finally {
+ if (isJdbc4) {
+ free(clob);
+ }
+ }
+ }
+ };
+ }
+
+ // XML -> string
+ case Types.SQLXML: {
+ return rs -> {
+ SQLXML xml = rs.getSQLXML(col);
+ return xml != null ? xml.getString() : null;
+ };
+ }
+
+ case Types.NULL:
+ case Types.ARRAY:
+ case Types.JAVA_OBJECT:
+ case Types.OTHER:
+ case Types.DISTINCT:
+ case Types.STRUCT:
+ case Types.REF:
+ case Types.ROWID:
+ default: {
+ // These are not currently supported, but we don't want to log something for every single
+ // record we translate. There will already be errors logged for the schema translation
+ break;
+ }
}
- }
+ return null;
+ }
- // 16 bits int
- case Types.SMALLINT: {
- if (defn.isSignedNumber()) {
- return rs -> rs.getShort(col);
- } else {
- return rs -> rs.getInt(col);
+ protected int decimalScale(ColumnDefinition defn) {
+ return defn.scale() == NUMERIC_TYPE_SCALE_UNSET ? NUMERIC_TYPE_SCALE_HIGH : defn.scale();
+ }
+
+ /**
+ * Called when the object has been fully read and {@link Blob#free()} should be called.
+ *
+ * @param blob the Blob; never null
+ * @throws SQLException if there is a problem calling free()
+ */
+ protected void free(Blob blob) throws SQLException {
+ blob.free();
+ }
+
+ /**
+ * Called when the object has been fully read and {@link Clob#free()} should be called.
+ *
+ * @param clob the Clob; never null
+ * @throws SQLException if there is a problem calling free()
+ */
+ protected void free(Clob clob) throws SQLException {
+ clob.free();
+ }
+
+ @Override
+ public String buildInsertStatement(
+ TableId table,
+ Collection<ColumnId> keyColumns,
+ Collection<ColumnId> nonKeyColumns
+ ) {
+ ExpressionBuilder builder = expressionBuilder();
+ builder.append("INSERT INTO ");
+ builder.append(table);
+ builder.append("(");
+ builder.appendList()
+ .delimitedBy(",")
+ .transformedBy(ExpressionBuilder.columnNames())
+ .of(keyColumns, nonKeyColumns);
+ builder.append(") VALUES(");
+ builder.appendMultiple(",", "?", keyColumns.size() + nonKeyColumns.size());
+ builder.append(")");
+ return builder.toString();
+ }
+
+ @Override
+ public String buildUpdateStatement(
+ TableId table,
+ Collection<ColumnId> keyColumns,
+ Collection<ColumnId> nonKeyColumns
+ ) {
+ ExpressionBuilder builder = expressionBuilder();
+ builder.append("UPDATE ");
+ builder.append(table);
+ builder.append(" SET ");
+ builder.appendList()
+ .delimitedBy(", ")
+ .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
+ .of(nonKeyColumns);
+ if (!keyColumns.isEmpty()) {
+ builder.append(" WHERE ");
+ builder.appendList()
+ .delimitedBy(" AND ")
+ .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
+ .of(keyColumns);
}
- }
+ return builder.toString();
+ }
- // 32 bits int
- case Types.INTEGER: {
- if (defn.isSignedNumber()) {
- return rs -> rs.getInt(col);
- } else {
- return rs -> rs.getLong(col);
+ @Override
+ public String buildUpsertQueryStatement(
+ TableId table,
+ Collection<ColumnId> keyColumns,
+ Collection<ColumnId> nonKeyColumns
+ ) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public final String buildDeleteStatement(
+ TableId table,
+ Collection<ColumnId> keyColumns
+ ) {
+ ExpressionBuilder builder = expressionBuilder();
+ builder.append("DELETE FROM ");
+ builder.append(table);
+ if (!keyColumns.isEmpty()) {
+ builder.append(" WHERE ");
+ builder.appendList()
+ .delimitedBy(" AND ")
+ .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
+ .of(keyColumns);
}
- }
-
- // 64 bits int
- case Types.BIGINT: {
- return rs -> rs.getLong(col);
- }
-
- // REAL is a single precision floating point value, i.e. a Java float
- case Types.REAL: {
- return rs -> rs.getFloat(col);
- }
-
- // FLOAT is, confusingly, double precision and effectively the same as DOUBLE. See REAL
- // for single precision
- case Types.FLOAT:
- case Types.DOUBLE: {
- return rs -> rs.getDouble(col);
- }
-
- case Types.NUMERIC:
- if (mapNumerics == NumericMapping.PRECISION_ONLY) {
- int precision = defn.precision();
- int scale = defn.scale();
- glog.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (scale == 0 && precision <= MAX_INTEGER_TYPE_PRECISION) { // integer
- if (precision > 9) {
- return rs -> rs.getLong(col);
- } else if (precision > 4) {
- return rs -> rs.getInt(col);
- } else if (precision > 2) {
- return rs -> rs.getShort(col);
+ return builder.toString();
+ }
+
+ /**
+ * table moode
+ *
+ * @return
+ */
+ @Override
+ public String buildSelectTableMode() {
+ return "SELECT * FROM ";
+ }
+
+ @Override
+ public void buildSelectTable(ExpressionBuilder builder, TableId tableId) {
+ String mode = buildSelectTableMode();
+ builder.append(mode).append(tableId);
+ }
+
+ @Override
+ public StatementBinder statementBinder(
+ PreparedStatement statement,
+ JdbcSinkConfig.PrimaryKeyMode pkMode,
+ SchemaPair schemaPair,
+ FieldsMetadata fieldsMetadata,
+ TableDefinition tableDefinition,
+ JdbcSinkConfig.InsertMode insertMode) {
+ return new PreparedStatementBinder(
+ this,
+ statement,
+ pkMode,
+ schemaPair,
+ fieldsMetadata,
+ tableDefinition,
+ insertMode
+ );
+ }
+
+ @Override
+ public void bindField(
+ PreparedStatement statement,
+ int index, Schema schema,
+ Object value,
+ ColumnDefinition colDef) throws SQLException {
+ if (value == null) {
+ Integer type = getSqlTypeForSchema(schema);
+ if (type != null) {
+ statement.setNull(index, type);
} else {
- return rs -> rs.getByte(col);
+ statement.setObject(index, null);
}
- }
- } else if (mapNumerics == NumericMapping.BEST_FIT) {
- int precision = defn.precision();
- int scale = defn.scale();
- glog.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
- if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
- if (precision > 9) {
- return rs -> rs.getLong(col);
- } else if (precision > 4) {
- return rs -> rs.getInt(col);
- } else if (precision > 2) {
- return rs -> rs.getShort(col);
- } else {
- return rs -> rs.getByte(col);
- }
- } else if (scale > 0) { // floating point - use double in all cases
- return rs -> rs.getDouble(col);
+ } else {
+ boolean bound = maybeBindLogical(statement, index, schema, value, null);
+ if (!bound) {
+ bound = maybeBindPrimitive(statement, index, schema, value);
}
- }
- } else if (mapNumerics == NumericMapping.BEST_FIT_EAGER_DOUBLE) {
- int precision = defn.precision();
- int scale = defn.scale();
- glog.trace("NUMERIC with precision: '{}' and scale: '{}'", precision, scale);
- if (scale < 1 && scale >= NUMERIC_TYPE_SCALE_LOW) { // integer
- if (precision <= MAX_INTEGER_TYPE_PRECISION) { // fits in primitive data types.
- if (precision > 9) {
- return rs -> rs.getLong(col);
- } else if (precision > 4) {
- return rs -> rs.getInt(col);
- } else if (precision > 2) {
- return rs -> rs.getShort(col);
- } else {
- return rs -> rs.getByte(col);
- }
+ if (!bound) {
+ throw new io.openmessaging.connector.api.errors.ConnectException("Unsupported source data type: " + schema.getFieldType());
}
- } else if (scale > 0) { // floating point - use double in all cases
- return rs -> rs.getDouble(col);
- }
}
- // fallthrough
-
- case Types.DECIMAL: {
- final int precision = defn.precision();
- glog.debug("DECIMAL with precision: '{}' and scale: '{}'", precision, defn.scale());
- final int scale = decimalScale(defn);
- return rs -> rs.getBigDecimal(col, scale);
- }
-
- case Types.CHAR:
- case Types.VARCHAR:
- case Types.LONGVARCHAR: {
- return rs -> rs.getString(col);
- }
-
- case Types.NCHAR:
- case Types.NVARCHAR:
- case Types.LONGNVARCHAR: {
- return rs -> rs.getNString(col);
- }
-
- // Binary == fixed, VARBINARY and LONGVARBINARY == bytes
- case Types.BINARY:
- case Types.VARBINARY:
- case Types.LONGVARBINARY: {
- return rs -> rs.getBytes(col);
- }
-
- // Date is day + month + year
- case Types.DATE: {
- return rs -> rs.getDate(col,
- DateTimeUtils.getTimeZoneCalendar(TimeZone.getTimeZone(ZoneOffset.UTC)));
- }
-
- // Time is a time of day -- hour, minute, seconds, nanoseconds
- case Types.TIME: {
- return rs -> rs.getTime(col, DateTimeUtils.getTimeZoneCalendar(timeZone));
- }
-
- // Timestamp is a date + time
- case Types.TIMESTAMP: {
- return rs -> rs.getTimestamp(col, DateTimeUtils.getTimeZoneCalendar(timeZone));
- }
-
- // Datalink is basically a URL -> string
- case Types.DATALINK: {
- return rs -> {
- URL url = rs.getURL(col);
- return (url != null ? url.toString() : null);
- };
- }
+ }
- // BLOB == fixed
- case Types.BLOB: {
- return rs -> {
- Blob blob = rs.getBlob(col);
- if (blob == null) {
- return null;
- } else {
- try {
- if (blob.length() > Integer.MAX_VALUE) {
- throw new IOException("Can't process BLOBs longer than " + Integer.MAX_VALUE);
- }
- return blob.getBytes(1, (int) blob.length());
- } finally {
- if (isJdbc4) {
- free(blob);
- }
+ protected boolean maybeBindLogical(
+ PreparedStatement statement,
+ int index,
+ Schema schema,
+ Object value,
+ ColumnDefinition colDef
+ ) throws SQLException {
+ if (schema.getName() != null) {
+ switch (schema.getName()) {
+ case DateColumnParser.LOGICAL_NAME:
+ java.sql.Date date;
+ if (value instanceof java.util.Date) {
+ date = new java.sql.Date(((java.util.Date) value).getTime());
+ } else {
+ date = new java.sql.Date((int) value);
+ }
+ statement.setDate(
+ index, date,
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ case TimeColumnParser.LOGICAL_NAME:
+ java.sql.Time time;
+ if (value instanceof java.util.Date) {
+ time = new java.sql.Time(((java.util.Date) value).getTime());
+ } else {
+ time = new java.sql.Time((int) value);
+ }
+ statement.setTime(
+ index, time,
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ case TimestampColumnParser.LOGICAL_NAME:
+ java.sql.Timestamp timestamp;
+ if (value instanceof java.util.Date) {
+ timestamp = new java.sql.Timestamp(((java.util.Date) value).getTime());
+ } else {
+ timestamp = new java.sql.Timestamp((long) value);
+ }
+ statement.setTimestamp(
+ index, timestamp,
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ default:
+ return false;
}
- }
- };
- }
- case Types.CLOB:
- return rs -> {
- Clob clob = rs.getClob(col);
- if (clob == null) {
+ }
+ return false;
+ }
+
+
+ @Override
+ public TimestampIncrementingCriteria criteriaFor(ColumnId incrementingColumn, List<ColumnId> timestampColumns) {
+ return new TimestampIncrementingCriteria(
+ incrementingColumn, timestampColumns, timeZone);
+ }
+
+ @Override
+ public Long getMinTimestampValue(Connection con, String tableOrQuery, List<String> timestampColumns) throws SQLException {
+ if (timestampColumns == null || timestampColumns.isEmpty()) {
return null;
- } else {
- try {
- if (clob.length() > Integer.MAX_VALUE) {
- throw new IOException("Can't process CLOBs longer than " + Integer.MAX_VALUE);
- }
- return clob.getSubString(1, (int) clob.length());
- } finally {
- if (isJdbc4) {
- free(clob);
- }
+ }
+ StringBuilder builder = new StringBuilder();
+ builder.append("SELECT ");
+ boolean appendComma = false;
+ for (String column : timestampColumns) {
+ builder.append("MIN(");
+ builder.append(column);
+ builder.append(")");
+ if (appendComma) {
+ builder.append(",");
+ } else {
+ appendComma = true;
}
- }
- };
- case Types.NCLOB: {
- return rs -> {
- Clob clob = rs.getNClob(col);
- if (clob == null) {
- return null;
- } else {
- try {
- if (clob.length() > Integer.MAX_VALUE) {
- throw new IOException("Can't process NCLOBs longer than " + Integer.MAX_VALUE);
- }
- return clob.getSubString(1, (int) clob.length());
- } finally {
- if (isJdbc4) {
- free(clob);
- }
+ }
+ builder.append(" FROM ");
+ builder.append(tableOrQuery);
+ String querySql = builder.toString();
+ PreparedStatement st = con.prepareStatement(querySql);
+ ResultSet resultSet = st.executeQuery();
+ ResultSetMetaData metaData = resultSet.getMetaData();
+ long minTimestampValue = Long.MAX_VALUE;
+ for (int i = 1; i <= metaData.getColumnCount(); ++i) {
+ long t = resultSet.getLong(i);
+ minTimestampValue = Math.min(minTimestampValue, t);
+ }
+ st.close();
+ return minTimestampValue;
+ }
+
+ /**
+ * Dialects not supporting `setObject(index, null)` can override this method
+ * to provide a specific sqlType, as per the JDBC documentation
+ * https://docs.oracle.com/javase/7/docs/api/java/sql/PreparedStatement.html
+ *
+ * @param schema the schema
+ * @return the SQL type
+ */
+ protected Integer getSqlTypeForSchema(Schema schema) {
+ return null;
+ }
+
+ protected boolean maybeBindPrimitive(
+ PreparedStatement statement,
+ int index,
+ Schema schema,
+ Object value
+ ) throws SQLException {
+ switch (schema.getFieldType()) {
+ case INT8:
+ statement.setByte(index, (Byte) value);
+ break;
+ case INT32:
+ statement.setInt(index, (Integer) value);
+ break;
+ case INT64:
+ statement.setLong(index, (Long) value);
+ break;
+ case FLOAT32:
+ statement.setFloat(index, (Float) value);
+ break;
+ case FLOAT64:
+ statement.setDouble(index, (Double) value);
+ break;
+ case BOOLEAN:
+ statement.setBoolean(index, (Boolean) value);
+ break;
+ case STRING:
+ statement.setString(index, (String) value);
+ break;
+ case BYTES:
+ final byte[] bytes;
+ if (value instanceof ByteBuffer) {
+ final ByteBuffer buffer = ((ByteBuffer) value).slice();
+ bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ } else {
+ bytes = (byte[]) value;
+ }
+ statement.setBytes(index, bytes);
+ break;
+ case DATETIME:
+ java.sql.Date date;
+ if (value instanceof java.util.Date) {
+ date = new java.sql.Date(((java.util.Date) value).getTime());
+ } else {
+ date = new java.sql.Date((int) value);
+ }
+ statement.setDate(
+ index, date,
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+
+
+ break;
+ default:
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public String buildCreateTableStatement(
+ TableId table,
+ Collection<SinkRecordField> fields) {
+ ExpressionBuilder builder = expressionBuilder();
+ final List<String> pkFieldNames = extractPrimaryKeyFieldNames(fields);
+ builder.append("CREATE TABLE ");
+ builder.append(table);
+ builder.append(" (");
+ writeColumnsSpec(builder, fields);
+ if (!pkFieldNames.isEmpty()) {
+ builder.append(",");
+ builder.append(System.lineSeparator());
+ builder.append("PRIMARY KEY(");
+ builder.appendList()
+ .delimitedBy(",")
+ .transformedBy(ExpressionBuilder.quote())
+ .of(pkFieldNames);
+ builder.append(")");
+ }
+ builder.append(")");
+ return builder.toString();
+ }
+
+ @Override
+ public String buildDropTableStatement(
+ TableId table,
+ DropOptions options
+ ) {
+ ExpressionBuilder builder = expressionBuilder();
+
+ builder.append("DROP TABLE ");
+ builder.append(table);
+ if (options.ifExists()) {
+ builder.append(" IF EXISTS");
+ }
+ if (options.cascade()) {
+ builder.append(" CASCADE");
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public List<String> buildAlterTable(
+ TableId table,
+ Collection<SinkRecordField> fields
+ ) {
+ final boolean newlines = fields.size() > 1;
+
+ final ExpressionBuilder.Transform<SinkRecordField> transform = (builder, field) -> {
+ if (newlines) {
+ builder.appendNewLine();
}
- }
+ builder.append("ADD ");
+ writeColumnSpec(builder, field);
};
- }
- // XML -> string
- case Types.SQLXML: {
- return rs -> {
- SQLXML xml = rs.getSQLXML(col);
- return xml != null ? xml.getString() : null;
+ ExpressionBuilder builder = expressionBuilder();
+ builder.append("ALTER TABLE ");
+ builder.append(table);
+ builder.append(" ");
+ builder.appendList()
+ .delimitedBy(",")
+ .transformedBy(transform)
+ .of(fields);
+ return Collections.singletonList(builder.toString());
+ }
+
+ @Override
+ public void validateColumnTypes(
+ ResultSetMetaData rsMetadata,
+ List<ColumnId> columns
+ ) throws io.openmessaging.connector.api.errors.ConnectException {
+ }
+
+ protected List<String> extractPrimaryKeyFieldNames(Collection<SinkRecordField> fields) {
+ final List<String> pks = new ArrayList<>();
+ for (SinkRecordField f : fields) {
+ // todo add check column info
+ if (f.isPrimaryKey()) {
+ pks.add(f.name());
+ }
+ }
+ return pks;
+ }
+
+ protected void writeColumnsSpec(
+ ExpressionBuilder builder,
+ Collection<SinkRecordField> fields
+ ) {
+ ExpressionBuilder.Transform<SinkRecordField> transform = (b, field) -> {
+ b.append(System.lineSeparator());
+ writeColumnSpec(b, field);
};
- }
-
- case Types.NULL:
- case Types.ARRAY:
- case Types.JAVA_OBJECT:
- case Types.OTHER:
- case Types.DISTINCT:
- case Types.STRUCT:
- case Types.REF:
- case Types.ROWID:
- default: {
- // These are not currently supported, but we don't want to log something for every single
- // record we translate. There will already be errors logged for the schema translation
- break;
- }
- }
- return null;
- }
-
- protected int decimalScale(ColumnDefinition defn) {
- return defn.scale() == NUMERIC_TYPE_SCALE_UNSET ? NUMERIC_TYPE_SCALE_HIGH : defn.scale();
- }
-
- /**
- * Called when the object has been fully read and {@link Blob#free()} should be called.
- *
- * @param blob the Blob; never null
- * @throws SQLException if there is a problem calling free()
- */
- protected void free(Blob blob) throws SQLException {
- blob.free();
- }
-
- /**
- * Called when the object has been fully read and {@link Clob#free()} should be called.
- *
- * @param clob the Clob; never null
- * @throws SQLException if there is a problem calling free()
- */
- protected void free(Clob clob) throws SQLException {
- clob.free();
- }
-
- @Override
- @SuppressWarnings("deprecation")
- public String buildInsertStatement(
- TableId table,
- Collection<ColumnId> keyColumns,
- Collection<ColumnId> nonKeyColumns
- ) {
- ExpressionBuilder builder = expressionBuilder();
- builder.append("INSERT INTO ");
- builder.append(table);
- builder.append("(");
- builder.appendList()
- .delimitedBy(",")
- .transformedBy(ExpressionBuilder.columnNames())
- .of(keyColumns, nonKeyColumns);
- builder.append(") VALUES(");
- builder.appendMultiple(",", "?", keyColumns.size() + nonKeyColumns.size());
- builder.append(")");
- return builder.toString();
- }
-
- @Override
- public String buildUpdateStatement(
- TableId table,
- Collection<ColumnId> keyColumns,
- Collection<ColumnId> nonKeyColumns
- ) {
- ExpressionBuilder builder = expressionBuilder();
- builder.append("UPDATE ");
- builder.append(table);
- builder.append(" SET ");
- builder.appendList()
- .delimitedBy(", ")
- .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
- .of(nonKeyColumns);
- if (!keyColumns.isEmpty()) {
- builder.append(" WHERE ");
- builder.appendList()
- .delimitedBy(" AND ")
- .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
- .of(keyColumns);
- }
- return builder.toString();
- }
-
- @Override
- @SuppressWarnings("deprecation")
- public String buildUpsertQueryStatement(
- TableId table,
- Collection<ColumnId> keyColumns,
- Collection<ColumnId> nonKeyColumns
- ) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public final String buildDeleteStatement(
- TableId table,
- Collection<ColumnId> keyColumns
- ) {
- ExpressionBuilder builder = expressionBuilder();
- builder.append("DELETE FROM ");
- builder.append(table);
- if (!keyColumns.isEmpty()) {
- builder.append(" WHERE ");
- builder.appendList()
- .delimitedBy(" AND ")
- .transformedBy(ExpressionBuilder.columnNamesWith(" = ?"))
- .of(keyColumns);
- }
- return builder.toString();
- }
-
- /**
- * table moode
- * @return
- */
- @Override
- public String buildSelectTableMode() {
- return "SELECT * FROM ";
- }
-
- @Override
- public void buildSelectTable(ExpressionBuilder builder, TableId tableId) {
- String mode = buildSelectTableMode();
- builder.append(mode).append(tableId);
- }
-
- @Override
- public StatementBinder statementBinder(
- PreparedStatement statement,
- JdbcSinkConfig.PrimaryKeyMode pkMode,
- SchemaPair schemaPair,
- FieldsMetadata fieldsMetadata,
- TableDefinition tableDefinition,
- JdbcSinkConfig.InsertMode insertMode) {
- return new PreparedStatementBinder(
- this,
- statement,
- pkMode,
- schemaPair,
- fieldsMetadata,
- tableDefinition,
- insertMode
- );
- }
-
- @Override
- public void bindField(
- PreparedStatement statement,
- int index, Schema schema,
- Object value,
- ColumnDefinition colDef) throws SQLException {
- if (value == null) {
- Integer type = getSqlTypeForSchema(schema);
- if (type != null) {
- statement.setNull(index, type);
- } else {
- statement.setObject(index, null);
- }
- } else {
- boolean bound = maybeBindLogical(statement, index, schema, value, null);
- if (!bound) {
- bound = maybeBindPrimitive(statement, index, schema, value);
- }
- if (!bound) {
- throw new io.openmessaging.connector.api.errors.ConnectException("Unsupported source data type: " + schema.getFieldType());
- }
- }
- }
-
- protected boolean maybeBindLogical(
- PreparedStatement statement,
- int index,
- Schema schema,
- Object value,
- ColumnDefinition colDef
- ) throws SQLException {
- if (schema.getName() != null) {
- switch (schema.getName()) {
- case DateColumnParser.LOGICAL_NAME:
- java.sql.Date date;
- if(value instanceof java.util.Date){
- date = new java.sql.Date(((java.util.Date) value).getTime());
- }else{
- date = new java.sql.Date((int)value);
- }
- statement.setDate(
- index, date,
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
- case TimeColumnParser.LOGICAL_NAME:
- java.sql.Time time;
- if(value instanceof java.util.Date){
- time = new java.sql.Time(((java.util.Date) value).getTime());
- }else{
- time = new java.sql.Time((int)value);
- }
- statement.setTime(
- index, time,
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
- case TimestampColumnParser.LOGICAL_NAME:
- java.sql.Timestamp timestamp;
- if(value instanceof java.util.Date){
- timestamp = new java.sql.Timestamp(((java.util.Date) value).getTime());
- }else{
- timestamp = new java.sql.Timestamp((long)value);
- }
- statement.setTimestamp(
- index, timestamp,
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
- default:
- return false;
- }
- }
- return false;
- }
-
-
- @Override
- public TimestampIncrementingCriteria criteriaFor(ColumnId incrementingColumn, List<ColumnId> timestampColumns) {
- return new TimestampIncrementingCriteria(
- incrementingColumn, timestampColumns, timeZone);
- }
-
- @Override
- public Long getMinTimestampValue(Connection con, String tableOrQuery, List<String> timestampColumns) throws SQLException {
- if(timestampColumns==null || timestampColumns.isEmpty()){
- return null;
- }
- StringBuilder builder = new StringBuilder();
- builder.append("SELECT ");
- boolean appendComma = false;
- for(String column : timestampColumns){
- builder.append("MIN(");
- builder.append(column);
- builder.append(")");
- if(appendComma){
- builder.append(",");
- }else{
- appendComma = true;
- }
- }
- builder.append(" FROM ");
- builder.append(tableOrQuery);
- String querySql = builder.toString();
- PreparedStatement st = con.prepareStatement(querySql);
- ResultSet resultSet = st.executeQuery();
- ResultSetMetaData metaData = resultSet.getMetaData();
- long minTimestampValue = Long.MAX_VALUE;
- for(int i = 1; i <= metaData.getColumnCount(); ++i){
- long t = resultSet.getLong(i);
- minTimestampValue = Math.min(minTimestampValue, t);
- }
- st.close();
- return minTimestampValue;
- }
-
- /**
- * Dialects not supporting `setObject(index, null)` can override this method
- * to provide a specific sqlType, as per the JDBC documentation
- * https://docs.oracle.com/javase/7/docs/api/java/sql/PreparedStatement.html
- *
- * @param schema the schema
- * @return the SQL type
- */
- protected Integer getSqlTypeForSchema(Schema schema) {
- return null;
- }
-
- protected boolean maybeBindPrimitive(
- PreparedStatement statement,
- int index,
- Schema schema,
- Object value
- ) throws SQLException {
- switch (schema.getFieldType()) {
- case INT8:
- statement.setByte(index, (Byte) value);
- break;
- case INT32:
- statement.setInt(index, (Integer) value);
- break;
- case INT64:
- statement.setLong(index, (Long) value);
- break;
- case FLOAT32:
- statement.setFloat(index, (Float) value);
- break;
- case FLOAT64:
- statement.setDouble(index, (Double) value);
- break;
- case BOOLEAN:
- statement.setBoolean(index, (Boolean) value);
- break;
- case STRING:
- statement.setString(index, (String) value);
- break;
- case BYTES:
- final byte[] bytes;
- if (value instanceof ByteBuffer) {
- final ByteBuffer buffer = ((ByteBuffer) value).slice();
- bytes = new byte[buffer.remaining()];
- buffer.get(bytes);
+ builder.appendList().delimitedBy(",").transformedBy(transform).of(fields);
+ }
+
+ protected void writeColumnSpec(
+ ExpressionBuilder builder,
+ SinkRecordField f
+ ) {
+ builder.appendColumnName(f.name());
+ builder.append(" ");
+ String sqlType = getSqlType(f);
+ builder.append(sqlType);
+ if (f.defaultValue() != null) {
+ builder.append(" DEFAULT ");
+ formatColumnValue(
+ builder,
+ f.schemaType(),
+ f.defaultValue()
+ );
+ } else if (isColumnOptional(f)) {
+ builder.append(" NULL");
} else {
- bytes = (byte[]) value;
+ builder.append(" NOT NULL");
}
- statement.setBytes(index, bytes);
- break;
- case DATETIME:
- java.sql.Date date;
- if(value instanceof java.util.Date){
- date = new java.sql.Date(((java.util.Date) value).getTime());
- }else{
- date = new java.sql.Date((int)value);
+ }
+
+ protected boolean isColumnOptional(SinkRecordField field) {
+ return field.isOptional();
+ }
+
+ protected void formatColumnValue(
+ ExpressionBuilder builder,
+ FieldType type,
+ Object value) {
+ switch (type) {
+ case INT8:
+ case INT32:
+ case INT64:
+ case FLOAT32:
+ case FLOAT64:
+ // no escaping required
+ builder.append(value);
+ break;
+ case BOOLEAN:
+ // 1 & 0 for boolean is more portable rather than TRUE/FALSE
+ builder.append((Boolean) value ? '1' : '0');
+ break;
+ case STRING:
+ builder.appendStringQuoted(value);
+ break;
+ case BYTES:
+ final byte[] bytes;
+ if (value instanceof ByteBuffer) {
+ final ByteBuffer buffer = ((ByteBuffer) value).slice();
+ bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ } else {
+ bytes = (byte[]) value;
+ }
+ builder.appendBinaryLiteral(bytes);
+ break;
+ default:
+ throw new io.openmessaging.connector.api.errors.ConnectException("Unsupported type for column value: " + type);
}
- statement.setDate(
- index, date,
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
+ }
+ protected String getSqlType(SinkRecordField f) {
+ throw new io.openmessaging.connector.api.errors.ConnectException(String.format(
+ "%s (%s) type doesn't have a mapping to the SQL database column type", f.schemaName(),
+ f.schemaType()
+ ));
+ }
- break;
- default:
- return false;
+ /**
+ * @param url
+ * @return
+ */
+ protected String sanitizedUrl(String url) {
+ // Only replace standard URL-type properties ...
+ return url.replaceAll("(?i)([?&]([^=&]*)password([^=&]*)=)[^&]*", "$1****");
+ }
+
+ @Override
+ public String identifier() {
+ return name() + " database " + sanitizedUrl(jdbcUrl);
+ }
+
+ @Override
+ public String toString() {
+ return name();
}
- return true;
- }
-
- @Override
- public String buildCreateTableStatement(
- TableId table,
- Collection<SinkRecordField> fields) {
- ExpressionBuilder builder = expressionBuilder();
- final List<String> pkFieldNames = extractPrimaryKeyFieldNames(fields);
- builder.append("CREATE TABLE ");
- builder.append(table);
- builder.append(" (");
- writeColumnsSpec(builder, fields);
- if (!pkFieldNames.isEmpty()) {
- builder.append(",");
- builder.append(System.lineSeparator());
- builder.append("PRIMARY KEY(");
- builder.appendList()
- .delimitedBy(",")
- .transformedBy(ExpressionBuilder.quote())
- .of(pkFieldNames);
- builder.append(")");
- }
- builder.append(")");
- return builder.toString();
- }
-
- @Override
- public String buildDropTableStatement(
- TableId table,
- DropOptions options
- ) {
- ExpressionBuilder builder = expressionBuilder();
-
- builder.append("DROP TABLE ");
- builder.append(table);
- if (options.ifExists()) {
- builder.append(" IF EXISTS");
- }
- if (options.cascade()) {
- builder.append(" CASCADE");
- }
- return builder.toString();
- }
-
- @Override
- public List<String> buildAlterTable(
- TableId table,
- Collection<SinkRecordField> fields
- ) {
- final boolean newlines = fields.size() > 1;
-
- final ExpressionBuilder.Transform<SinkRecordField> transform = (builder, field) -> {
- if (newlines) {
- builder.appendNewLine();
- }
- builder.append("ADD ");
- writeColumnSpec(builder, field);
- };
-
- ExpressionBuilder builder = expressionBuilder();
- builder.append("ALTER TABLE ");
- builder.append(table);
- builder.append(" ");
- builder.appendList()
- .delimitedBy(",")
- .transformedBy(transform)
- .of(fields);
- return Collections.singletonList(builder.toString());
- }
-
- @Override
- public void validateColumnTypes(
- ResultSetMetaData rsMetadata,
- List<ColumnId> columns
- ) throws io.openmessaging.connector.api.errors.ConnectException { }
-
- protected List<String> extractPrimaryKeyFieldNames(Collection<SinkRecordField> fields) {
- final List<String> pks = new ArrayList<>();
- for (SinkRecordField f : fields) {
- // todo add check column info
- if (f.isPrimaryKey()) {
- pks.add(f.name());
- }
- }
- return pks;
- }
-
- protected void writeColumnsSpec(
- ExpressionBuilder builder,
- Collection<SinkRecordField> fields
- ) {
- ExpressionBuilder.Transform<SinkRecordField> transform = (b, field) -> {
- b.append(System.lineSeparator());
- writeColumnSpec(b, field);
- };
- builder.appendList().delimitedBy(",").transformedBy(transform).of(fields);
- }
-
- protected void writeColumnSpec(
- ExpressionBuilder builder,
- SinkRecordField f
- ) {
- builder.appendColumnName(f.name());
- builder.append(" ");
- String sqlType = getSqlType(f);
- builder.append(sqlType);
- if (f.defaultValue() != null) {
- builder.append(" DEFAULT ");
- formatColumnValue(
- builder,
- f.schemaType(),
- f.defaultValue()
- );
- } else if (isColumnOptional(f)) {
- builder.append(" NULL");
- } else {
- builder.append(" NOT NULL");
- }
- }
-
- protected boolean isColumnOptional(SinkRecordField field) {
- return field.isOptional();
- }
-
- protected void formatColumnValue(
- ExpressionBuilder builder,
- FieldType type,
- Object value) {
- switch (type) {
- case INT8:
- case INT32:
- case INT64:
- case FLOAT32:
- case FLOAT64:
- // no escaping required
- builder.append(value);
- break;
- case BOOLEAN:
- // 1 & 0 for boolean is more portable rather than TRUE/FALSE
- builder.append((Boolean) value ? '1' : '0');
- break;
- case STRING:
- builder.appendStringQuoted(value);
- break;
- case BYTES:
- final byte[] bytes;
- if (value instanceof ByteBuffer) {
- final ByteBuffer buffer = ((ByteBuffer) value).slice();
- bytes = new byte[buffer.remaining()];
- buffer.get(bytes);
- } else {
- bytes = (byte[]) value;
- }
- builder.appendBinaryLiteral(bytes);
- break;
- default:
- throw new io.openmessaging.connector.api.errors.ConnectException("Unsupported type for column value: " + type);
- }
- }
-
- protected String getSqlType(SinkRecordField f) {
- throw new io.openmessaging.connector.api.errors.ConnectException(String.format(
- "%s (%s) type doesn't have a mapping to the SQL database column type", f.schemaName(),
- f.schemaType()
- ));
- }
-
-
- /**
- * 隐藏密码
- * @param url
- * @return
- */
- protected String sanitizedUrl(String url) {
- // Only replace standard URL-type properties ...
- return url.replaceAll("(?i)([?&]([^=&]*)password([^=&]*)=)[^&]*", "$1****");
- }
-
- @Override
- public String identifier() {
- return name() + " database " + sanitizedUrl(jdbcUrl);
- }
-
- @Override
- public String toString() {
- return name();
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/MySqlDatabaseDialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/MySqlDatabaseDialect.java
index b6bd1dc..0eb8724 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/MySqlDatabaseDialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/MySqlDatabaseDialect.java
@@ -38,112 +38,114 @@ import java.util.Collection;
*/
public class MySqlDatabaseDialect extends GenericDatabaseDialect {
- private final Logger log = LoggerFactory.getLogger(MySqlDatabaseDialect.class);
+ private final Logger log = LoggerFactory.getLogger(MySqlDatabaseDialect.class);
- /**
- * The provider for {@link MySqlDatabaseDialect}.
- */
- public static class Provider extends DatabaseDialectProvider {
- @SneakyThrows
- public Provider() {
- super(MySqlDatabaseDialect.class.getSimpleName(), "mysql");
- Class.forName("com.mysql.cj.jdbc.Driver");
- }
+ /**
+ * The provider for {@link MySqlDatabaseDialect}.
+ */
+ public static class Provider extends DatabaseDialectProvider {
+ @SneakyThrows
+ public Provider() {
+ super(MySqlDatabaseDialect.class.getSimpleName(), "mysql");
+ Class.forName("com.mysql.cj.jdbc.Driver");
+ }
- @Override
- public DatabaseDialect create(AbstractConfig config) {
- return new MySqlDatabaseDialect(config);
+ @Override
+ public DatabaseDialect create(AbstractConfig config) {
+ return new MySqlDatabaseDialect(config);
+ }
}
- }
- /**
- * Create a new dialect instance with the given connector configuration.
- *
- * @param config the connector configuration; may not be null
- */
- public MySqlDatabaseDialect(AbstractConfig config) {
- super(config, new IdentifierRules(".", "`", "`"));
- }
+ /**
+ * Create a new dialect instance with the given connector configuration.
+ *
+ * @param config the connector configuration; may not be null
+ */
+ public MySqlDatabaseDialect(AbstractConfig config) {
+ super(config, new IdentifierRules(".", "`", "`"));
+ }
- /**
- * initialize prepared statement
- * @param stmt
- * @throws SQLException
- */
- @Override
- protected void initializePreparedStatement(PreparedStatement stmt) throws SQLException {
- stmt.setFetchSize(Integer.MIN_VALUE);
- log.trace("Initializing PreparedStatement fetch direction to FETCH_FORWARD for '{}'", stmt);
- stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
- }
+ /**
+ * initialize prepared statement
+ *
+ * @param stmt
+ * @throws SQLException
+ */
+ @Override
+ protected void initializePreparedStatement(PreparedStatement stmt) throws SQLException {
+ stmt.setFetchSize(Integer.MIN_VALUE);
+ log.trace("Initializing PreparedStatement fetch direction to FETCH_FORWARD for '{}'", stmt);
+ stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
+ }
- /**
- * get sql type
- * @param field
- * @return
- */
- @Override
- protected String getSqlType(SinkRecordField field) {
- switch (field.schemaType()) {
- case INT8:
- return "TINYINT";
- case INT32:
- return "INT";
- case INT64:
- return "BIGINT";
- case FLOAT32:
- return "FLOAT";
- case FLOAT64:
- return "DOUBLE";
- case BOOLEAN:
- return "TINYINT";
- case STRING:
- return "TEXT";
- case BYTES:
- return "VARBINARY(1024)";
- default:
- return super.getSqlType(field);
+ /**
+ * get sql type
+ *
+ * @param field
+ * @return
+ */
+ @Override
+ protected String getSqlType(SinkRecordField field) {
+ switch (field.schemaType()) {
+ case INT8:
+ return "TINYINT";
+ case INT32:
+ return "INT";
+ case INT64:
+ return "BIGINT";
+ case FLOAT32:
+ return "FLOAT";
+ case FLOAT64:
+ return "DOUBLE";
+ case BOOLEAN:
+ return "TINYINT";
+ case STRING:
+ return "TEXT";
+ case BYTES:
+ return "VARBINARY(1024)";
+ default:
+ return super.getSqlType(field);
+ }
}
- }
- @Override
- public String buildUpsertQueryStatement(
- TableId table,
- Collection<ColumnId> keyColumns,
- Collection<ColumnId> nonKeyColumns
- ) {
- //MySql doesn't support SQL 2003:merge so here how the upsert is handled
- final ExpressionBuilder.Transform<ColumnId> transform = (builder, col) -> {
- builder.appendColumnName(col.name());
- builder.append("=values(");
- builder.appendColumnName(col.name());
- builder.append(")");
- };
+ @Override
+ public String buildUpsertQueryStatement(
+ TableId table,
+ Collection<ColumnId> keyColumns,
+ Collection<ColumnId> nonKeyColumns
+ ) {
+ //MySql doesn't support SQL 2003:merge so here how the upsert is handled
+ final ExpressionBuilder.Transform<ColumnId> transform = (builder, col) -> {
+ builder.appendColumnName(col.name());
+ builder.append("=values(");
+ builder.appendColumnName(col.name());
+ builder.append(")");
+ };
- ExpressionBuilder builder = expressionBuilder();
- builder.append("insert into ");
- builder.append(table);
- builder.append("(");
- builder.appendList()
- .delimitedBy(",")
- .transformedBy(ExpressionBuilder.columnNames())
- .of(keyColumns, nonKeyColumns);
- builder.append(") values(");
- builder.appendMultiple(",", "?", keyColumns.size() + nonKeyColumns.size());
- builder.append(") on duplicate key update ");
- builder.appendList()
- .delimitedBy(",")
- .transformedBy(transform)
- .of(nonKeyColumns.isEmpty() ? keyColumns : nonKeyColumns);
- return builder.toString();
- }
+ ExpressionBuilder builder = expressionBuilder();
+ builder.append("insert into ");
+ builder.append(table);
+ builder.append("(");
+ builder.appendList()
+ .delimitedBy(",")
+ .transformedBy(ExpressionBuilder.columnNames())
+ .of(keyColumns, nonKeyColumns);
+ builder.append(") values(");
+ builder.appendMultiple(",", "?", keyColumns.size() + nonKeyColumns.size());
+ builder.append(") on duplicate key update ");
+ builder.appendList()
+ .delimitedBy(",")
+ .transformedBy(transform)
+ .of(nonKeyColumns.isEmpty() ? keyColumns : nonKeyColumns);
+ return builder.toString();
+ }
- @Override
- protected String sanitizedUrl(String url) {
- // MySQL can also have "username:password@" at the beginning of the host list and
- // in parenthetical properties
- return super.sanitizedUrl(url)
+ @Override
+ protected String sanitizedUrl(String url) {
+ // MySQL can also have "username:password@" at the beginning of the host list and
+ // in parenthetical properties
+ return super.sanitizedUrl(url)
.replaceAll("(?i)([(,]password=)[^,)]*", "$1****")
.replaceAll("(://[^:]*:)([^@]*)@", "$1****@");
- }
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/OpenMLDBDatabaseDialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/OpenMLDBDatabaseDialect.java
index daf4ace..0c70626 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/OpenMLDBDatabaseDialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/OpenMLDBDatabaseDialect.java
@@ -30,39 +30,40 @@ import org.slf4j.LoggerFactory;
*/
public class OpenMLDBDatabaseDialect extends GenericDatabaseDialect {
- private final Logger log = LoggerFactory.getLogger(OpenMLDBDatabaseDialect.class);
+ private final Logger log = LoggerFactory.getLogger(OpenMLDBDatabaseDialect.class);
- /**
- * The provider for {@link OpenMLDBDatabaseDialect}.
- */
- public static class Provider extends DatabaseDialectProvider {
- @SneakyThrows
- public Provider() {
- super(OpenMLDBDatabaseDialect.class.getSimpleName(), "openmldb");
- Class.forName("com._4paradigm.openmldb.jdbc.SQLDriver");
- }
+ /**
+ * The provider for {@link OpenMLDBDatabaseDialect}.
+ */
+ public static class Provider extends DatabaseDialectProvider {
+ @SneakyThrows
+ public Provider() {
+ super(OpenMLDBDatabaseDialect.class.getSimpleName(), "openmldb");
+ Class.forName("com._4paradigm.openmldb.jdbc.SQLDriver");
+ }
- @Override
- public DatabaseDialect create(AbstractConfig config) {
- return new OpenMLDBDatabaseDialect(config);
+ @Override
+ public DatabaseDialect create(AbstractConfig config) {
+ return new OpenMLDBDatabaseDialect(config);
+ }
}
- }
- /**
- * create openMLDB database dialect
- * @param config
- */
- public OpenMLDBDatabaseDialect(AbstractConfig config) {
- super(config, new IdentifierRules(".", "`", "`"));
- }
+ /**
+ * create openMLDB database dialect
+ *
+ * @param config
+ */
+ public OpenMLDBDatabaseDialect(AbstractConfig config) {
+ super(config, new IdentifierRules(".", "`", "`"));
+ }
- @Override
- protected String sanitizedUrl(String url) {
- // MySQL can also have "username:password@" at the beginning of the host list and
- // in parenthetical properties
- return super.sanitizedUrl(url)
+ @Override
+ protected String sanitizedUrl(String url) {
+ // MySQL can also have "username:password@" at the beginning of the host list and
+ // in parenthetical properties
+ return super.sanitizedUrl(url)
.replaceAll("(?i)([(,]password=)[^,)]*", "$1****")
.replaceAll("(://[^:]*:)([^@]*)@", "$1****@");
- }
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/CachedConnectionProvider.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/CachedConnectionProvider.java
index 4286a04..8607223 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/CachedConnectionProvider.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/CachedConnectionProvider.java
@@ -28,101 +28,101 @@ import java.sql.SQLException;
*/
public class CachedConnectionProvider implements ConnectionProvider {
- private static final Logger log = LoggerFactory.getLogger(CachedConnectionProvider.class);
+ private static final Logger log = LoggerFactory.getLogger(CachedConnectionProvider.class);
- private static final int VALIDITY_CHECK_TIMEOUT_S = 5;
+ private static final int VALIDITY_CHECK_TIMEOUT_S = 5;
- private final ConnectionProvider provider;
- private final int maxConnectionAttempts;
- private final long connectionRetryBackoff;
+ private final ConnectionProvider provider;
+ private final int maxConnectionAttempts;
+ private final long connectionRetryBackoff;
- private int count = 0;
- private Connection connection;
+ private int count = 0;
+ private Connection connection;
- public CachedConnectionProvider(
- ConnectionProvider provider,
- int maxConnectionAttempts,
- long connectionRetryBackoff
- ) {
- this.provider = provider;
- this.maxConnectionAttempts = maxConnectionAttempts;
- this.connectionRetryBackoff = connectionRetryBackoff;
- }
+ public CachedConnectionProvider(
+ ConnectionProvider provider,
+ int maxConnectionAttempts,
+ long connectionRetryBackoff
+ ) {
+ this.provider = provider;
+ this.maxConnectionAttempts = maxConnectionAttempts;
+ this.connectionRetryBackoff = connectionRetryBackoff;
+ }
- @Override
- public synchronized Connection getConnection() {
- try {
- if (connection == null) {
- newConnection();
- } else if (!isConnectionValid(connection, VALIDITY_CHECK_TIMEOUT_S)) {
- log.info("The database connection is invalid. Reconnecting...");
- close();
- newConnection();
- }
- } catch (SQLException sqle) {
- throw new ConnectException(sqle);
+ @Override
+ public synchronized Connection getConnection() {
+ try {
+ if (connection == null) {
+ newConnection();
+ } else if (!isConnectionValid(connection, VALIDITY_CHECK_TIMEOUT_S)) {
+ log.info("The database connection is invalid. Reconnecting...");
+ close();
+ newConnection();
+ }
+ } catch (SQLException sqle) {
+ throw new ConnectException(sqle);
+ }
+ return connection;
}
- return connection;
- }
- @Override
- public boolean isConnectionValid(Connection connection, int timeout) {
- try {
- return provider.isConnectionValid(connection, timeout);
- } catch (SQLException sqle) {
- log.debug("Unable to check if the underlying connection is valid", sqle);
- return false;
+ @Override
+ public boolean isConnectionValid(Connection connection, int timeout) {
+ try {
+ return provider.isConnectionValid(connection, timeout);
+ } catch (SQLException sqle) {
+ log.debug("Unable to check if the underlying connection is valid", sqle);
+ return false;
+ }
}
- }
- private void newConnection() throws SQLException {
- int attempts = 0;
- while (attempts < maxConnectionAttempts) {
- try {
- ++count;
- log.info("Attempting to open connection #{} to {}", count, provider);
- connection = provider.getConnection();
- onConnect(connection);
- return;
- } catch (SQLException sqle) {
- attempts++;
- if (attempts < maxConnectionAttempts) {
- log.info("Unable to connect to database on attempt {}/{}. Will retry in {} ms.", attempts,
- maxConnectionAttempts, connectionRetryBackoff, sqle
- );
- try {
- Thread.sleep(connectionRetryBackoff);
- } catch (InterruptedException e) {
- // this is ok because just woke up early
- }
- } else {
- throw sqle;
+ private void newConnection() throws SQLException {
+ int attempts = 0;
+ while (attempts < maxConnectionAttempts) {
+ try {
+ ++count;
+ log.info("Attempting to open connection #{} to {}", count, provider);
+ connection = provider.getConnection();
+ onConnect(connection);
+ return;
+ } catch (SQLException sqle) {
+ attempts++;
+ if (attempts < maxConnectionAttempts) {
+ log.info("Unable to connect to database on attempt {}/{}. Will retry in {} ms.", attempts,
+ maxConnectionAttempts, connectionRetryBackoff, sqle
+ );
+ try {
+ Thread.sleep(connectionRetryBackoff);
+ } catch (InterruptedException e) {
+ // this is ok because just woke up early
+ }
+ } else {
+ throw sqle;
+ }
+ }
}
- }
}
- }
- @Override
- public synchronized void close() {
- if (connection != null) {
- try {
- log.info("Closing connection #{} to {}", count, provider);
- connection.close();
- } catch (SQLException sqle) {
- log.warn("Ignoring error closing connection", sqle);
- } finally {
- connection = null;
- provider.close();
- }
+ @Override
+ public synchronized void close() {
+ if (connection != null) {
+ try {
+ log.info("Closing connection #{} to {}", count, provider);
+ connection.close();
+ } catch (SQLException sqle) {
+ log.warn("Ignoring error closing connection", sqle);
+ } finally {
+ connection = null;
+ provider.close();
+ }
+ }
}
- }
- @Override
- public String identifier() {
- return provider.identifier();
- }
+ @Override
+ public String identifier() {
+ return provider.identifier();
+ }
- protected void onConnect(Connection connection) throws SQLException {
- }
+ protected void onConnect(Connection connection) throws SQLException {
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/ConnectionProvider.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/ConnectionProvider.java
index 3c5f1d3..6e6c7a7 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/ConnectionProvider.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/ConnectionProvider.java
@@ -24,42 +24,43 @@ import java.sql.SQLException;
*/
public interface ConnectionProvider extends AutoCloseable {
- /**
- * Create a connection.
- * @return the connection; never null
- * @throws SQLException if there is a problem getting the connection
- */
- Connection getConnection() throws SQLException;
+ /**
+ * Create a connection.
+ *
+ * @return the connection; never null
+ * @throws SQLException if there is a problem getting the connection
+ */
+ Connection getConnection() throws SQLException;
- /**
- * Determine if the specified connection is valid.
- *
- * @param connection the database connection; may not be null
- * @param timeout The time in seconds to wait for the database operation used to validate
- * the connection to complete. If the timeout period expires before the
- * operation completes, this method returns false. A value of 0 indicates a
- * timeout is not applied to the database operation.
- * @return true if it is valid, or false otherwise
- * @throws SQLException if there is an error with the database connection
- */
- boolean isConnectionValid(
- Connection connection,
- int timeout
- ) throws SQLException;
+ /**
+ * Determine if the specified connection is valid.
+ *
+ * @param connection the database connection; may not be null
+ * @param timeout The time in seconds to wait for the database operation used to validate
+ * the connection to complete. If the timeout period expires before the
+ * operation completes, this method returns false. A value of 0 indicates a
+ * timeout is not applied to the database operation.
+ * @return true if it is valid, or false otherwise
+ * @throws SQLException if there is an error with the database connection
+ */
+ boolean isConnectionValid(
+ Connection connection,
+ int timeout
+ ) throws SQLException;
- /**
- * Close this connection provider.
- */
- @Override
- void close();
+ /**
+ * Close this connection provider.
+ */
+ @Override
+ void close();
- /**
- * Get the publicly viewable identifier for this connection provider and / or the database.
- * The resulting value should not contain any secrets or passwords.
- *
- * @return the identifier; never null
- */
- default String identifier() {
- return toString();
- }
+ /**
+ * Get the publicly viewable identifier for this connection provider and / or the database.
+ * The resulting value should not contain any secrets or passwords.
+ *
+ * @return the identifier; never null
+ */
+ default String identifier() {
+ return toString();
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/DatabaseDialectProvider.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/DatabaseDialectProvider.java
index d72da5d..be91bad 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/DatabaseDialectProvider.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/DatabaseDialectProvider.java
@@ -26,33 +26,35 @@ import org.apache.rocketmq.connect.jdbc.dialect.DatabaseDialect;
public abstract class DatabaseDialectProvider {
- private final String name;
- private final String protocol;
-
- protected DatabaseDialectProvider(String name,String protocol) {
- assert name != null;
- this.name = name;
- this.protocol = protocol;
- }
- /**
- * create database dialect
- * @param config
- * @return
- */
- public abstract DatabaseDialect create(AbstractConfig config);
-
- public String protocolName() {
- return protocol;
- }
-
- public String dialectName() {
- return name;
- }
-
- @Override
- public String toString() {
- return dialectName();
- }
+ private final String name;
+ private final String protocol;
+
+ protected DatabaseDialectProvider(String name, String protocol) {
+ assert name != null;
+ this.name = name;
+ this.protocol = protocol;
+ }
+
+ /**
+ * create database dialect
+ *
+ * @param config
+ * @return
+ */
+ public abstract DatabaseDialect create(AbstractConfig config);
+
+ public String protocolName() {
+ return protocol;
+ }
+
+ public String dialectName() {
+ return name;
+ }
+
+ @Override
+ public String toString() {
+ return dialectName();
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/JdbcUrlInfo.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/JdbcUrlInfo.java
index 82d4136..4e182c7 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/JdbcUrlInfo.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/provider/JdbcUrlInfo.java
@@ -39,4 +39,4 @@ public interface JdbcUrlInfo {
* @return the URL.
*/
String url();
- }
+}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/ConfigException.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/ConfigException.java
index 1c181a4..4d4ebfd 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/ConfigException.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/ConfigException.java
@@ -28,7 +28,7 @@ public class ConfigException extends RuntimeException {
}
public ConfigException(String name, Object value) {
- this(name, value, (String)null);
+ this(name, value, null);
}
public ConfigException(String name, Object value, String message) {
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/TableAlterOrCreateException.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/TableAlterOrCreateException.java
index e32f26d..97c27d1 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/TableAlterOrCreateException.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/exception/TableAlterOrCreateException.java
@@ -23,12 +23,12 @@ import io.openmessaging.connector.api.errors.ConnectException;
*/
public class TableAlterOrCreateException extends ConnectException {
- public TableAlterOrCreateException(String reason) {
- super(reason);
- }
+ public TableAlterOrCreateException(String reason) {
+ super(reason);
+ }
- public TableAlterOrCreateException(String reason, Throwable throwable) {
- super(reason, throwable);
- }
+ public TableAlterOrCreateException(String reason, Throwable throwable) {
+ super(reason, throwable);
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefAdjuster.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefAdjuster.java
index 4f589c0..5e8b6c3 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefAdjuster.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefAdjuster.java
@@ -33,10 +33,10 @@ public class ColumnDefAdjuster {
String catalogPattern,
String schemaPattern,
String tablePattern,
- String columnPattern){
+ String columnPattern) {
ColumnDefAdjuster adjuster = new ColumnDefAdjuster();
- try(ResultSet rs = conn.getMetaData().getColumns(
- catalogPattern, schemaPattern, tablePattern, columnPattern)){
+ try (ResultSet rs = conn.getMetaData().getColumns(
+ catalogPattern, schemaPattern, tablePattern, columnPattern)) {
final int rsColumnCount = rs.getMetaData().getColumnCount();
while (rs.next()) {
final String columnName = rs.getString(4);
@@ -56,15 +56,15 @@ public class ColumnDefAdjuster {
}
adjuster.nullable.put(columnName, nullability);
}
- }catch (SQLException e){
+ } catch (SQLException e) {
//pass
}
return adjuster;
}
- public ColumnDefinition.Nullability nullable(String columnName){
- if(nullable == null || !nullable.containsKey(columnName)){
+ public ColumnDefinition.Nullability nullable(String columnName) {
+ if (nullable == null || !nullable.containsKey(columnName)) {
return null;
}
return nullable.get(columnName);
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefinition.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefinition.java
index 25ff655..fceadef 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefinition.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnDefinition.java
@@ -26,314 +26,314 @@ import java.util.Objects;
*/
public class ColumnDefinition {
- /**
- * The nullability of a column.
- */
- public enum Nullability {
- NULL, NOT_NULL, UNKNOWN
- }
-
- /**
- * The mutability of a column.
- */
- public enum Mutability {
- READ_ONLY, MAYBE_WRITABLE, WRITABLE, UNKNOWN
- }
+ /**
+ * The nullability of a column.
+ */
+ public enum Nullability {
+ NULL, NOT_NULL, UNKNOWN
+ }
- private final ColumnId id;
- private final String typeName;
- private final int jdbcType;
- private final int displaySize;
- private final int precision;
- private final int scale;
- private final boolean autoIncremented;
- private final boolean caseSensitive;
- private final boolean searchable;
- private final boolean currency;
- private final boolean signedNumbers;
- private final boolean isPrimaryKey;
- private final Nullability nullability;
- private final Mutability mutability;
- private final String classNameForType;
+ /**
+ * The mutability of a column.
+ */
+ public enum Mutability {
+ READ_ONLY, MAYBE_WRITABLE, WRITABLE, UNKNOWN
+ }
- public ColumnDefinition(
- ColumnId id,
- int jdbcType,
- String typeName,
- String classNameForType,
- Nullability nullability,
- Mutability mutability,
- int precision,
- int scale,
- boolean signedNumbers,
- int displaySize,
- boolean autoIncremented,
- boolean caseSensitive,
- boolean searchable,
- boolean currency,
- boolean isPrimaryKey
- ) {
- this.id = id;
- this.typeName = typeName;
- this.jdbcType = jdbcType;
- this.displaySize = displaySize;
- this.precision = precision;
- this.scale = scale;
- this.autoIncremented = autoIncremented;
- this.caseSensitive = caseSensitive;
- this.searchable = searchable;
- this.currency = currency;
- this.signedNumbers = signedNumbers;
- this.nullability = nullability != null ? nullability : Nullability.UNKNOWN;
- this.mutability = mutability != null ? mutability : Mutability.MAYBE_WRITABLE;
- this.classNameForType = classNameForType;
- this.isPrimaryKey = isPrimaryKey;
- }
+ private final ColumnId id;
+ private final String typeName;
+ private final int jdbcType;
+ private final int displaySize;
+ private final int precision;
+ private final int scale;
+ private final boolean autoIncremented;
+ private final boolean caseSensitive;
+ private final boolean searchable;
+ private final boolean currency;
+ private final boolean signedNumbers;
+ private final boolean isPrimaryKey;
+ private final Nullability nullability;
+ private final Mutability mutability;
+ private final String classNameForType;
+ public ColumnDefinition(
+ ColumnId id,
+ int jdbcType,
+ String typeName,
+ String classNameForType,
+ Nullability nullability,
+ Mutability mutability,
+ int precision,
+ int scale,
+ boolean signedNumbers,
+ int displaySize,
+ boolean autoIncremented,
+ boolean caseSensitive,
+ boolean searchable,
+ boolean currency,
+ boolean isPrimaryKey
+ ) {
+ this.id = id;
+ this.typeName = typeName;
+ this.jdbcType = jdbcType;
+ this.displaySize = displaySize;
+ this.precision = precision;
+ this.scale = scale;
+ this.autoIncremented = autoIncremented;
+ this.caseSensitive = caseSensitive;
+ this.searchable = searchable;
+ this.currency = currency;
+ this.signedNumbers = signedNumbers;
+ this.nullability = nullability != null ? nullability : Nullability.UNKNOWN;
+ this.mutability = mutability != null ? mutability : Mutability.MAYBE_WRITABLE;
+ this.classNameForType = classNameForType;
+ this.isPrimaryKey = isPrimaryKey;
+ }
- /**
- * Indicates whether the column is automatically numbered.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isAutoIncrement() {
- return autoIncremented;
- }
- /**
- * Indicates whether the column's case matters.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isCaseSensitive() {
- return caseSensitive;
- }
+ /**
+ * Indicates whether the column is automatically numbered.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isAutoIncrement() {
+ return autoIncremented;
+ }
- /**
- * Indicates whether the column can be used in a where clause.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isSearchable() {
- return searchable;
- }
+ /**
+ * Indicates whether the column's case matters.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isCaseSensitive() {
+ return caseSensitive;
+ }
- /**
- * Indicates whether the column is a cash value.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isCurrency() {
- return currency;
- }
+ /**
+ * Indicates whether the column can be used in a where clause.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isSearchable() {
+ return searchable;
+ }
- /**
- * Indicates whether the column is part of the table's primary key.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isPrimaryKey() {
- return isPrimaryKey;
- }
+ /**
+ * Indicates whether the column is a cash value.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isCurrency() {
+ return currency;
+ }
- /**
- * Indicates the nullability of values in the column.
- *
- * @return the nullability status of the given column; never null
- */
- public Nullability nullability() {
- return nullability;
- }
+ /**
+ * Indicates whether the column is part of the table's primary key.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isPrimaryKey() {
+ return isPrimaryKey;
+ }
- /**
- * Indicates whether values in the column are optional. This is equivalent to calling:
- * <pre>
- * nullability() == Nullability.NULL || nullability() == Nullability.UNKNOWN
- * </pre>
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isOptional() {
- return nullability == Nullability.NULL || nullability == Nullability.UNKNOWN;
- }
+ /**
+ * Indicates the nullability of values in the column.
+ *
+ * @return the nullability status of the given column; never null
+ */
+ public Nullability nullability() {
+ return nullability;
+ }
- /**
- * Indicates whether values in the column are signed numbers.
- *
- * @return <code>true</code> if so; <code>false</code> otherwise
- */
- public boolean isSignedNumber() {
- return signedNumbers;
- }
+ /**
+ * Indicates whether values in the column are optional. This is equivalent to calling:
+ * <pre>
+ * nullability() == Nullability.NULL || nullability() == Nullability.UNKNOWN
+ * </pre>
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isOptional() {
+ return nullability == Nullability.NULL || nullability == Nullability.UNKNOWN;
+ }
- /**
- * Indicates the column's normal maximum width in characters.
- *
- * @return the normal maximum number of characters allowed as the width of the designated column
- */
- public int displaySize() {
- return displaySize;
- }
+ /**
+ * Indicates whether values in the column are signed numbers.
+ *
+ * @return <code>true</code> if so; <code>false</code> otherwise
+ */
+ public boolean isSignedNumber() {
+ return signedNumbers;
+ }
- /**
- * Get the column's identifier.
- *
- * @return column identifier; never null
- */
- public ColumnId id() {
- return id;
- }
+ /**
+ * Indicates the column's normal maximum width in characters.
+ *
+ * @return the normal maximum number of characters allowed as the width of the designated column
+ */
+ public int displaySize() {
+ return displaySize;
+ }
- /**
- * Get the column's table identifier.
- *
- * @return the table identifier; never null
- */
- public TableId tableId() {
- return id.tableId();
- }
+ /**
+ * Get the column's identifier.
+ *
+ * @return column identifier; never null
+ */
+ public ColumnId id() {
+ return id;
+ }
- /**
- * Get the column's specified column size. For numeric data, this is the maximum precision. For
- * character data, this is the length in characters. For datetime datatypes, this is the length in
- * characters of the String representation (assuming the maximum allowed precision of the
- * fractional seconds component). For binary data, this is the length in bytes. For the ROWID
- * datatype, this is the length in bytes. 0 is returned for data types where the column size is
- * not applicable.
- *
- * @return precision
- */
- public int precision() {
- return precision;
- }
+ /**
+ * Get the column's table identifier.
+ *
+ * @return the table identifier; never null
+ */
+ public TableId tableId() {
+ return id.tableId();
+ }
- /**
- * Gets the column's number of digits to right of the decimal point. 0 is returned for data types
- * where the scale is not applicable.
- *
- * @return scale
- */
- public int scale() {
- return scale;
- }
+ /**
+ * Get the column's specified column size. For numeric data, this is the maximum precision. For
+ * character data, this is the length in characters. For datetime datatypes, this is the length in
+ * characters of the String representation (assuming the maximum allowed precision of the
+ * fractional seconds component). For binary data, this is the length in bytes. For the ROWID
+ * datatype, this is the length in bytes. 0 is returned for data types where the column size is
+ * not applicable.
+ *
+ * @return precision
+ */
+ public int precision() {
+ return precision;
+ }
- /**
- * Retrieves the column's JDBC type.
- *
- * @return SQL type from java.sql.Types
- * @see Types
- */
- public int type() {
- return jdbcType;
- }
+ /**
+ * Gets the column's number of digits to right of the decimal point. 0 is returned for data types
+ * where the scale is not applicable.
+ *
+ * @return scale
+ */
+ public int scale() {
+ return scale;
+ }
- /**
- * Retrieves the designated column's database-specific type name.
- *
- * @return type name used by the database. If the column type is a user-defined type, then a
- * fully-qualified type name is returned.
- */
- public String typeName() {
- return typeName;
- }
+ /**
+ * Retrieves the column's JDBC type.
+ *
+ * @return SQL type from java.sql.Types
+ * @see Types
+ */
+ public int type() {
+ return jdbcType;
+ }
- /**
- * Indicates whether the designated column is mutable.
- *
- * @return the mutability; never null
- */
- public Mutability mutability() {
- return mutability;
- }
+ /**
+ * Retrieves the designated column's database-specific type name.
+ *
+ * @return type name used by the database. If the column type is a user-defined type, then a
+ * fully-qualified type name is returned.
+ */
+ public String typeName() {
+ return typeName;
+ }
- /**
- * Returns the fully-qualified name of the Java class whose instances are manufactured if the
- * method {@link java.sql.ResultSet#getObject(int)} is called to retrieve a value from the column.
- * {@link java.sql.ResultSet#getObject(int)} may return a subclass of the class returned by this
- * method.
- *
- * @return the fully-qualified name of the class in the Java programming language that would be
- * used by the method <code>ResultSet.getObject</code> to retrieve the value in the specified
- * column. This is the class name used for custom mapping.
- */
- public String classNameForType() {
- return classNameForType;
- }
+ /**
+ * Indicates whether the designated column is mutable.
+ *
+ * @return the mutability; never null
+ */
+ public Mutability mutability() {
+ return mutability;
+ }
+ /**
+ * Returns the fully-qualified name of the Java class whose instances are manufactured if the
+ * method {@link java.sql.ResultSet#getObject(int)} is called to retrieve a value from the column.
+ * {@link java.sql.ResultSet#getObject(int)} may return a subclass of the class returned by this
+ * method.
+ *
+ * @return the fully-qualified name of the class in the Java programming language that would be
+ * used by the method <code>ResultSet.getObject</code> to retrieve the value in the specified
+ * column. This is the class name used for custom mapping.
+ */
+ public String classNameForType() {
+ return classNameForType;
+ }
- @Override
- public int hashCode() {
- return id.hashCode();
- }
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
+ @Override
+ public int hashCode() {
+ return id.hashCode();
}
- if (obj instanceof ColumnDefinition) {
- ColumnDefinition that = (ColumnDefinition) obj;
- return Objects.equals(this.id, that.id)
- && this.jdbcType == that.jdbcType
- && this.displaySize == that.displaySize
- && this.precision == that.precision
- && this.scale == that.scale
- && this.autoIncremented == that.autoIncremented
- && this.caseSensitive == that.caseSensitive
- && this.searchable == that.searchable
- && this.currency == that.currency
- && this.signedNumbers == that.signedNumbers
- && this.nullability == that.nullability
- && Objects.equals(this.typeName, that.typeName)
- && Objects.equals(this.classNameForType, that.classNameForType)
- && Objects.equals(this.isPrimaryKey, that.isPrimaryKey);
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (obj instanceof ColumnDefinition) {
+ ColumnDefinition that = (ColumnDefinition) obj;
+ return Objects.equals(this.id, that.id)
+ && this.jdbcType == that.jdbcType
+ && this.displaySize == that.displaySize
+ && this.precision == that.precision
+ && this.scale == that.scale
+ && this.autoIncremented == that.autoIncremented
+ && this.caseSensitive == that.caseSensitive
+ && this.searchable == that.searchable
+ && this.currency == that.currency
+ && this.signedNumbers == that.signedNumbers
+ && this.nullability == that.nullability
+ && Objects.equals(this.typeName, that.typeName)
+ && Objects.equals(this.classNameForType, that.classNameForType)
+ && Objects.equals(this.isPrimaryKey, that.isPrimaryKey);
+ }
+ return false;
}
- return false;
- }
- @Override
- public String toString() {
- return "Column{'" + id.name() + '\'' + ", isPrimaryKey=" + isPrimaryKey + ", allowsNull="
- + isOptional() + ", sqlType=" + typeName + '}';
- }
+ @Override
+ public String toString() {
+ return "Column{'" + id.name() + '\'' + ", isPrimaryKey=" + isPrimaryKey + ", allowsNull="
+ + isOptional() + ", sqlType=" + typeName + '}';
+ }
- /**
- * Obtain a {@link ColumnDefinition} that has all the same characteristics as this column except
- * that belongs to the specified table.
- *
- * @param tableId the new table ID; may not be null
- * @return this definition if the specified table ID matches this definition's {@link #tableId()},
- * or a new definition that is a copy of this definition except with an {@link #id() ID} that
- * uses the specified table; never null
- */
- public ColumnDefinition forTable(TableId tableId) {
- if (tableId().equals(tableId)) {
- return this;
+ /**
+ * Obtain a {@link ColumnDefinition} that has all the same characteristics as this column except
+ * that belongs to the specified table.
+ *
+ * @param tableId the new table ID; may not be null
+ * @return this definition if the specified table ID matches this definition's {@link #tableId()},
+ * or a new definition that is a copy of this definition except with an {@link #id() ID} that
+ * uses the specified table; never null
+ */
+ public ColumnDefinition forTable(TableId tableId) {
+ if (tableId().equals(tableId)) {
+ return this;
+ }
+ ColumnId newId = new ColumnId(tableId, this.id().name());
+ return new ColumnDefinition(newId, jdbcType, typeName, classNameForType, nullability,
+ mutability, precision, scale, signedNumbers, displaySize,
+ autoIncremented, caseSensitive, searchable, currency, isPrimaryKey
+ );
}
- ColumnId newId = new ColumnId(tableId, this.id().name());
- return new ColumnDefinition(newId, jdbcType, typeName, classNameForType, nullability,
- mutability, precision, scale, signedNumbers, displaySize,
- autoIncremented, caseSensitive, searchable, currency, isPrimaryKey
- );
- }
- /**
- * Obtain a {@link ColumnDefinition} that has all the same characteristics as this column except
- * that it does or does not belong to the table's primary key
- *
- * @param isPrimaryKey true if the resulting column definition should be part of the table's
- * primary key, or false otherwise
- * @return a definition that is the same as this definition except it is or is not part of the
- * tables primary key, or may be this object if {@link #isPrimaryKey()} already matches the
- * supplied parameter; never null
- */
- public ColumnDefinition asPartOfPrimaryKey(boolean isPrimaryKey) {
- if (isPrimaryKey == isPrimaryKey()) {
- return this;
+ /**
+ * Obtain a {@link ColumnDefinition} that has all the same characteristics as this column except
+ * that it does or does not belong to the table's primary key
+ *
+ * @param isPrimaryKey true if the resulting column definition should be part of the table's
+ * primary key, or false otherwise
+ * @return a definition that is the same as this definition except it is or is not part of the
+ * tables primary key, or may be this object if {@link #isPrimaryKey()} already matches the
+ * supplied parameter; never null
+ */
+ public ColumnDefinition asPartOfPrimaryKey(boolean isPrimaryKey) {
+ if (isPrimaryKey == isPrimaryKey()) {
+ return this;
+ }
+ return new ColumnDefinition(id, jdbcType, typeName, classNameForType, nullability, mutability,
+ precision, scale, signedNumbers, displaySize, autoIncremented,
+ caseSensitive, searchable, currency, isPrimaryKey
+ );
}
- return new ColumnDefinition(id, jdbcType, typeName, classNameForType, nullability, mutability,
- precision, scale, signedNumbers, displaySize, autoIncremented,
- caseSensitive, searchable, currency, isPrimaryKey
- );
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnId.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnId.java
index a237d70..8139958 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnId.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/ColumnId.java
@@ -27,79 +27,79 @@ import java.util.Objects;
*/
public class ColumnId implements ExpressionBuilder.Expressable {
- private final TableId tableId;
- private final String name;
- private final String alias;
- private final int hash;
+ private final TableId tableId;
+ private final String name;
+ private final String alias;
+ private final int hash;
- public ColumnId(
- TableId tableId,
- String columnName
- ) {
- this(tableId, columnName, null);
- }
-
- public ColumnId(
- TableId tableId,
- String columnName,
- String alias
- ) {
- assert columnName != null;
- this.tableId = tableId;
- this.name = columnName;
- this.alias = alias != null && !alias.trim().isEmpty() ? alias : name;
- this.hash = Objects.hash(this.tableId, this.name);
- }
+ public ColumnId(
+ TableId tableId,
+ String columnName
+ ) {
+ this(tableId, columnName, null);
+ }
- public TableId tableId() {
- return tableId;
- }
+ public ColumnId(
+ TableId tableId,
+ String columnName,
+ String alias
+ ) {
+ assert columnName != null;
+ this.tableId = tableId;
+ this.name = columnName;
+ this.alias = alias != null && !alias.trim().isEmpty() ? alias : name;
+ this.hash = Objects.hash(this.tableId, this.name);
+ }
- public String name() {
- return name;
- }
+ public TableId tableId() {
+ return tableId;
+ }
- public String aliasOrName() {
- return alias;
- }
+ public String name() {
+ return name;
+ }
- @Override
- public void appendTo(ExpressionBuilder builder, boolean useQuotes) {
- appendTo(builder, useQuotes ? QuoteMethod.ALWAYS : QuoteMethod.NEVER);
- }
+ public String aliasOrName() {
+ return alias;
+ }
- @Override
- public void appendTo(
- ExpressionBuilder builder,
- QuoteMethod useQuotes
- ) {
- if (tableId != null) {
- builder.append(tableId);
- builder.appendIdentifierDelimiter();
+ @Override
+ public void appendTo(ExpressionBuilder builder, boolean useQuotes) {
+ appendTo(builder, useQuotes ? QuoteMethod.ALWAYS : QuoteMethod.NEVER);
}
- builder.appendColumnName(this.name, useQuotes);
- }
- @Override
- public int hashCode() {
- return hash;
- }
+ @Override
+ public void appendTo(
+ ExpressionBuilder builder,
+ QuoteMethod useQuotes
+ ) {
+ if (tableId != null) {
+ builder.append(tableId);
+ builder.appendIdentifierDelimiter();
+ }
+ builder.appendColumnName(this.name, useQuotes);
+ }
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
+ @Override
+ public int hashCode() {
+ return hash;
}
- if (obj instanceof ColumnId) {
- ColumnId that = (ColumnId) obj;
- return Objects.equals(this.name, that.name) && Objects.equals(this.alias, that.alias)
- && Objects.equals(this.tableId, that.tableId);
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (obj instanceof ColumnId) {
+ ColumnId that = (ColumnId) obj;
+ return Objects.equals(this.name, that.name) && Objects.equals(this.alias, that.alias)
+ && Objects.equals(this.tableId, that.tableId);
+ }
+ return false;
}
- return false;
- }
- @Override
- public String toString() {
- return ExpressionBuilder.create().append(this).toString();
- }
+ @Override
+ public String toString() {
+ return ExpressionBuilder.create().append(this).toString();
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/DateColumnParser.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/DateColumnParser.java
index 386fe42..005c9db 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/DateColumnParser.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/DateColumnParser.java
@@ -44,7 +44,7 @@ public class DateColumnParser {
calendar.setTime(value);
if (calendar.get(11) == 0 && calendar.get(12) == 0 && calendar.get(13) == 0 && calendar.get(14) == 0) {
long unixMillis = calendar.getTimeInMillis();
- return (int)(unixMillis / 86400000L);
+ return (int) (unixMillis / 86400000L);
} else {
throw new ConnectException("Kafka Connect Date type should not have any time fields set to non-zero values.");
}
@@ -55,7 +55,7 @@ public class DateColumnParser {
if (!LOGICAL_NAME.equals(schema.getName())) {
throw new ConnectException("Requested conversion of Date object but the schema does not match.");
} else {
- return new java.util.Date((long)value * 86400000L);
+ return new java.util.Date((long) value * 86400000L);
}
}
}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/TimeColumnParser.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/TimeColumnParser.java
index e5a4198..282b60d 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/TimeColumnParser.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/column/parser/TimeColumnParser.java
@@ -25,7 +25,7 @@ import java.util.Calendar;
import java.util.Date;
import java.util.TimeZone;
-public class TimeColumnParser{
+public class TimeColumnParser {
public static final String LOGICAL_NAME = "org.apache.rocketmq.connect.jdbc.schema.column.parser.TimeColumnParser";
private static final long MILLIS_PER_DAY = 86400000L;
private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
@@ -43,7 +43,7 @@ public class TimeColumnParser{
calendar.setTime(value);
long unixMillis = calendar.getTimeInMillis();
if (unixMillis >= 0L && unixMillis <= 86400000L) {
- return (int)unixMillis;
+ return (int) unixMillis;
} else {
throw new ConnectException("Kafka Connect Time type should not have any date fields set to non-zero values.");
}
@@ -53,8 +53,8 @@ public class TimeColumnParser{
public static Date toLogical(Schema schema, int value) {
if (!LOGICAL_NAME.equals(schema.getName())) {
throw new ConnectException("Requested conversion of Date object but the schema does not match.");
- } else if (value >= 0 && (long)value <= 86400000L) {
- return new Date((long)value);
+ } else if (value >= 0 && (long) value <= 86400000L) {
+ return new Date((long) value);
} else {
throw new ConnectException("Time values must use number of milliseconds greater than 0 and less than 86400000");
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/db/DbStructure.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/db/DbStructure.java
index e2cc7e7..158c3f3 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/db/DbStructure.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/db/DbStructure.java
@@ -25,7 +25,7 @@ import org.apache.rocketmq.connect.jdbc.schema.table.TableId;
import org.apache.rocketmq.connect.jdbc.sink.metadata.FieldsMetadata;
import org.apache.rocketmq.connect.jdbc.sink.metadata.SinkRecordField;
import org.apache.rocketmq.connect.jdbc.exception.TableAlterOrCreateException;
-import org.apache.rocketmq.connect.jdbc.util.*;
+import org.apache.rocketmq.connect.jdbc.util.TableType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,268 +43,269 @@ import java.util.concurrent.ConcurrentHashMap;
*
*/
public class DbStructure {
- private static final Logger log = LoggerFactory.getLogger(DbStructure.class);
+ private static final Logger log = LoggerFactory.getLogger(DbStructure.class);
- private final DatabaseDialect dbDialect;
- private final TableDefinitions tableDefns;
+ private final DatabaseDialect dbDialect;
+ private final TableDefinitions tableDefns;
- public DbStructure(DatabaseDialect dbDialect) {
- this.dbDialect = dbDialect;
- this.tableDefns = new TableDefinitions(dbDialect);
- }
+ public DbStructure(DatabaseDialect dbDialect) {
+ this.dbDialect = dbDialect;
+ this.tableDefns = new TableDefinitions(dbDialect);
+ }
- /**
- * create 或者 alter 表
- * Create or amend table.
- *
- * @param config the connector configuration
- * @param connection the database connection handle
- * @param tableId the table ID
- * @param fieldsMetadata the fields metadata
- * @return whether a DDL operation was performed
- * @throws SQLException if a DDL operation was deemed necessary but failed
- */
- public boolean createOrAmendIfNecessary(
- final JdbcSinkConfig config,
- final Connection connection,
- final TableId tableId,
- final FieldsMetadata fieldsMetadata
- ) throws SQLException {
- if (tableDefns.get(connection, tableId) == null) {
- // Table does not yet exist, so attempt to create it ...
- try {
- create(config, connection, tableId, fieldsMetadata);
- } catch (SQLException sqle) {
- log.warn("Create failed, will attempt amend if table already exists", sqle);
- try {
- TableDefinition newDefn = tableDefns.refresh(connection, tableId);
- if (newDefn == null) {
- throw sqle;
- }
- } catch (SQLException e) {
- throw sqle;
+ /**
+ * Create or amend table.
+ *
+ * @param config the connector configuration
+ * @param connection the database connection handle
+ * @param tableId the table ID
+ * @param fieldsMetadata the fields metadata
+ * @return whether a DDL operation was performed
+ * @throws SQLException if a DDL operation was deemed necessary but failed
+ */
+ public boolean createOrAmendIfNecessary(
+ final JdbcSinkConfig config,
+ final Connection connection,
+ final TableId tableId,
+ final FieldsMetadata fieldsMetadata
+ ) throws SQLException {
+ if (tableDefns.get(connection, tableId) == null) {
+ // Table does not yet exist, so attempt to create it ...
+ try {
+ create(config, connection, tableId, fieldsMetadata);
+ } catch (SQLException sqle) {
+ log.warn("Create failed, will attempt amend if table already exists", sqle);
+ try {
+ TableDefinition newDefn = tableDefns.refresh(connection, tableId);
+ if (newDefn == null) {
+ throw sqle;
+ }
+ } catch (SQLException e) {
+ throw sqle;
+ }
+ }
}
- }
+ return amendIfNecessary(config, connection, tableId, fieldsMetadata, config.getMaxRetries());
}
- return amendIfNecessary(config, connection, tableId, fieldsMetadata, config.getMaxRetries());
- }
- public void applyDdl(String ddl) throws SQLException {
- Connection connection = dbDialect.getConnection();
- dbDialect.applyDdlStatements(connection, Collections.singletonList(ddl));
- }
- /**
- * Get the definition for the table with the given ID. This returns a cached definition if
- * there is one; otherwise, it reads the definition from the database
- *
- * @param connection the connection that may be used to fetch the table definition if not
- * already known; may not be null
- * @param tableId the ID of the table; may not be null
- * @return the table definition; or null if the table does not exist
- * @throws SQLException if there is an error getting the definition from the database
- */
- public TableDefinition tableDefinition(
- Connection connection,
- TableId tableId
- ) throws SQLException {
- TableDefinition defn = tableDefns.get(connection, tableId);
- if (defn != null) {
- return defn;
+ public void applyDdl(String ddl) throws SQLException {
+ Connection connection = dbDialect.getConnection();
+ dbDialect.applyDdlStatements(connection, Collections.singletonList(ddl));
}
- return tableDefns.refresh(connection, tableId);
- }
- /**
- * @throws SQLException if CREATE failed
- */
- void create(
- final JdbcSinkConfig config,
- final Connection connection,
- final TableId tableId,
- final FieldsMetadata fieldsMetadata
- ) throws SQLException{
- if (!config.isAutoCreate()) {
- throw new TableAlterOrCreateException(
- String.format("Table %s is missing and auto-creation is disabled", tableId)
- );
+ /**
+ * Get the definition for the table with the given ID. This returns a cached definition if
+ * there is one; otherwise, it reads the definition from the database
+ *
+ * @param connection the connection that may be used to fetch the table definition if not
+ * already known; may not be null
+ * @param tableId the ID of the table; may not be null
+ * @return the table definition; or null if the table does not exist
+ * @throws SQLException if there is an error getting the definition from the database
+ */
+ public TableDefinition tableDefinition(
+ Connection connection,
+ TableId tableId
+ ) throws SQLException {
+ TableDefinition defn = tableDefns.get(connection, tableId);
+ if (defn != null) {
+ return defn;
+ }
+ return tableDefns.refresh(connection, tableId);
}
- String sql = dbDialect.buildCreateTableStatement(tableId, fieldsMetadata.allFields.values());
- log.info("Creating table with sql: {}", sql);
- dbDialect.applyDdlStatements(connection, Collections.singletonList(sql));
- }
-
- /**
- * @return whether an ALTER was successfully performed
- * @throws SQLException if ALTER was deemed necessary but failed
- */
- boolean amendIfNecessary(
- final JdbcSinkConfig config,
- final Connection connection,
- final TableId tableId,
- final FieldsMetadata fieldsMetadata,
- final int maxRetries
- ) throws SQLException, TableAlterOrCreateException {
- final TableDefinition tableDefn = tableDefns.get(connection, tableId);
- final Set<SinkRecordField> missingFields = missingFields(
- fieldsMetadata.allFields.values(),
- tableDefn.columnNames()
- );
- if (missingFields.isEmpty()) {
- return false;
+ /**
+ * @throws SQLException if CREATE failed
+ */
+ void create(
+ final JdbcSinkConfig config,
+ final Connection connection,
+ final TableId tableId,
+ final FieldsMetadata fieldsMetadata
+ ) throws SQLException {
+ if (!config.isAutoCreate()) {
+ throw new TableAlterOrCreateException(
+ String.format("Table %s is missing and auto-creation is disabled", tableId)
+ );
+ }
+ String sql = dbDialect.buildCreateTableStatement(tableId, fieldsMetadata.allFields.values());
+ log.info("Creating table with sql: {}", sql);
+ dbDialect.applyDdlStatements(connection, Collections.singletonList(sql));
}
- // At this point there are missing fields
- TableType type = tableDefn.type();
- switch (type) {
- case TABLE:
- // Rather than embed the logic and change lots of lines, just break out
- break;
- case VIEW:
- default:
- throw new TableAlterOrCreateException(
- String.format(
- "%s %s is missing fields (%s) and ALTER %s is unsupported",
- type.capitalized(),
- tableId,
- missingFields,
- type.jdbcName()
- )
+
+ /**
+ * @return whether an ALTER was successfully performed
+ * @throws SQLException if ALTER was deemed necessary but failed
+ */
+ boolean amendIfNecessary(
+ final JdbcSinkConfig config,
+ final Connection connection,
+ final TableId tableId,
+ final FieldsMetadata fieldsMetadata,
+ final int maxRetries
+ ) throws SQLException, TableAlterOrCreateException {
+ final TableDefinition tableDefn = tableDefns.get(connection, tableId);
+ final Set<SinkRecordField> missingFields = missingFields(
+ fieldsMetadata.allFields.values(),
+ tableDefn.columnNames()
);
- }
- final Set<SinkRecordField> replacedMissingFields = new HashSet<>();
- for (SinkRecordField missingField: missingFields) {
- if (!missingField.isOptional() && missingField.defaultValue() == null) {
- throw new TableAlterOrCreateException(String.format(
- "Cannot ALTER %s %s to add missing field %s, as the field is not optional and does "
- + "not have a default value",
- type.jdbcName(),
- tableId,
- missingField
- ));
- }
- }
+ if (missingFields.isEmpty()) {
+ return false;
+ }
+ // At this point there are missing fields
+ TableType type = tableDefn.type();
+ switch (type) {
+ case TABLE:
+ // Rather than embed the logic and change lots of lines, just break out
+ break;
+ case VIEW:
+ default:
+ throw new TableAlterOrCreateException(
+ String.format(
+ "%s %s is missing fields (%s) and ALTER %s is unsupported",
+ type.capitalized(),
+ tableId,
+ missingFields,
+ type.jdbcName()
+ )
+ );
+ }
- if (!config.isAutoCreate()) {
- throw new TableAlterOrCreateException(String.format(
- "%s %s is missing fields (%s) and auto-evolution is disabled",
- type.capitalized(),
- tableId,
- replacedMissingFields
- ));
- }
+ final Set<SinkRecordField> replacedMissingFields = new HashSet<>();
+ for (SinkRecordField missingField : missingFields) {
+ if (!missingField.isOptional() && missingField.defaultValue() == null) {
+ throw new TableAlterOrCreateException(String.format(
+ "Cannot ALTER %s %s to add missing field %s, as the field is not optional and does "
+ + "not have a default value",
+ type.jdbcName(),
+ tableId,
+ missingField
+ ));
+ }
+ }
- final List<String> amendTableQueries = dbDialect.buildAlterTable(tableId, replacedMissingFields);
- log.info(
- "Amending {} to add missing fields:{} maxRetries:{} with SQL: {}",
- type,
- replacedMissingFields,
- maxRetries,
- amendTableQueries
- );
- try {
- dbDialect.applyDdlStatements(connection, amendTableQueries);
- } catch (SQLException sqle) {
- if (maxRetries <= 0) {
- throw new TableAlterOrCreateException(
- String.format(
- "Failed to amend %s '%s' to add missing fields: %s",
+ if (!config.isAutoCreate()) {
+ throw new TableAlterOrCreateException(String.format(
+ "%s %s is missing fields (%s) and auto-evolution is disabled",
+ type.capitalized(),
+ tableId,
+ replacedMissingFields
+ ));
+ }
+
+ final List<String> amendTableQueries = dbDialect.buildAlterTable(tableId, replacedMissingFields);
+ log.info(
+ "Amending {} to add missing fields:{} maxRetries:{} with SQL: {}",
type,
- tableId,
- replacedMissingFields
- ),
- sqle
+ replacedMissingFields,
+ maxRetries,
+ amendTableQueries
);
- }
- log.warn("Amend failed, re-attempting", sqle);
- tableDefns.refresh(connection, tableId);
- // Perhaps there was a race with other tasks to add the columns
- return amendIfNecessary(
- config,
- connection,
- tableId,
- fieldsMetadata,
- maxRetries - 1
- );
+ try {
+ dbDialect.applyDdlStatements(connection, amendTableQueries);
+ } catch (SQLException sqle) {
+ if (maxRetries <= 0) {
+ throw new TableAlterOrCreateException(
+ String.format(
+ "Failed to amend %s '%s' to add missing fields: %s",
+ type,
+ tableId,
+ replacedMissingFields
+ ),
+ sqle
+ );
+ }
+ log.warn("Amend failed, re-attempting", sqle);
+ tableDefns.refresh(connection, tableId);
+ // Perhaps there was a race with other tasks to add the columns
+ return amendIfNecessary(
+ config,
+ connection,
+ tableId,
+ fieldsMetadata,
+ maxRetries - 1
+ );
+ }
+
+ tableDefns.refresh(connection, tableId);
+ return true;
}
- tableDefns.refresh(connection, tableId);
- return true;
- }
+ Set<SinkRecordField> missingFields(
+ Collection<SinkRecordField> fields,
+ Set<String> dbColumnNames
+ ) {
+ final Set<SinkRecordField> missingFields = new HashSet<>();
+ for (SinkRecordField field : fields) {
+ if (!dbColumnNames.contains(field.name())) {
+ log.debug("Found missing field: {}", field);
+ missingFields.add(field);
+ }
+ }
- Set<SinkRecordField> missingFields(
- Collection<SinkRecordField> fields,
- Set<String> dbColumnNames
- ) {
- final Set<SinkRecordField> missingFields = new HashSet<>();
- for (SinkRecordField field : fields) {
- if (!dbColumnNames.contains(field.name())) {
- log.debug("Found missing field: {}", field);
- missingFields.add(field);
- }
- }
+ if (missingFields.isEmpty()) {
+ return missingFields;
+ }
- if (missingFields.isEmpty()) {
- return missingFields;
- }
+ // check if the missing fields can be located by ignoring case
+ Set<String> columnNamesLowerCase = new HashSet<>();
+ for (String columnName : dbColumnNames) {
+ columnNamesLowerCase.add(columnName.toLowerCase());
+ }
- // check if the missing fields can be located by ignoring case
- Set<String> columnNamesLowerCase = new HashSet<>();
- for (String columnName: dbColumnNames) {
- columnNamesLowerCase.add(columnName.toLowerCase());
- }
+ if (columnNamesLowerCase.size() != dbColumnNames.size()) {
+ log.warn(
+ "Table has column names that differ only by case. Original columns={}",
+ dbColumnNames
+ );
+ }
- if (columnNamesLowerCase.size() != dbColumnNames.size()) {
- log.warn(
- "Table has column names that differ only by case. Original columns={}",
- dbColumnNames
- );
- }
+ final Set<SinkRecordField> missingFieldsIgnoreCase = new HashSet<>();
+ for (SinkRecordField missing : missingFields) {
+ if (!columnNamesLowerCase.contains(missing.name().toLowerCase())) {
+ missingFieldsIgnoreCase.add(missing);
+ }
+ }
- final Set<SinkRecordField> missingFieldsIgnoreCase = new HashSet<>();
- for (SinkRecordField missing: missingFields) {
- if (!columnNamesLowerCase.contains(missing.name().toLowerCase())) {
- missingFieldsIgnoreCase.add(missing);
- }
- }
+ if (missingFieldsIgnoreCase.size() > 0) {
+ log.info(
+ "Unable to find fields {} among column names {}",
+ missingFieldsIgnoreCase,
+ dbColumnNames
+ );
+ }
- if (missingFieldsIgnoreCase.size() > 0) {
- log.info(
- "Unable to find fields {} among column names {}",
- missingFieldsIgnoreCase,
- dbColumnNames
- );
+ return missingFieldsIgnoreCase;
}
- return missingFieldsIgnoreCase;
- }
+ /**
+ * delete field
+ *
+ * @param fields
+ * @param dbColumnNames
+ * @return
+ */
+ Set<ColumnDefinition> alterDeleteFields(
+ Collection<SinkRecordField> fields,
+ Map<String, ColumnDefinition> dbColumnNames
+ ) {
- /**
- * delete field
- * @param fields
- * @param dbColumnNames
- * @return
- */
- Set<ColumnDefinition> alterDeleteFields(
- Collection<SinkRecordField> fields,
- Map<String, ColumnDefinition> dbColumnNames
- ) {
-
- final Map<String,SinkRecordField> recordFields = new ConcurrentHashMap<>();
- for (SinkRecordField field : fields) {
- recordFields.put(field.name(),field);
- }
+ final Map<String, SinkRecordField> recordFields = new ConcurrentHashMap<>();
+ for (SinkRecordField field : fields) {
+ recordFields.put(field.name(), field);
+ }
- Set<ColumnDefinition> deleteColumns=new HashSet<>();
- // filter delete column name
- for (String columnName : dbColumnNames.keySet()) {
- if (!recordFields.containsKey(columnName)) {
- log.debug("Found delete field: {}", columnName);
- ColumnDefinition columnDefinition=dbColumnNames.get(columnName);
- deleteColumns.add(dbColumnNames.get(columnName));
- }
+ Set<ColumnDefinition> deleteColumns = new HashSet<>();
+ // filter delete column name
+ for (String columnName : dbColumnNames.keySet()) {
+ if (!recordFields.containsKey(columnName)) {
+ log.debug("Found delete field: {}", columnName);
+ ColumnDefinition columnDefinition = dbColumnNames.get(columnName);
+ deleteColumns.add(dbColumnNames.get(columnName));
+ }
+ }
+ return deleteColumns;
}
- return deleteColumns;
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinition.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinition.java
index 7027e9b..be64724 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinition.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinition.java
@@ -30,94 +30,94 @@ import java.util.Set;
* A description of a table.
*/
public class TableDefinition {
- private final TableId id;
- private final Map<String, ColumnDefinition> columnsByName = new LinkedHashMap<>();
- private final Map<String, String> pkColumnNames = new LinkedHashMap<>();
- private final TableType type;
-
- public TableDefinition(
- TableId id,
- Iterable<ColumnDefinition> columns
- ) {
- this(id, columns, TableType.TABLE);
- }
-
- public TableDefinition(
- TableId id,
- Iterable<ColumnDefinition> columns,
- TableType type
- ) {
- this.id = id;
- this.type = Objects.requireNonNull(type);
- for (ColumnDefinition defn : columns) {
- String columnName = defn.id().name();
- columnsByName.put(
- columnName,
- defn.forTable(this.id)
- );
- if (defn.isPrimaryKey()) {
- this.pkColumnNames.put(
- columnName,
- columnName
- );
- }
+ private final TableId id;
+ private final Map<String, ColumnDefinition> columnsByName = new LinkedHashMap<>();
+ private final Map<String, String> pkColumnNames = new LinkedHashMap<>();
+ private final TableType type;
+
+ public TableDefinition(
+ TableId id,
+ Iterable<ColumnDefinition> columns
+ ) {
+ this(id, columns, TableType.TABLE);
}
- }
- public TableId id() {
- return id;
- }
+ public TableDefinition(
+ TableId id,
+ Iterable<ColumnDefinition> columns,
+ TableType type
+ ) {
+ this.id = id;
+ this.type = Objects.requireNonNull(type);
+ for (ColumnDefinition defn : columns) {
+ String columnName = defn.id().name();
+ columnsByName.put(
+ columnName,
+ defn.forTable(this.id)
+ );
+ if (defn.isPrimaryKey()) {
+ this.pkColumnNames.put(
+ columnName,
+ columnName
+ );
+ }
+ }
+ }
- public TableType type() {
- return type;
- }
+ public TableId id() {
+ return id;
+ }
- public int columnCount() {
- return columnsByName.size();
- }
+ public TableType type() {
+ return type;
+ }
- public ColumnDefinition definitionForColumn(String name) {
- return columnsByName.get(name);
- }
+ public int columnCount() {
+ return columnsByName.size();
+ }
- public Collection<ColumnDefinition> definitionsForColumns() {
- return columnsByName.values();
- }
+ public ColumnDefinition definitionForColumn(String name) {
+ return columnsByName.get(name);
+ }
- public Collection<String> primaryKeyColumnNames() {
- return pkColumnNames.values();
- }
+ public Collection<ColumnDefinition> definitionsForColumns() {
+ return columnsByName.values();
+ }
- public Set<String> columnNames() {
- return columnsByName.keySet();
- }
+ public Collection<String> primaryKeyColumnNames() {
+ return pkColumnNames.values();
+ }
+
+ public Set<String> columnNames() {
+ return columnsByName.keySet();
+ }
- @Override
- public int hashCode() {
- return id.hashCode();
- }
+ @Override
+ public int hashCode() {
+ return id.hashCode();
+ }
- @Override
- public boolean equals(Object obj) {
- if (obj == null) {
- return true;
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) {
+ return true;
+ }
+ if (obj instanceof TableDefinition) {
+ TableDefinition that = (TableDefinition) obj;
+ return Objects.equals(this.id(), that.id())
+ && Objects.equals(this.type(), that.type())
+ && Objects.equals(this.definitionsForColumns(), that.definitionsForColumns());
+ }
+ return false;
}
- if (obj instanceof TableDefinition) {
- TableDefinition that = (TableDefinition) obj;
- return Objects.equals(this.id(), that.id())
- && Objects.equals(this.type(), that.type())
- && Objects.equals(this.definitionsForColumns(), that.definitionsForColumns());
+
+ @Override
+ public String toString() {
+ return String.format(
+ "Table{name='%s', type=%s columns=%s}",
+ id,
+ type,
+ definitionsForColumns()
+ );
}
- return false;
- }
-
- @Override
- public String toString() {
- return String.format(
- "Table{name='%s', type=%s columns=%s}",
- id,
- type,
- definitionsForColumns()
- );
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinitions.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinitions.java
index 5169490..719ee2e 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinitions.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableDefinitions.java
@@ -30,64 +30,64 @@ import java.util.Map;
*/
public class TableDefinitions {
- private static final Logger log = LoggerFactory.getLogger(TableDefinitions.class);
+ private static final Logger log = LoggerFactory.getLogger(TableDefinitions.class);
- private final Map<TableId, TableDefinition> cache = new HashMap<>();
- private final DatabaseDialect dialect;
+ private final Map<TableId, TableDefinition> cache = new HashMap<>();
+ private final DatabaseDialect dialect;
- /**
- * Create an instance that uses the specified database dialect.
- *
- * @param dialect the database dialect; may not be null
- */
- public TableDefinitions(DatabaseDialect dialect) {
- this.dialect = dialect;
- }
+ /**
+ * Create an instance that uses the specified database dialect.
+ *
+ * @param dialect the database dialect; may not be null
+ */
+ public TableDefinitions(DatabaseDialect dialect) {
+ this.dialect = dialect;
+ }
- /**
- * Get the {@link TableDefinition} for the given table.
- *
- * @param connection the JDBC connection to use; may not be null
- * @param tableId the table identifier; may not be null
- * @return the cached {@link TableDefinition}, or null if there is no such table
- * @throws SQLException if there is any problem using the connection
- */
- public TableDefinition get(
- Connection connection,
- final TableId tableId
- ) throws SQLException {
- TableDefinition dbTable = cache.get(tableId);
- if (dbTable == null) {
- if (dialect.tableExists(connection, tableId)) {
- dbTable = dialect.describeTable(connection, tableId);
- if (dbTable != null) {
- log.info("Setting metadata for table {} to {}", tableId, dbTable);
- cache.put(tableId, dbTable);
+ /**
+ * Get the {@link TableDefinition} for the given table.
+ *
+ * @param connection the JDBC connection to use; may not be null
+ * @param tableId the table identifier; may not be null
+ * @return the cached {@link TableDefinition}, or null if there is no such table
+ * @throws SQLException if there is any problem using the connection
+ */
+ public TableDefinition get(
+ Connection connection,
+ final TableId tableId
+ ) throws SQLException {
+ TableDefinition dbTable = cache.get(tableId);
+ if (dbTable == null) {
+ if (dialect.tableExists(connection, tableId)) {
+ dbTable = dialect.describeTable(connection, tableId);
+ if (dbTable != null) {
+ log.info("Setting metadata for table {} to {}", tableId, dbTable);
+ cache.put(tableId, dbTable);
+ }
+ }
}
- }
+ return dbTable;
}
- return dbTable;
- }
- /**
- * Refresh the cached {@link TableDefinition} for the given table.
- *
- * @param connection the JDBC connection to use; may not be null
- * @param tableId the table identifier; may not be null
- * @return the refreshed {@link TableDefinition}, or null if there is no such table
- * @throws SQLException if there is any problem using the connection
- */
- public TableDefinition refresh(
- Connection connection,
- TableId tableId
- ) throws SQLException {
- TableDefinition dbTable = dialect.describeTable(connection, tableId);
- if (dbTable != null) {
- log.info("Refreshing metadata for table {} to {}", tableId, dbTable);
- cache.put(dbTable.id(), dbTable);
- } else {
- log.warn("Failed to refresh metadata for table {}", tableId);
+ /**
+ * Refresh the cached {@link TableDefinition} for the given table.
+ *
+ * @param connection the JDBC connection to use; may not be null
+ * @param tableId the table identifier; may not be null
+ * @return the refreshed {@link TableDefinition}, or null if there is no such table
+ * @throws SQLException if there is any problem using the connection
+ */
+ public TableDefinition refresh(
+ Connection connection,
+ TableId tableId
+ ) throws SQLException {
+ TableDefinition dbTable = dialect.describeTable(connection, tableId);
+ if (dbTable != null) {
+ log.info("Refreshing metadata for table {} to {}", tableId, dbTable);
+ cache.put(dbTable.id(), dbTable);
+ } else {
+ log.warn("Failed to refresh metadata for table {}", tableId);
+ }
+ return dbTable;
}
- return dbTable;
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableId.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableId.java
index e004fd2..776f786 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableId.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/schema/table/TableId.java
@@ -24,121 +24,121 @@ import java.util.Objects;
public class TableId implements Comparable<TableId>, ExpressionBuilder.Expressable {
- private final String catalogName;
- private final String schemaName;
- private final String tableName;
- private final int hash;
+ private final String catalogName;
+ private final String schemaName;
+ private final String tableName;
+ private final int hash;
- public TableId(
- String catalogName,
- String schemaName,
- String tableName
- ) {
- this.catalogName = catalogName == null || catalogName.isEmpty() ? null : catalogName;
- this.schemaName = schemaName == null || schemaName.isEmpty() ? null : schemaName;
- this.tableName = tableName;
- this.hash = Objects.hash(catalogName, schemaName, tableName);
- }
-
- public String catalogName() {
- return catalogName;
- }
-
- public String schemaName() {
- return schemaName;
- }
-
- public String dbName(){
- if(StringUtils.isNotBlank(catalogName)){
- return catalogName;
+ public TableId(
+ String catalogName,
+ String schemaName,
+ String tableName
+ ) {
+ this.catalogName = catalogName == null || catalogName.isEmpty() ? null : catalogName;
+ this.schemaName = schemaName == null || schemaName.isEmpty() ? null : schemaName;
+ this.tableName = tableName;
+ this.hash = Objects.hash(catalogName, schemaName, tableName);
}
- return schemaName;
- }
- public String tableName() {
- return tableName;
- }
-
- @Override
- public void appendTo(ExpressionBuilder builder, boolean useQuotes) {
- appendTo(builder, useQuotes ? QuoteMethod.ALWAYS : QuoteMethod.NEVER);
- }
-
- @Override
- public void appendTo(
- ExpressionBuilder builder,
- QuoteMethod useQuotes
- ) {
- if (catalogName != null) {
- builder.appendIdentifier(catalogName, useQuotes);
- builder.appendIdentifierDelimiter();
+ public String catalogName() {
+ return catalogName;
}
- if (schemaName != null) {
- builder.appendIdentifier(schemaName, useQuotes);
- builder.appendIdentifierDelimiter();
+
+ public String schemaName() {
+ return schemaName;
}
- builder.appendTableName(tableName, useQuotes);
- }
- @Override
- public int hashCode() {
- return hash;
- }
+ public String dbName() {
+ if (StringUtils.isNotBlank(catalogName)) {
+ return catalogName;
+ }
+ return schemaName;
+ }
- @Override
- public boolean equals(Object obj) {
- if (obj == this) {
- return true;
+ public String tableName() {
+ return tableName;
}
- if (obj instanceof TableId) {
- TableId that = (TableId) obj;
- return Objects.equals(this.catalogName, that.catalogName)
- && Objects.equals(this.schemaName, that.schemaName)
- && Objects.equals(this.tableName, that.tableName);
+
+ @Override
+ public void appendTo(ExpressionBuilder builder, boolean useQuotes) {
+ appendTo(builder, useQuotes ? QuoteMethod.ALWAYS : QuoteMethod.NEVER);
}
- return false;
- }
- @Override
- public int compareTo(TableId that) {
- if (that == this) {
- return 0;
+ @Override
+ public void appendTo(
+ ExpressionBuilder builder,
+ QuoteMethod useQuotes
+ ) {
+ if (catalogName != null) {
+ builder.appendIdentifier(catalogName, useQuotes);
+ builder.appendIdentifierDelimiter();
+ }
+ if (schemaName != null) {
+ builder.appendIdentifier(schemaName, useQuotes);
+ builder.appendIdentifierDelimiter();
+ }
+ builder.appendTableName(tableName, useQuotes);
}
- int diff = this.tableName.compareTo(that.tableName);
- if (diff != 0) {
- return diff;
+
+ @Override
+ public int hashCode() {
+ return hash;
}
- if (this.schemaName == null) {
- if (that.schemaName != null) {
- return -1;
- }
- } else {
- if (that.schemaName == null) {
- return 1;
- }
- diff = this.schemaName.compareTo(that.schemaName);
- if (diff != 0) {
- return diff;
- }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (obj instanceof TableId) {
+ TableId that = (TableId) obj;
+ return Objects.equals(this.catalogName, that.catalogName)
+ && Objects.equals(this.schemaName, that.schemaName)
+ && Objects.equals(this.tableName, that.tableName);
+ }
+ return false;
}
- if (this.catalogName == null) {
- if (that.catalogName != null) {
- return -1;
- }
- } else {
- if (that.catalogName == null) {
- return 1;
- }
- diff = this.catalogName.compareTo(that.catalogName);
- if (diff != 0) {
- return diff;
- }
+
+ @Override
+ public int compareTo(TableId that) {
+ if (that == this) {
+ return 0;
+ }
+ int diff = this.tableName.compareTo(that.tableName);
+ if (diff != 0) {
+ return diff;
+ }
+ if (this.schemaName == null) {
+ if (that.schemaName != null) {
+ return -1;
+ }
+ } else {
+ if (that.schemaName == null) {
+ return 1;
+ }
+ diff = this.schemaName.compareTo(that.schemaName);
+ if (diff != 0) {
+ return diff;
+ }
+ }
+ if (this.catalogName == null) {
+ if (that.catalogName != null) {
+ return -1;
+ }
+ } else {
+ if (that.catalogName == null) {
+ return 1;
+ }
+ diff = this.catalogName.compareTo(that.catalogName);
+ if (diff != 0) {
+ return diff;
+ }
+ }
+ return 0;
}
- return 0;
- }
- @Override
- public String toString() {
- return ExpressionBuilder.create().append(this).toString();
- }
+ @Override
+ public String toString() {
+ return ExpressionBuilder.create().append(this).toString();
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/BufferedRecords.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/BufferedRecords.java
index f20927c..87773fb 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/BufferedRecords.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/BufferedRecords.java
@@ -21,6 +21,8 @@ import io.openmessaging.connector.api.data.Schema;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConfig;
import org.apache.rocketmq.connect.jdbc.dialect.DatabaseDialect;
+import org.apache.rocketmq.connect.jdbc.dialect.impl.GenericDatabaseDialect;
+import org.apache.rocketmq.connect.jdbc.dialect.impl.OpenMLDBDatabaseDialect;
import org.apache.rocketmq.connect.jdbc.schema.db.DbStructure;
import org.apache.rocketmq.connect.jdbc.sink.metadata.FieldsMetadata;
import org.apache.rocketmq.connect.jdbc.sink.metadata.SchemaPair;
@@ -39,6 +41,7 @@ import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
+
import static java.util.Objects.isNull;
import static java.util.Objects.nonNull;
@@ -46,285 +49,303 @@ import static java.util.Objects.nonNull;
* buffered records
*/
public class BufferedRecords {
- private static final Logger log = LoggerFactory.getLogger(BufferedRecords.class);
-
-
- private final TableId tableId;
- private final JdbcSinkConfig config;
- private final DatabaseDialect dbDialect;
- private final DbStructure dbStructure;
- private final Connection connection;
+ private static final Logger log = LoggerFactory.getLogger(BufferedRecords.class);
- private List<ConnectRecord> records = new ArrayList<>();
- private Schema schema;
- private FieldsMetadata fieldsMetadata;
- private PreparedStatement updatePreparedStatement;
- private PreparedStatement deletePreparedStatement;
- private DatabaseDialect.StatementBinder updateStatementBinder;
- private DatabaseDialect.StatementBinder deleteStatementBinder;
+ private final TableId tableId;
+ private final JdbcSinkConfig config;
+ private final DatabaseDialect dbDialect;
+ private final DbStructure dbStructure;
+ private final Connection connection;
- public BufferedRecords(
- JdbcSinkConfig config,
- TableId tableId,
- DatabaseDialect dbDialect,
- DbStructure dbStructure,
- Connection connection
- ) {
- this.tableId = tableId;
- this.config = config;
- this.dbDialect = dbDialect;
- this.dbStructure = dbStructure;
- this.connection = connection;
- }
+ private List<ConnectRecord> records = new ArrayList<>();
+ private Schema schema;
+ private FieldsMetadata fieldsMetadata;
+ private PreparedStatement updatePreparedStatement;
+ private PreparedStatement deletePreparedStatement;
+ private DatabaseDialect.StatementBinder updateStatementBinder;
+ private DatabaseDialect.StatementBinder deleteStatementBinder;
- /**
- * add record
- * @param record
- * @return
- * @throws SQLException
- */
- public List<ConnectRecord> add(ConnectRecord record) throws SQLException {
- final List<ConnectRecord> flushed = new ArrayList<>();
- boolean schemaChanged = false;
- // check and update schema
- if (!Objects.equals(schema, record.getSchema())) {
- schema = record.getSchema();
- schemaChanged = true;
+ public BufferedRecords(
+ JdbcSinkConfig config,
+ TableId tableId,
+ DatabaseDialect dbDialect,
+ DbStructure dbStructure,
+ Connection connection
+ ) {
+ this.tableId = tableId;
+ this.config = config;
+ this.dbDialect = dbDialect;
+ this.dbStructure = dbStructure;
+ this.connection = connection;
}
- if (schemaChanged || updateStatementBinder == null) {
- // Each batch needs to have the same schemas, so get the buffered records out
- flushed.addAll(flush());
- // re-initialize everything that depends on the record schema
- final SchemaPair schemaPair = new SchemaPair(
- record.getSchema(),
- record.getExtensions()
- );
- // 抽取字段
- fieldsMetadata = FieldsMetadata.extract(
- tableId.tableName(),
- config.pkMode,
- config.getPkFields(),
- config.getFieldsWhitelist(),
- schemaPair
- );
- // create or alter table
- dbStructure.createOrAmendIfNecessary(
- config,
- connection,
- tableId,
- fieldsMetadata
- );
- final String insertSql = getInsertSql();
- final String deleteSql = getDeleteSql();
- log.debug(
- "{} sql: {} deleteSql: {} meta: {}",
- config.getInsertMode(),
- insertSql,
- deleteSql,
- fieldsMetadata
- );
- close();
- updatePreparedStatement = dbDialect.createPreparedStatement(connection, insertSql);
- updateStatementBinder = dbDialect.statementBinder(
- updatePreparedStatement,
- config.pkMode,
- schemaPair,
- fieldsMetadata,
- dbStructure.tableDefinition(connection, tableId),
- config.getInsertMode()
- );
- if (config.isDeleteEnabled() && nonNull(deleteSql)) {
- deletePreparedStatement = dbDialect.createPreparedStatement(connection, deleteSql);
- deleteStatementBinder = dbDialect.statementBinder(
- deletePreparedStatement,
- config.pkMode,
- schemaPair,
- fieldsMetadata,
- dbStructure.tableDefinition(connection, tableId),
- config.getInsertMode()
- );
- }
- }
+ /**
+ * add record
+ *
+ * @param record
+ * @return
+ * @throws SQLException
+ */
+ public List<ConnectRecord> add(ConnectRecord record) throws SQLException {
+ final List<ConnectRecord> flushed = new ArrayList<>();
+ boolean schemaChanged = false;
+ // check and update schema
+ if (!Objects.equals(schema, record.getSchema())) {
+ schema = record.getSchema();
+ schemaChanged = true;
+ }
- records.add(record);
- if (records.size() >= config.getBatchSize()) {
- flushed.addAll(flush());
- }
- return flushed;
- }
+ if (schemaChanged || updateStatementBinder == null) {
+ // Each batch needs to have the same schemas, so get the buffered records out
+ flushed.addAll(flush());
+ // re-initialize everything that depends on the record schema
+ final SchemaPair schemaPair = new SchemaPair(
+ record.getSchema(),
+ record.getExtensions()
+ );
+ // extract field
+ fieldsMetadata = FieldsMetadata.extract(
+ tableId.tableName(),
+ config.pkMode,
+ config.getPkFields(),
+ config.getFieldsWhitelist(),
+ schemaPair
+ );
+ // create or alter table
+ dbStructure.createOrAmendIfNecessary(
+ config,
+ connection,
+ tableId,
+ fieldsMetadata
+ );
+ final String insertSql = getInsertSql();
+ final String deleteSql = getDeleteSql();
+ log.debug(
+ "{} sql: {} deleteSql: {} meta: {}",
+ config.getInsertMode(),
+ insertSql,
+ deleteSql,
+ fieldsMetadata
+ );
+ close();
+ updatePreparedStatement = dbDialect.createPreparedStatement(connection, insertSql);
+ updateStatementBinder = dbDialect.statementBinder(
+ updatePreparedStatement,
+ config.pkMode,
+ schemaPair,
+ fieldsMetadata,
+ dbStructure.tableDefinition(connection, tableId),
+ config.getInsertMode()
+ );
+ if (config.isDeleteEnabled() && nonNull(deleteSql)) {
+ deletePreparedStatement = dbDialect.createPreparedStatement(connection, deleteSql);
+ deleteStatementBinder = dbDialect.statementBinder(
+ deletePreparedStatement,
+ config.pkMode,
+ schemaPair,
+ fieldsMetadata,
+ dbStructure.tableDefinition(connection, tableId),
+ config.getInsertMode()
+ );
+ }
+ }
- public List<ConnectRecord> flush() throws SQLException {
- if (records.isEmpty()) {
- log.debug("Records is empty");
- return new ArrayList<>();
- }
- log.debug("Flushing {} buffered records", records.size());
- for (ConnectRecord record : records) {
- if (isNull(record.getData()) && nonNull(deleteStatementBinder)) {
- deleteStatementBinder.bindRecord(record);
- } else {
- updateStatementBinder.bindRecord(record);
- }
+ records.add(record);
+ if (records.size() >= config.getBatchSize()) {
+ flushed.addAll(flush());
+ }
+ return flushed;
}
- Optional<Long> totalUpdateCount = executeUpdates();
- long totalDeleteCount = executeDeletes();
- final long expectedCount = updateRecordCount();
- log.trace("{} records:{} resulting in totalUpdateCount:{} totalDeleteCount:{}",
- config.getInsertMode(), records.size(), totalUpdateCount, totalDeleteCount
- );
- if (totalUpdateCount.filter(total -> total != expectedCount).isPresent()
- && config.getInsertMode() == JdbcSinkConfig.InsertMode.INSERT) {
- throw new ConnectException(String.format(
- "Update count (%d) did not sum up to total number of records inserted (%d)",
- totalUpdateCount.get(),
- expectedCount
- ));
- }
- if (!totalUpdateCount.isPresent()) {
- log.info(
- "{} records:{} , but no count of the number of rows it affected is available",
- config.getInsertMode(),
- records.size()
- );
- }
+ public List<ConnectRecord> flush() throws SQLException {
+ if (records.isEmpty()) {
+ log.debug("Records is empty");
+ return new ArrayList<>();
+ }
+ log.debug("Flushing {} buffered records", records.size());
+ for (ConnectRecord record : records) {
+ if (isNull(record.getData()) && nonNull(deleteStatementBinder)) {
+ deleteStatementBinder.bindRecord(record);
+ } else {
+ updateStatementBinder.bindRecord(record);
+ }
+ }
+ Optional<Long> totalUpdateCount = executeUpdates();
+ long totalDeleteCount = executeDeletes();
+ final long expectedCount = updateRecordCount();
+ log.trace("{} records:{} resulting in totalUpdateCount:{} totalDeleteCount:{}",
+ config.getInsertMode(),
+ records.size(),
+ totalUpdateCount,
+ totalDeleteCount
+ );
+ if (totalUpdateCount.filter(total -> total != expectedCount).isPresent()
+ && config.getInsertMode() == JdbcSinkConfig.InsertMode.INSERT) {
+ if (dbDialect.name().equals(GenericDatabaseDialect.DialectName.generateDialectName(OpenMLDBDatabaseDialect.class)) && totalUpdateCount.get() == 0) {
+ // openMLDB execute success result 0; do nothing
+ } else {
+ throw new ConnectException(
+ String.format(
+ "Update count (%d) did not sum up to total number of records inserted (%d)",
+ totalUpdateCount.get(),
+ expectedCount
+ )
+ );
+ }
+ }
+ if (!totalUpdateCount.isPresent()) {
+ log.info(
+ "{} records:{} , but no count of the number of rows it affected is available",
+ config.getInsertMode(),
+ records.size()
+ );
+ }
- final List<ConnectRecord> flushedRecords = records;
- records = new ArrayList<>();
- return flushedRecords;
- }
+ final List<ConnectRecord> flushedRecords = records;
+ records = new ArrayList<>();
+ return flushedRecords;
+ }
- /**
- * @return an optional count of all updated rows or an empty optional if no info is available
- */
- private Optional<Long> executeUpdates() throws SQLException {
- Optional<Long> count = Optional.empty();
- if (nonNull(updatePreparedStatement)){
- try{
- for (int updateCount : updatePreparedStatement.executeBatch()) {
- if (updateCount != Statement.SUCCESS_NO_INFO) {
- count = count.isPresent()
- ? count.map(total -> total + updateCount)
- : Optional.of((long) updateCount);
- }
+ /**
+ * @return an optional count of all updated rows or an empty optional if no info is available
+ */
+ private Optional<Long> executeUpdates() throws SQLException {
+ Optional<Long> count = Optional.empty();
+ if (nonNull(updatePreparedStatement)) {
+ try {
+ for (int updateCount : updatePreparedStatement.executeBatch()) {
+ if (updateCount != Statement.SUCCESS_NO_INFO) {
+ count = count.isPresent()
+ ? count.map(total -> total + updateCount)
+ : Optional.of((long) updateCount);
+ }
+ }
+ } catch (SQLException e) {
+ log.error("updatePreparedStatement.executeBatch failed, " +
+ "errCode={}, " +
+ "sqlState={}, " +
+ "error msg={}, " +
+ "cause={}, " +
+ "sql={}",
+ e.getErrorCode(),
+ e.getSQLState(),
+ e.getMessage(),
+ e.getCause(),
+ updatePreparedStatement
+ );
+ throw e;
+ }
}
- }catch(SQLException e){
- log.error("updatePreparedStatement.executeBatch failed, errCode={}, sqlState={}, error msg={}, cause={}, sql={}",
- e.getErrorCode(), e.getSQLState(), e.getMessage(), e.getCause(), updatePreparedStatement);
- throw e;
- }
+ return count;
}
- return count;
- }
- private long executeDeletes() throws SQLException {
- long totalDeleteCount = 0;
- if (nonNull(deletePreparedStatement)) {
- for (int updateCount : deletePreparedStatement.executeBatch()) {
- if (updateCount != Statement.SUCCESS_NO_INFO) {
- totalDeleteCount += updateCount;
+ private long executeDeletes() throws SQLException {
+ long totalDeleteCount = 0;
+ if (nonNull(deletePreparedStatement)) {
+ for (int updateCount : deletePreparedStatement.executeBatch()) {
+ if (updateCount != Statement.SUCCESS_NO_INFO) {
+ totalDeleteCount += updateCount;
+ }
+ }
}
- }
+ return totalDeleteCount;
}
- return totalDeleteCount;
- }
-
- private long updateRecordCount() {
- return records
- .stream()
- // ignore deletes
- .filter(record -> nonNull(record.getData()) || !config.isDeleteEnabled())
- .count();
- }
- public void close() throws SQLException {
- log.debug(
- "Closing BufferedRecords with updatePreparedStatement: {} deletePreparedStatement: {}",
- updatePreparedStatement,
- deletePreparedStatement
- );
- if (nonNull(updatePreparedStatement)) {
- updatePreparedStatement.close();
- updatePreparedStatement = null;
+ private long updateRecordCount() {
+ return records
+ .stream()
+ // ignore deletes
+ .filter(record -> nonNull(record.getData()) || !config.isDeleteEnabled())
+ .count();
}
- if (nonNull(deletePreparedStatement)) {
- deletePreparedStatement.close();
- deletePreparedStatement = null;
- }
- }
- private String getInsertSql() throws SQLException {
- switch (config.getInsertMode()) {
- case INSERT:
- return dbDialect.buildInsertStatement(
- tableId,
- asColumns(fieldsMetadata.keyFieldNames),
- asColumns(fieldsMetadata.nonKeyFieldNames)
+ public void close() throws SQLException {
+ log.debug(
+ "Closing BufferedRecords with updatePreparedStatement: {} deletePreparedStatement: {}",
+ updatePreparedStatement,
+ deletePreparedStatement
);
- case UPSERT:
- if (fieldsMetadata.keyFieldNames.isEmpty()) {
- throw new ConnectException(String.format(
- "Write to table '%s' in UPSERT mode requires key field names to be known, check the"
- + " primary key configuration",
- tableId
- ));
+ if (nonNull(updatePreparedStatement)) {
+ updatePreparedStatement.close();
+ updatePreparedStatement = null;
}
- try {
- return dbDialect.buildUpsertQueryStatement(
- tableId,
- asColumns(fieldsMetadata.keyFieldNames),
- asColumns(fieldsMetadata.nonKeyFieldNames)
- );
- } catch (UnsupportedOperationException e) {
- throw new ConnectException(String.format(
- "Write to table '%s' in UPSERT mode is not supported with the %s dialect.",
- tableId,
- dbDialect.name()
- ));
+ if (nonNull(deletePreparedStatement)) {
+ deletePreparedStatement.close();
+ deletePreparedStatement = null;
}
- case UPDATE:
- return dbDialect.buildUpdateStatement(
- tableId,
- asColumns(fieldsMetadata.keyFieldNames),
- asColumns(fieldsMetadata.nonKeyFieldNames)
- );
- default:
- throw new ConnectException("Invalid insert mode");
}
- }
- private String getDeleteSql() {
- String sql = null;
- if (config.isDeleteEnabled()) {
- switch (config.pkMode) {
- case RECORD_KEY:
- if (fieldsMetadata.keyFieldNames.isEmpty()) {
- throw new ConnectException("Require primary keys to support delete");
- }
- try {
- sql = dbDialect.buildDeleteStatement(
- tableId,
- asColumns(fieldsMetadata.keyFieldNames)
- );
- } catch (UnsupportedOperationException e) {
- throw new ConnectException(String.format(
- "Deletes to table '%s' are not supported with the %s dialect.",
- tableId,
- dbDialect.name()
- ));
- }
- break;
- default:
- throw new ConnectException("Deletes are only supported for pk.mode record_key");
- }
+ private String getInsertSql() {
+ switch (config.getInsertMode()) {
+ case INSERT:
+ return dbDialect.buildInsertStatement(
+ tableId,
+ asColumns(fieldsMetadata.keyFieldNames),
+ asColumns(fieldsMetadata.nonKeyFieldNames)
+ );
+ case UPSERT:
+ if (fieldsMetadata.keyFieldNames.isEmpty()) {
+ throw new ConnectException(String.format(
+ "Write to table '%s' in UPSERT mode requires key field names to be known, check the"
+ + " primary key configuration",
+ tableId
+ ));
+ }
+ try {
+ return dbDialect.buildUpsertQueryStatement(
+ tableId,
+ asColumns(fieldsMetadata.keyFieldNames),
+ asColumns(fieldsMetadata.nonKeyFieldNames)
+ );
+ } catch (UnsupportedOperationException e) {
+ throw new ConnectException(String.format(
+ "Write to table '%s' in UPSERT mode is not supported with the %s dialect.",
+ tableId,
+ dbDialect.name()
+ ));
+ }
+ case UPDATE:
+ return dbDialect.buildUpdateStatement(
+ tableId,
+ asColumns(fieldsMetadata.keyFieldNames),
+ asColumns(fieldsMetadata.nonKeyFieldNames)
+ );
+ default:
+ throw new ConnectException("Invalid insert mode");
+ }
}
- return sql;
- }
- private Collection<ColumnId> asColumns(Collection<String> names) {
- return names.stream()
- .map(name -> new ColumnId(tableId, name))
- .collect(Collectors.toList());
- }
+ private String getDeleteSql() {
+ String sql = null;
+ if (config.isDeleteEnabled()) {
+ switch (config.pkMode) {
+ case RECORD_KEY:
+ if (fieldsMetadata.keyFieldNames.isEmpty()) {
+ throw new ConnectException("Require primary keys to support delete");
+ }
+ try {
+ sql = dbDialect.buildDeleteStatement(
+ tableId,
+ asColumns(fieldsMetadata.keyFieldNames)
+ );
+ } catch (UnsupportedOperationException e) {
+ throw new ConnectException(String.format(
+ "Deletes to table '%s' are not supported with the %s dialect.",
+ tableId,
+ dbDialect.name()
+ ));
+ }
+ break;
+ default:
+ throw new ConnectException("Deletes are only supported for pk.mode record_key");
+ }
+ }
+ return sql;
+ }
+
+ private Collection<ColumnId> asColumns(Collection<String> names) {
+ return names.stream()
+ .map(name -> new ColumnId(tableId, name))
+ .collect(Collectors.toList());
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/Updater.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/Updater.java
index 8f146dc..dcd22f9 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/Updater.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/Updater.java
@@ -37,69 +37,69 @@ import java.util.Map;
*/
public class Updater {
- private static final Logger log = LoggerFactory.getLogger(Updater.class);
- private final JdbcSinkConfig config;
- private final DatabaseDialect dbDialect;
- private final DbStructure dbStructure;
- final CachedConnectionProvider cachedConnectionProvider;
+ private static final Logger log = LoggerFactory.getLogger(Updater.class);
+ private final JdbcSinkConfig config;
+ private final DatabaseDialect dbDialect;
+ private final DbStructure dbStructure;
+ final CachedConnectionProvider cachedConnectionProvider;
- public Updater(final JdbcSinkConfig config, DatabaseDialect dbDialect, DbStructure dbStructure) {
- this.config = config;
- this.dbDialect = dbDialect;
- this.dbStructure = dbStructure;
+ public Updater(final JdbcSinkConfig config, DatabaseDialect dbDialect, DbStructure dbStructure) {
+ this.config = config;
+ this.dbDialect = dbDialect;
+ this.dbStructure = dbStructure;
- this.cachedConnectionProvider = connectionProvider(
- config.getAttempts(),
- config.getRetryBackoffMs()
- );
- }
+ this.cachedConnectionProvider = connectionProvider(
+ config.getAttempts(),
+ config.getRetryBackoffMs()
+ );
+ }
- protected CachedConnectionProvider connectionProvider(int maxConnAttempts, long retryBackoff) {
- return new CachedConnectionProvider(this.dbDialect, maxConnAttempts, retryBackoff) {
- @Override
- protected void onConnect(final Connection connection) throws SQLException {
- connection.setAutoCommit(false);
- }
- };
- }
+ protected CachedConnectionProvider connectionProvider(int maxConnAttempts, long retryBackoff) {
+ return new CachedConnectionProvider(this.dbDialect, maxConnAttempts, retryBackoff) {
+ @Override
+ protected void onConnect(final Connection connection) throws SQLException {
+ connection.setAutoCommit(false);
+ }
+ };
+ }
- public void write(final Collection<ConnectRecord> records)
- throws SQLException, TableAlterOrCreateException {
- final Connection connection = cachedConnectionProvider.getConnection();
- try {
- final Map<TableId, BufferedRecords> bufferByTable = new HashMap<>();
- for (ConnectRecord record : records) {
- // destination table
- final TableId tableId = destinationTable(record);
- if (!config.filterWhiteTable(dbDialect,tableId)){
- continue;
- }
- BufferedRecords buffer = bufferByTable.get(tableId);
- if (buffer == null) {
- buffer = new BufferedRecords(config, tableId, dbDialect, dbStructure, connection);
- bufferByTable.put(tableId, buffer);
+ public void write(final Collection<ConnectRecord> records)
+ throws SQLException, TableAlterOrCreateException {
+ final Connection connection = cachedConnectionProvider.getConnection();
+ try {
+ final Map<TableId, BufferedRecords> bufferByTable = new HashMap<>();
+ for (ConnectRecord record : records) {
+ // destination table
+ final TableId tableId = destinationTable(record);
+ if (!config.filterWhiteTable(dbDialect, tableId)) {
+ continue;
+ }
+ BufferedRecords buffer = bufferByTable.get(tableId);
+ if (buffer == null) {
+ buffer = new BufferedRecords(config, tableId, dbDialect, dbStructure, connection);
+ bufferByTable.put(tableId, buffer);
+ }
+ buffer.add(record);
+ }
+ for (Map.Entry<TableId, BufferedRecords> entry : bufferByTable.entrySet()) {
+ TableId tableId = entry.getKey();
+ BufferedRecords buffer = entry.getValue();
+ log.debug("Flushing records in JDBC Writer for table ID: {}", tableId);
+ buffer.flush();
+ buffer.close();
+ }
+ connection.commit();
+ } catch (SQLException | TableAlterOrCreateException e) {
+ connection.rollback();
}
- buffer.add(record);
- }
- for (Map.Entry<TableId, BufferedRecords> entry : bufferByTable.entrySet()) {
- TableId tableId = entry.getKey();
- BufferedRecords buffer = entry.getValue();
- log.debug("Flushing records in JDBC Writer for table ID: {}", tableId);
- buffer.flush();
- buffer.close();
- }
- connection.commit();
- } catch (SQLException | TableAlterOrCreateException e) {
- connection.rollback();
}
- }
- public void closeQuietly() {
- cachedConnectionProvider.close();
- }
+ public void closeQuietly() {
+ cachedConnectionProvider.close();
+ }
- TableId destinationTable(ConnectRecord record) {
- // todo table from header
- return dbDialect.parseToTableId(record.getSchema().getName());
- }
+ TableId destinationTable(ConnectRecord record) {
+ // todo table from header
+ return dbDialect.parseToTableId(record.getSchema().getName());
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/FieldsMetadata.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/FieldsMetadata.java
index d6f224d..d185218 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/FieldsMetadata.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/FieldsMetadata.java
@@ -38,202 +38,204 @@ import java.util.Set;
*/
public class FieldsMetadata {
- public final Set<String> keyFieldNames;
- public final Set<String> nonKeyFieldNames;
- public final Map<String, SinkRecordField> allFields;
-
- // visible for testing
- public FieldsMetadata(
- Set<String> keyFieldNames,
- Set<String> nonKeyFieldNames,
- Map<String, SinkRecordField> allFields
- ) {
- boolean fieldCountsMatch = (keyFieldNames.size() + nonKeyFieldNames.size() == allFields.size());
- boolean allFieldsContained = (allFields.keySet().containsAll(keyFieldNames)
- && allFields.keySet().containsAll(nonKeyFieldNames));
- if (!fieldCountsMatch || !allFieldsContained) {
- throw new IllegalArgumentException(String.format(
- "Validation fail -- keyFieldNames:%s nonKeyFieldNames:%s allFields:%s",
- keyFieldNames, nonKeyFieldNames, allFields
- ));
- }
- this.keyFieldNames = keyFieldNames;
- this.nonKeyFieldNames = nonKeyFieldNames;
- this.allFields = allFields;
- }
-
- public static FieldsMetadata extract(
- final String tableName,
- final JdbcSinkConfig.PrimaryKeyMode pkMode,
- final List<String> configuredPkFields,
- final Set<String> fieldsWhitelist,
- final SchemaPair schemaPair
- ) {
- return extract(
- tableName,
- pkMode,
- configuredPkFields,
- fieldsWhitelist,
- schemaPair.schema,
- schemaPair.extensions
- );
- }
-
- /**
- * extract metadata info
- * @param tableName
- * @param pkMode
- * @param configuredPkFields
- * @param fieldsWhitelist
- * @param schema
- * @param headers
- * @return
- */
- public static FieldsMetadata extract(
- final String tableName,
- final JdbcSinkConfig.PrimaryKeyMode pkMode,
- final List<String> configuredPkFields,
- final Set<String> fieldsWhitelist,
- final Schema schema,
- final KeyValue headers
- ) {
- if (schema != null && schema.getFieldType() != FieldType.STRUCT) {
- throw new ConnectException("Value schema must be of type Struct");
+ public final Set<String> keyFieldNames;
+ public final Set<String> nonKeyFieldNames;
+ public final Map<String, SinkRecordField> allFields;
+
+ // visible for testing
+ public FieldsMetadata(
+ Set<String> keyFieldNames,
+ Set<String> nonKeyFieldNames,
+ Map<String, SinkRecordField> allFields
+ ) {
+ boolean fieldCountsMatch = (keyFieldNames.size() + nonKeyFieldNames.size()) == allFields.size();
+ boolean allFieldsContained = allFields.keySet().containsAll(keyFieldNames)
+ && allFields.keySet().containsAll(nonKeyFieldNames);
+ if (!fieldCountsMatch || !allFieldsContained) {
+ throw new IllegalArgumentException(String.format(
+ "Validation fail -- keyFieldNames:%s nonKeyFieldNames:%s allFields:%s",
+ keyFieldNames, nonKeyFieldNames, allFields
+ ));
+ }
+ this.keyFieldNames = keyFieldNames;
+ this.nonKeyFieldNames = nonKeyFieldNames;
+ this.allFields = allFields;
}
- final Map<String, SinkRecordField> allFields = new HashMap<>();
- final Set<String> keyFieldNames = new LinkedHashSet<>();
- switch (pkMode) {
- case NONE:
- break;
- case RECORD_VALUE:
- extractRecordValuePk(tableName, configuredPkFields, schema, headers, allFields, keyFieldNames);
- break;
- default:
- throw new ConnectException("Unknown primary key mode: " + pkMode);
+
+ public static FieldsMetadata extract(
+ final String tableName,
+ final JdbcSinkConfig.PrimaryKeyMode pkMode,
+ final List<String> configuredPkFields,
+ final Set<String> fieldsWhitelist,
+ final SchemaPair schemaPair
+ ) {
+ return extract(
+ tableName,
+ pkMode,
+ configuredPkFields,
+ fieldsWhitelist,
+ schemaPair.schema,
+ schemaPair.extensions
+ );
}
- final Set<String> nonKeyFieldNames = new LinkedHashSet<>();
- if (schema != null) {
- for (Field field : schema.getFields()) {
- if (keyFieldNames.contains(field.getName())) {
- continue;
+
+ /**
+ * extract metadata info
+ *
+ * @param tableName
+ * @param pkMode
+ * @param configuredPkFields
+ * @param fieldsWhitelist
+ * @param schema
+ * @param headers
+ * @return
+ */
+ public static FieldsMetadata extract(
+ final String tableName,
+ final JdbcSinkConfig.PrimaryKeyMode pkMode,
+ final List<String> configuredPkFields,
+ final Set<String> fieldsWhitelist,
+ final Schema schema,
+ final KeyValue headers
+ ) {
+ if (schema != null && schema.getFieldType() != FieldType.STRUCT) {
+ throw new ConnectException("Value schema must be of type Struct");
}
- if (!fieldsWhitelist.isEmpty() && !fieldsWhitelist.contains(field.getName())) {
- continue;
+ final Map<String, SinkRecordField> allFields = new HashMap<>();
+ final Set<String> keyFieldNames = new LinkedHashSet<>();
+ switch (pkMode) {
+ case NONE:
+ break;
+ case RECORD_VALUE:
+ extractRecordValuePk(tableName, configuredPkFields, schema, headers, allFields, keyFieldNames);
+ break;
+ default:
+ throw new ConnectException("Unknown primary key mode: " + pkMode);
+ }
+ final Set<String> nonKeyFieldNames = new LinkedHashSet<>();
+ if (schema != null) {
+ for (Field field : schema.getFields()) {
+ if (keyFieldNames.contains(field.getName())) {
+ continue;
+ }
+ if (!fieldsWhitelist.isEmpty() && !fieldsWhitelist.contains(field.getName())) {
+ continue;
+ }
+ nonKeyFieldNames.add(field.getName());
+ final Schema fieldSchema = field.getSchema();
+ allFields.put(field.getName(), new SinkRecordField(fieldSchema, field.getName(), false));
+ }
}
- nonKeyFieldNames.add(field.getName());
- final Schema fieldSchema = field.getSchema();
- allFields.put(field.getName(), new SinkRecordField(fieldSchema, field.getName(), false));
- }
- }
- if (allFields.isEmpty()) {
- throw new ConnectException(
- "No fields found using key and value schemas for table: " + tableName
- );
- }
+ if (allFields.isEmpty()) {
+ throw new ConnectException(
+ "No fields found using key and value schemas for table: " + tableName
+ );
+ }
- final Map<String, SinkRecordField> allFieldsOrdered = new LinkedHashMap<>();
+ final Map<String, SinkRecordField> allFieldsOrdered = new LinkedHashMap<>();
- if (schema != null) {
- for (Field field : schema.getFields()) {
- String fieldName = field.getName();
- if (allFields.containsKey(fieldName)) {
- allFieldsOrdered.put(fieldName, allFields.get(fieldName));
+ if (schema != null) {
+ for (Field field : schema.getFields()) {
+ String fieldName = field.getName();
+ if (allFields.containsKey(fieldName)) {
+ allFieldsOrdered.put(fieldName, allFields.get(fieldName));
+ }
+ }
}
- }
- }
- if (allFieldsOrdered.size() < allFields.size()) {
- ArrayList<String> fieldKeys = new ArrayList<>(allFields.keySet());
- Collections.sort(fieldKeys);
- for (String fieldName : fieldKeys) {
- if (!allFieldsOrdered.containsKey(fieldName)) {
- allFieldsOrdered.put(fieldName, allFields.get(fieldName));
+ if (allFieldsOrdered.size() < allFields.size()) {
+ ArrayList<String> fieldKeys = new ArrayList<>(allFields.keySet());
+ Collections.sort(fieldKeys);
+ for (String fieldName : fieldKeys) {
+ if (!allFieldsOrdered.containsKey(fieldName)) {
+ allFieldsOrdered.put(fieldName, allFields.get(fieldName));
+ }
+ }
}
- }
- }
- return new FieldsMetadata(keyFieldNames, nonKeyFieldNames, allFieldsOrdered);
- }
-
-
- /**
- * record value
- * @param tableName
- * @param configuredPkFields
- * @param valueSchema
- * @param headers
- * @param allFields
- * @param keyFieldNames
- */
- private static void extractRecordValuePk(
- final String tableName,
- final List<String> configuredPkFields,
- final Schema valueSchema,
- final KeyValue headers,
- final Map<String, SinkRecordField> allFields,
- final Set<String> keyFieldNames
- ) {
- if (valueSchema == null) {
- throw new ConnectException(String.format(
- "PK mode for table '%s' is %s, but record value schema is missing",
- tableName,
- JdbcSinkConfig.PrimaryKeyMode.RECORD_VALUE)
- );
+ return new FieldsMetadata(keyFieldNames, nonKeyFieldNames, allFieldsOrdered);
}
- List<String> pkFields = new ArrayList<>(configuredPkFields);
- if (pkFields.isEmpty()) {
- for (Field keyField : valueSchema.getFields()) {
- keyFieldNames.add(keyField.getName());
- }
- } else {
- for (Field keyField : valueSchema.getFields()) {
- keyFieldNames.add(keyField.getName());
- }
- for (String fieldName : pkFields) {
- if (!keyFieldNames.contains(fieldName)) {
- throw new ConnectException(String.format(
- "PK mode for table '%s' is %s with configured PK fields %s, but record value "
- + "schema does not contain field: %s",
- tableName,
- JdbcSinkConfig.PrimaryKeyMode.RECORD_VALUE,
- pkFields,
- fieldName
- ));
+
+
+ /**
+ * record value
+ *
+ * @param tableName
+ * @param configuredPkFields
+ * @param valueSchema
+ * @param headers
+ * @param allFields
+ * @param keyFieldNames
+ */
+ private static void extractRecordValuePk(
+ final String tableName,
+ final List<String> configuredPkFields,
+ final Schema valueSchema,
+ final KeyValue headers,
+ final Map<String, SinkRecordField> allFields,
+ final Set<String> keyFieldNames
+ ) {
+ if (valueSchema == null) {
+ throw new ConnectException(String.format(
+ "PK mode for table '%s' is %s, but record value schema is missing",
+ tableName,
+ JdbcSinkConfig.PrimaryKeyMode.RECORD_VALUE)
+ );
+ }
+ List<String> pkFields = new ArrayList<>(configuredPkFields);
+ if (pkFields.isEmpty()) {
+ for (Field keyField : valueSchema.getFields()) {
+ keyFieldNames.add(keyField.getName());
+ }
+ } else {
+ for (Field keyField : valueSchema.getFields()) {
+ keyFieldNames.add(keyField.getName());
+ }
+ for (String fieldName : pkFields) {
+ if (!keyFieldNames.contains(fieldName)) {
+ throw new ConnectException(String.format(
+ "PK mode for table '%s' is %s with configured PK fields %s, but record value "
+ + "schema does not contain field: %s",
+ tableName,
+ JdbcSinkConfig.PrimaryKeyMode.RECORD_VALUE,
+ pkFields,
+ fieldName
+ ));
+ }
+ }
+ keyFieldNames.addAll(pkFields);
}
- }
- keyFieldNames.addAll(pkFields);
+ for (String fieldName : keyFieldNames) {
+ final Schema fieldSchema = valueSchema.getField(fieldName).getSchema();
+ allFields.put(fieldName, new SinkRecordField(fieldSchema, fieldName, true));
+ }
+
}
- for (String fieldName : keyFieldNames) {
- final Schema fieldSchema = valueSchema.getField(fieldName).getSchema();
- allFields.put(fieldName, new SinkRecordField(fieldSchema, fieldName, true));
+
+ public static boolean isPrimitive(FieldType type) {
+ switch (type) {
+ case INT8:
+ case INT32:
+ case INT64:
+ case FLOAT32:
+ case FLOAT64:
+ case BOOLEAN:
+ case STRING:
+ case BYTES:
+ return true;
+ default:
+ return false;
+ }
}
- }
-
- public static boolean isPrimitive(FieldType type) {
- switch(type) {
- case INT8:
- case INT32:
- case INT64:
- case FLOAT32:
- case FLOAT64:
- case BOOLEAN:
- case STRING:
- case BYTES:
- return true;
- default:
- return false;
+ @Override
+ public String toString() {
+ return "FieldsMetadata{"
+ + "keyFieldNames=" + keyFieldNames
+ + ", nonKeyFieldNames=" + nonKeyFieldNames
+ + ", allFields=" + allFields
+ + '}';
}
- }
-
- @Override
- public String toString() {
- return "FieldsMetadata{"
- + "keyFieldNames=" + keyFieldNames
- + ", nonKeyFieldNames=" + nonKeyFieldNames
- + ", allFields=" + allFields
- + '}';
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SchemaPair.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SchemaPair.java
index e499520..77955d7 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SchemaPair.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SchemaPair.java
@@ -25,38 +25,38 @@ import java.util.Objects;
* schema pair
*/
public class SchemaPair {
- public final Schema schema;
- public final KeyValue extensions;
-
- public SchemaPair(Schema schema) {
- this.schema = schema;
- this.extensions = null;
- }
-
- public SchemaPair(Schema schema, KeyValue extensions) {
- this.schema = schema;
- this.extensions = extensions;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
+ public final Schema schema;
+ public final KeyValue extensions;
+
+ public SchemaPair(Schema schema) {
+ this.schema = schema;
+ this.extensions = null;
+ }
+
+ public SchemaPair(Schema schema, KeyValue extensions) {
+ this.schema = schema;
+ this.extensions = extensions;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SchemaPair that = (SchemaPair) o;
+ return Objects.equals(schema, that.schema);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(schema);
}
- if (o == null || getClass() != o.getClass()) {
- return false;
+
+ @Override
+ public String toString() {
+ return String.format("<SchemaPair: %s>", schema);
}
- SchemaPair that = (SchemaPair) o;
- return Objects.equals(schema, that.schema);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(schema);
- }
-
- @Override
- public String toString() {
- return String.format("<SchemaPair: %s>", schema);
- }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SinkRecordField.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SinkRecordField.java
index f06eabb..fef7bff 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SinkRecordField.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/sink/metadata/SinkRecordField.java
@@ -19,76 +19,74 @@ package org.apache.rocketmq.connect.jdbc.sink.metadata;
import io.openmessaging.connector.api.data.FieldType;
import io.openmessaging.connector.api.data.Schema;
-/**
- * @author xiaoyi
- */
+
public class SinkRecordField {
- private final Schema schema;
- private final String name;
- private final boolean isPrimaryKey;
+ private final Schema schema;
+ private final String name;
+ private final boolean isPrimaryKey;
- public SinkRecordField(Schema schema, String name, boolean isPrimaryKey) {
- this.schema = schema;
- this.name = name;
- this.isPrimaryKey = isPrimaryKey;
- }
+ public SinkRecordField(Schema schema, String name, boolean isPrimaryKey) {
+ this.schema = schema;
+ this.name = name;
+ this.isPrimaryKey = isPrimaryKey;
+ }
- public Schema schema() {
- return schema;
- }
+ public Schema schema() {
+ return schema;
+ }
- public String schemaName() {
- return schema.getName();
- }
+ public String schemaName() {
+ return schema.getName();
+ }
- public FieldType schemaType() {
- return schema.getFieldType();
- }
+ public FieldType schemaType() {
+ return schema.getFieldType();
+ }
- public String name() {
- return name;
- }
+ public String name() {
+ return name;
+ }
- public boolean isOptional() {
- return !isPrimaryKey;
- }
+ public boolean isOptional() {
+ return !isPrimaryKey;
+ }
- public Object defaultValue() {
+ public Object defaultValue() {
// return schema.defaultValue();
- return null;
- }
-
- public boolean isPrimaryKey() {
- return isPrimaryKey;
- }
-
-
- public Object getDefaultValue(FieldType type){
- switch (type){
- case BOOLEAN:
- case INT8:
- return (byte)0;
- case INT32:
- return 0;
- case INT64:
- return 0L;
- case FLOAT32:
- return 0.0f;
- case FLOAT64:
- return 0.0d;
- default:
return null;
}
- }
+ public boolean isPrimaryKey() {
+ return isPrimaryKey;
+ }
+
+
+ public Object getDefaultValue(FieldType type) {
+ switch (type) {
+ case BOOLEAN:
+ case INT8:
+ return (byte) 0;
+ case INT32:
+ return 0;
+ case INT64:
+ return 0L;
+ case FLOAT32:
+ return 0.0f;
+ case FLOAT64:
+ return 0.0d;
+ default:
+ return null;
+ }
- @Override
- public String toString() {
- return "SinkRecordField{"
- + "schema=" + schema
- + ", name='" + name + '\''
- + ", isPrimaryKey=" + isPrimaryKey
- + '}';
- }
+ }
+
+ @Override
+ public String toString() {
+ return "SinkRecordField{"
+ + "schema=" + schema
+ + ", name='" + name + '\''
+ + ", isPrimaryKey=" + isPrimaryKey
+ + '}';
+ }
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/source/TimestampIncrementingCriteria.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/source/TimestampIncrementingCriteria.java
index 24057ce..fcd612b 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/source/TimestampIncrementingCriteria.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/source/TimestampIncrementingCriteria.java
@@ -37,310 +37,310 @@ import java.util.stream.Collectors;
public class TimestampIncrementingCriteria {
- /**
- * The values that can be used in a statement's WHERE clause.
- */
- public interface CriteriaValues {
-
- /**
- * Get the beginning of the time period.
- *
- * @return the beginning timestamp; may be null
- * @throws SQLException if there is a problem accessing the value
- */
- Timestamp beginTimestampValue() throws SQLException;
-
- /**
- * Get the end of the time period.
- *
- * @return the ending timestamp; never null
- * @throws SQLException if there is a problem accessing the value
- */
- Timestamp endTimestampValue(Timestamp beginTime) throws SQLException;
-
/**
- * Get the last incremented value seen.
- *
- * @return the last incremented value from one of the rows
- * @throws SQLException if there is a problem accessing the value
+ * The values that can be used in a statement's WHERE clause.
*/
- Long lastIncrementedValue() throws SQLException;
- }
+ public interface CriteriaValues {
- protected static final BigDecimal LONG_MAX_VALUE_AS_BIGDEC = new BigDecimal(Long.MAX_VALUE);
+ /**
+ * Get the beginning of the time period.
+ *
+ * @return the beginning timestamp; may be null
+ * @throws SQLException if there is a problem accessing the value
+ */
+ Timestamp beginTimestampValue() throws SQLException;
- protected final Logger log = LoggerFactory.getLogger(getClass());
- protected final List<ColumnId> timestampColumns;
- protected final ColumnId incrementingColumn;
- protected final TimeZone timeZone;
+ /**
+ * Get the end of the time period.
+ *
+ * @return the ending timestamp; never null
+ * @throws SQLException if there is a problem accessing the value
+ */
+ Timestamp endTimestampValue(Timestamp beginTime) throws SQLException;
- public TimestampIncrementingCriteria(
- ColumnId incrementingColumn,
- List<ColumnId> timestampColumns,
- TimeZone timeZone
- ) {
- this.timestampColumns =
- timestampColumns != null ? timestampColumns : Collections.<ColumnId>emptyList();
- this.incrementingColumn = incrementingColumn;
- this.timeZone = timeZone;
- }
+ /**
+ * Get the last incremented value seen.
+ *
+ * @return the last incremented value from one of the rows
+ * @throws SQLException if there is a problem accessing the value
+ */
+ Long lastIncrementedValue() throws SQLException;
+ }
- protected boolean hasTimestampColumns() {
- return !timestampColumns.isEmpty();
- }
+ protected static final BigDecimal LONG_MAX_VALUE_AS_BIGDEC = new BigDecimal(Long.MAX_VALUE);
- protected boolean hasIncrementedColumn() {
- return incrementingColumn != null;
- }
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+ protected final List<ColumnId> timestampColumns;
+ protected final ColumnId incrementingColumn;
+ protected final TimeZone timeZone;
- /**
- * Build the WHERE clause for the columns used in this criteria.
- *
- * @param builder the string builder to which the WHERE clause should be appended; never null
- */
- public void whereClause(ExpressionBuilder builder) {
- if (hasTimestampColumns() && hasIncrementedColumn()) {
- timestampIncrementingWhereClause(builder);
- } else if (hasTimestampColumns()) {
- timestampWhereClause(builder);
- } else if (hasIncrementedColumn()) {
- incrementingWhereClause(builder);
+ public TimestampIncrementingCriteria(
+ ColumnId incrementingColumn,
+ List<ColumnId> timestampColumns,
+ TimeZone timeZone
+ ) {
+ this.timestampColumns =
+ timestampColumns != null ? timestampColumns : Collections.<ColumnId>emptyList();
+ this.incrementingColumn = incrementingColumn;
+ this.timeZone = timeZone;
}
- }
- /**
- * Set the query parameters on the prepared statement whose WHERE clause was generated with the
- * previous call to {@link #whereClause(ExpressionBuilder)}.
- *
- * @param stmt the prepared statement; never null
- * @param values the values that can be used in the criteria parameters; never null
- * @throws SQLException if there is a problem using the prepared statement
- */
- public void setQueryParameters(
- PreparedStatement stmt,
- CriteriaValues values
- ) throws SQLException {
- if (hasTimestampColumns() && hasIncrementedColumn()) {
- setQueryParametersTimestampIncrementing(stmt, values);
- } else if (hasTimestampColumns()) {
- setQueryParametersTimestamp(stmt, values);
- } else if (hasIncrementedColumn()) {
- setQueryParametersIncrementing(stmt, values);
+ protected boolean hasTimestampColumns() {
+ return !timestampColumns.isEmpty();
}
- }
- protected void setQueryParametersTimestampIncrementing(
- PreparedStatement stmt,
- CriteriaValues values
- ) throws SQLException{
- Timestamp beginTime = values.beginTimestampValue();
- Timestamp endTime = values.endTimestampValue(beginTime);
- Long incOffset = values.lastIncrementedValue();
- stmt.setTimestamp(1, endTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
- stmt.setTimestamp(2, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
- stmt.setLong(3, incOffset);
- stmt.setTimestamp(4, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
- log.debug(
- "Executing prepared statement with start time value = {} end time = {} and incrementing"
- + " value = {}", DateTimeUtils.formatTimestamp(beginTime, timeZone),
- DateTimeUtils.formatTimestamp(endTime, timeZone), incOffset
- );
- }
+ protected boolean hasIncrementedColumn() {
+ return incrementingColumn != null;
+ }
- protected void setQueryParametersIncrementing(
- PreparedStatement stmt,
- CriteriaValues values
- ) throws SQLException {
- Long incOffset = values.lastIncrementedValue();
- stmt.setLong(1, incOffset);
- log.debug("Executing prepared statement with incrementing value = {}", incOffset);
- }
+ /**
+ * Build the WHERE clause for the columns used in this criteria.
+ *
+ * @param builder the string builder to which the WHERE clause should be appended; never null
+ */
+ public void whereClause(ExpressionBuilder builder) {
+ if (hasTimestampColumns() && hasIncrementedColumn()) {
+ timestampIncrementingWhereClause(builder);
+ } else if (hasTimestampColumns()) {
+ timestampWhereClause(builder);
+ } else if (hasIncrementedColumn()) {
+ incrementingWhereClause(builder);
+ }
+ }
- protected void setQueryParametersTimestamp(
- PreparedStatement stmt,
- CriteriaValues values
- ) throws SQLException {
- Timestamp beginTime = values.beginTimestampValue();
- Timestamp endTime = values.endTimestampValue(beginTime);
- stmt.setTimestamp(1, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
- stmt.setTimestamp(2, endTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
- log.debug("Executing prepared statement with timestamp value = {} end time = {}",
- DateTimeUtils.formatTimestamp(beginTime, timeZone),
- DateTimeUtils.formatTimestamp(endTime, timeZone)
- );
- }
+ /**
+ * Set the query parameters on the prepared statement whose WHERE clause was generated with the
+ * previous call to {@link #whereClause(ExpressionBuilder)}.
+ *
+ * @param stmt the prepared statement; never null
+ * @param values the values that can be used in the criteria parameters; never null
+ * @throws SQLException if there is a problem using the prepared statement
+ */
+ public void setQueryParameters(
+ PreparedStatement stmt,
+ CriteriaValues values
+ ) throws SQLException {
+ if (hasTimestampColumns() && hasIncrementedColumn()) {
+ setQueryParametersTimestampIncrementing(stmt, values);
+ } else if (hasTimestampColumns()) {
+ setQueryParametersTimestamp(stmt, values);
+ } else if (hasIncrementedColumn()) {
+ setQueryParametersIncrementing(stmt, values);
+ }
+ }
- /**
- * Extract the offset values from the row.
- *
- * @param schema the record's schema; never null
- * @param record the record's struct; never null
- * @param previousOffset a previous timestamp offset if the table has timestamp columns
- * @return the timestamp for this row; may not be null
- */
- public TimestampIncrementingOffset extractValues(
- Schema schema,
- Object[] record,
- TimestampIncrementingOffset previousOffset
- ) {
- Timestamp extractedTimestamp = null;
- if (hasTimestampColumns()) {
- extractedTimestamp = extractOffsetTimestamp(schema, record);
- assert previousOffset == null || (previousOffset.getTimestampOffset() != null
- && previousOffset.getTimestampOffset().compareTo(
- extractedTimestamp) <= 0
- );
+ protected void setQueryParametersTimestampIncrementing(
+ PreparedStatement stmt,
+ CriteriaValues values
+ ) throws SQLException {
+ Timestamp beginTime = values.beginTimestampValue();
+ Timestamp endTime = values.endTimestampValue(beginTime);
+ Long incOffset = values.lastIncrementedValue();
+ stmt.setTimestamp(1, endTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ stmt.setTimestamp(2, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ stmt.setLong(3, incOffset);
+ stmt.setTimestamp(4, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ log.debug(
+ "Executing prepared statement with start time value = {} end time = {} and incrementing"
+ + " value = {}", DateTimeUtils.formatTimestamp(beginTime, timeZone),
+ DateTimeUtils.formatTimestamp(endTime, timeZone), incOffset
+ );
}
- Long extractedId = null;
- if (hasIncrementedColumn()) {
- extractedId = extractOffsetIncrementedId(schema, record);
- // If we are only using an incrementing column, then this must be incrementing.
- // If we are also using a timestamp, then we may see updates to older rows.
- assert previousOffset == null || previousOffset.getIncrementingOffset() == -1L
- || extractedId > previousOffset.getIncrementingOffset() || hasTimestampColumns();
+
+ protected void setQueryParametersIncrementing(
+ PreparedStatement stmt,
+ CriteriaValues values
+ ) throws SQLException {
+ Long incOffset = values.lastIncrementedValue();
+ stmt.setLong(1, incOffset);
+ log.debug("Executing prepared statement with incrementing value = {}", incOffset);
}
- return new TimestampIncrementingOffset(extractedTimestamp, extractedId);
- }
- /**
- * Extract the timestamp from the row.
- *
- * @param schema the record's schema; never null
- * @param record the record's struct; never null
- * @return the timestamp for this row; may not be null
- */
- protected Timestamp extractOffsetTimestamp(
- Schema schema,
- Object[] record
- ) {
- for (ColumnId timestampColumn : timestampColumns) {
- Field field=schema.getField(timestampColumn.name());
- Timestamp ts = (Timestamp) record[field.getIndex()];
- if (ts != null) {
- return ts;
- }
+ protected void setQueryParametersTimestamp(
+ PreparedStatement stmt,
+ CriteriaValues values
+ ) throws SQLException {
+ Timestamp beginTime = values.beginTimestampValue();
+ Timestamp endTime = values.endTimestampValue(beginTime);
+ stmt.setTimestamp(1, beginTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ stmt.setTimestamp(2, endTime, DateTimeUtils.getTimeZoneCalendar(timeZone));
+ log.debug("Executing prepared statement with timestamp value = {} end time = {}",
+ DateTimeUtils.formatTimestamp(beginTime, timeZone),
+ DateTimeUtils.formatTimestamp(endTime, timeZone)
... 4121 lines suppressed ...