This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 333e6e98d [INLONG-4243][Sort] Add Oracle data extract support (#4382)
333e6e98d is described below
commit 333e6e98d07ee2d08e1e01e30036ef7819e2ea11
Author: yunqingmoswu <[email protected]>
AuthorDate: Thu May 26 15:21:22 2022 +0800
[INLONG-4243][Sort] Add Oracle data extract support (#4382)
---
.../inlong/sort/protocol/BuiltInFieldInfo.java | 43 ++++++-
.../sort/protocol/constant/OracleConstant.java | 116 +++++++++++++++++
.../inlong/sort/protocol/node/ExtractNode.java | 4 +-
.../org/apache/inlong/sort/protocol/node/Node.java | 7 +-
.../protocol/node/extract/OracleExtractNode.java | 139 +++++++++++++++++++++
.../node/extract/OracleExtractNodeTest.java | 44 +++++++
inlong-sort/sort-connectors/pom.xml | 5 +
.../inlong/sort/parser/impl/FlinkSqlParser.java | 39 ++++++
.../sort/parser/OracleExtractSqlParseTest.java | 128 +++++++++++++++++++
licenses/inlong-sort/LICENSE | 1 +
licenses/inlong-sort/NOTICE | 8 ++
pom.xml | 11 +-
12 files changed, 539 insertions(+), 6 deletions(-)
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
index 81568a8ca..56d728a15 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
@@ -74,6 +74,7 @@ public class BuiltInFieldInfo extends FieldInfo {
/**
* The event time of flink
*/
+ @Deprecated
DATA_TIME,
/**
* The process time of flink
@@ -81,23 +82,63 @@ public class BuiltInFieldInfo extends FieldInfo {
PROCESS_TIME,
/**
* The name of the database containing this Row
+ * It is deprecated and can be replaced by ${@link
BuiltInField#DATABASE_NAME}
+ * and will be removed in a future version.
*/
+ @Deprecated
MYSQL_METADATA_DATABASE,
/**
* The name of the table containing this Row
+ * It is deprecated and can be replaced by ${@link
BuiltInField#TABLE_NAME}
+ * and will be removed in a future version.
*/
+ @Deprecated
MYSQL_METADATA_TABLE,
/**
- * The time when the Row made changes in the database
+ * The time when the Row made changes in the database.
+ * It is deprecated and can be replaced by ${@link BuiltInField#OP_TS}
+ * and will be removed in a future version.
*/
+ @Deprecated
MYSQL_METADATA_EVENT_TIME,
+ /**
+ * Name of the schema that contain the row.
+ */
+ SCHEMA_NAME,
+ /**
+ * Name of the database that contain the row.
+ */
+ DATABASE_NAME,
+ /**
+ * Name of the table that contain the row.
+ */
+ TABLE_NAME,
+ /**
+ * It indicates the time that the change was made in the database.
+ * If the record is read from snapshot of the table instead of the
change stream, the value is always 0
+ */
+ OP_TS,
+ /**
+ * Whether the DDL statement
+ */
+ IS_DDL,
/**
* Whether the DDL statement
+ * It is deprecated and can be replaced by ${@link BuiltInField#IS_DDL}
+ * and will be removed in a future version.
*/
+ @Deprecated
MYSQL_METADATA_IS_DDL,
/**
* Type of database operation, such as INSERT/DELETE, etc.
*/
+ OP_TYPE,
+ /**
+ * Type of database operation, such as INSERT/DELETE, etc.
+ * It is deprecated and can be replaced by ${@link
BuiltInField#OP_TYPE}
+ * and will be removed in a future version.
+ */
+ @Deprecated
MYSQL_METADATA_EVENT_TYPE,
/**
* MySQL binlog data Row
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/constant/OracleConstant.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/constant/OracleConstant.java
new file mode 100644
index 000000000..f8143e9ba
--- /dev/null
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/constant/OracleConstant.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.protocol.constant;
+
+import lombok.Getter;
+
+/**
+ * Oracle options constant
+ */
+public class OracleConstant {
+
+ /**
+ * The key of flink connector defined in flink table
+ */
+ public static final String CONNECTOR = "connector";
+ /**
+ * Specify what flink connector to use for extract data from Oracle
database, here should be 'oracle-cdc'
+ */
+ public static final String ORACLE_CDC = "oracle-cdc";
+ /**
+ * Database name of the Oracle server to monitor
+ */
+ public static final String DATABASE_NAME = "database-name";
+ /**
+ * IP address or hostname of the Oracle database server
+ */
+ public static final String HOSTNAME = "hostname";
+ /**
+ * Integer port number of the Oracle database server.
+ */
+ public static final String PORT = "port";
+ /**
+ * Name of the Oracle database to use when connecting to the Oracle
database server
+ */
+ public static final String USERNAME = "username";
+ /**
+ * Password to use when connecting to the Oracle database server
+ */
+ public static final String PASSWORD = "password";
+ /**
+ * Table name of the Oracle database to monitor
+ */
+ public static final String TABLE_NAME = "table-name";
+ /**
+ * Schema name of the Oracle database to monitor
+ */
+ public static final String SCHEMA_NAME = "schema-name";
+ /**
+ * <p>The mining strategy controls how Oracle LogMiner builds
+ * and uses a given data dictionary for resolving table and column ids to
names.</p>
+ * <p>redo_log_catalog - Writes the data dictionary to the online redo logs
+ * causing more archive logs to be generated over time.
+ * This also enables tracking DDL changes against captured tables,
+ * so if the schema changes frequently this is the ideal choice.</p>
+ * <p>online_catalog - Uses the database’s current data dictionary to
resolve object ids
+ * and does not write any extra information to the online redo logs.
+ * This allows LogMiner to mine substantially faster but at the expense
that DDL changes cannot be tracked.
+ * If the captured table(s) schema changes infrequently or never, this is
the ideal choice.</p>
+ */
+ public static final String LOG_MINING_STRATEGY =
"debezium.log.mining.strategy";
+ /**
+ * If true,CONTINUOUS_MINE option will be added to the log mining session.
+ * This will manage log files switches seamlessly.
+ */
+ public static final String LOG_MINING_CONTINUOUS_MINE =
"debezium.log.mining.continuous.mine";
+ /**
+ * Deprecated: Case insensitive table names;set to 'true' for Oracle
11g,'false'(default) otherwise.
+ */
+ public static final String TABLENAME_CASE_INSENSITIVE =
"debezium.database.tablename.case.insensitive";
+ /**
+ * The key of ${@link ScanStartUpMode}
+ */
+ public static final String SCAN_STARTUP_MODE = "scan.startup.mode";
+
+ /**
+ * Optional startup mode for Oracle CDC consumer,
+ * valid enumerations are "initial" and "latest-offset".
+ * Please see Startup Reading Positionsection for more detailed
information.
+ */
+ @Getter
+ public enum ScanStartUpMode {
+ /**
+ * Performs an initial snapshot on the monitored database tables upon
first startup,
+ * and continue to read the latest binlog.
+ */
+ INITIAL("initial"),
+ /**
+ * Never to perform a snapshot on the monitored database tables upon
first startup,
+ * just read from the change since the connector was started.
+ */
+ LATEST_OFFSET("latest-offset");
+
+ final String value;
+
+ ScanStartUpMode(String value) {
+ this.value = value;
+ }
+ }
+
+}
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java
index 86f5b4d0c..6fa212564 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java
@@ -31,6 +31,7 @@ import
org.apache.inlong.sort.protocol.node.extract.FileSystemExtractNode;
import org.apache.inlong.sort.protocol.node.extract.KafkaExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MongoExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MySqlExtractNode;
+import org.apache.inlong.sort.protocol.node.extract.OracleExtractNode;
import org.apache.inlong.sort.protocol.node.extract.PostgresExtractNode;
import org.apache.inlong.sort.protocol.node.extract.SqlServerExtractNode;
import org.apache.inlong.sort.protocol.transformation.WatermarkField;
@@ -51,8 +52,9 @@ import java.util.Map;
@JsonSubTypes.Type(value = KafkaExtractNode.class, name =
"kafkaExtract"),
@JsonSubTypes.Type(value = PostgresExtractNode.class, name =
"postgresExtract"),
@JsonSubTypes.Type(value = FileSystemExtractNode.class, name =
"fileSystemExtract"),
- @JsonSubTypes.Type(value = MongoExtractNode.class, name =
"MongoExtract"),
+ @JsonSubTypes.Type(value = MongoExtractNode.class, name =
"mongoExtract"),
@JsonSubTypes.Type(value = SqlServerExtractNode.class, name =
"sqlserverExtract"),
+ @JsonSubTypes.Type(value = OracleExtractNode.class, name =
"oracleExtract")
})
@Data
@NoArgsConstructor
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
index bdd6cddca..99363bc2a 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
@@ -26,6 +26,7 @@ import
org.apache.inlong.sort.protocol.node.extract.FileSystemExtractNode;
import org.apache.inlong.sort.protocol.node.extract.KafkaExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MongoExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MySqlExtractNode;
+import org.apache.inlong.sort.protocol.node.extract.OracleExtractNode;
import org.apache.inlong.sort.protocol.node.extract.PostgresExtractNode;
import org.apache.inlong.sort.protocol.node.load.ClickHouseLoadNode;
import org.apache.inlong.sort.protocol.node.extract.PulsarExtractNode;
@@ -39,6 +40,7 @@ import
org.apache.inlong.sort.protocol.node.load.SqlServerLoadNode;
import org.apache.inlong.sort.protocol.node.transform.DistinctNode;
import org.apache.inlong.sort.protocol.node.transform.TransformNode;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@@ -58,9 +60,10 @@ import java.util.TreeMap;
@JsonSubTypes.Type(value = SqlServerExtractNode.class, name =
"sqlserverExtract"),
@JsonSubTypes.Type(value = PulsarExtractNode.class, name =
"pulsarExtract"),
@JsonSubTypes.Type(value = MongoExtractNode.class, name =
"mongoExtract"),
+ @JsonSubTypes.Type(value = OracleExtractNode.class, name =
"oracleExtract"),
@JsonSubTypes.Type(value = TransformNode.class, name =
"baseTransform"),
- @JsonSubTypes.Type(value = KafkaLoadNode.class, name = "kafkaLoad"),
@JsonSubTypes.Type(value = DistinctNode.class, name = "distinct"),
+ @JsonSubTypes.Type(value = KafkaLoadNode.class, name = "kafkaLoad"),
@JsonSubTypes.Type(value = HiveLoadNode.class, name = "hiveLoad"),
@JsonSubTypes.Type(value = HbaseLoadNode.class, name = "hbaseLoad"),
@JsonSubTypes.Type(value = PostgresLoadNode.class, name =
"postgresLoad"),
@@ -84,7 +87,7 @@ public interface Node {
@JsonInclude(Include.NON_NULL)
default Map<String, String> tableOptions() {
- Map<String, String> options = new TreeMap<>();
+ Map<String, String> options = new LinkedHashMap<>();
if (getProperties() != null && !getProperties().isEmpty()) {
options.putAll(getProperties());
}
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNode.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNode.java
new file mode 100644
index 000000000..6b9a59ee3
--- /dev/null
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNode.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.protocol.node.extract;
+
+import com.google.common.base.Preconditions;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.constant.OracleConstant;
+import org.apache.inlong.sort.protocol.node.ExtractNode;
+import org.apache.inlong.sort.protocol.transformation.WatermarkField;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Oracle extract node for extract data from oracle(Currently support oracle
11,12,19)
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("oracleExtract")
+@Data
+public class OracleExtractNode extends ExtractNode implements Serializable {
+
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Nullable
+ @JsonProperty("primaryKey")
+ private String primaryKey;
+ @Nonnull
+ @JsonProperty("hostname")
+ private String hostname;
+ @Nonnull
+ @JsonProperty("username")
+ private String username;
+ @Nonnull
+ @JsonProperty("password")
+ private String password;
+ @Nonnull
+ @JsonProperty("database")
+ private String database;
+ @Nonnull
+ @JsonProperty("schemaName")
+ private String schemaName;
+ @Nonnull
+ @JsonProperty("tableName")
+ private String tableName;
+ @JsonProperty(value = "port", defaultValue = "1521")
+ private Integer port;
+ @Nullable
+ @JsonProperty("scanStartupMode")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private OracleConstant.ScanStartUpMode scanStartupMode;
+
+ @JsonCreator
+ public OracleExtractNode(@JsonProperty("id") String id,
+ @JsonProperty("name") String name,
+ @JsonProperty("fields") List<FieldInfo> fields,
+ @Nullable @JsonProperty("watermark_field")
WatermarkField watermarkField,
+ @JsonProperty("properties") Map<String, String>
properties,
+ @Nullable @JsonProperty("primaryKey") String
primaryKey,
+ @JsonProperty("hostname") String hostname,
+ @JsonProperty("username") String username,
+ @JsonProperty("password") String password,
+ @JsonProperty("database") String database,
+ @JsonProperty("schemaName") String schemaName,
+ @JsonProperty("tableName") String tableName,
+ @JsonProperty(value = "port", defaultValue =
"1521") Integer port,
+ @Nullable @JsonProperty("scanStartupMode")
+ OracleConstant.ScanStartUpMode scanStartupMode) {
+ super(id, name, fields, watermarkField, properties);
+ this.primaryKey = primaryKey;
+ this.hostname = Preconditions.checkNotNull(hostname, "hostname is
null");
+ this.username = Preconditions.checkNotNull(username, "username is
null");
+ this.password = Preconditions.checkNotNull(password, "password is
null");
+ this.database = Preconditions.checkNotNull(database, "database is
null");
+ this.schemaName = Preconditions.checkNotNull(schemaName, "schemaName
is null");
+ this.tableName = Preconditions.checkNotNull(tableName, "tableName is
null");
+ this.port = port;
+ this.scanStartupMode = scanStartupMode;
+ }
+
+ @Override
+ public Map<String, String> tableOptions() {
+ Map<String, String> options = super.tableOptions();
+ options.put(OracleConstant.CONNECTOR, OracleConstant.ORACLE_CDC);
+ options.put(OracleConstant.HOSTNAME, hostname);
+ options.put(OracleConstant.USERNAME, username);
+ options.put(OracleConstant.PASSWORD, password);
+ options.put(OracleConstant.DATABASE_NAME, database);
+ options.put(OracleConstant.SCHEMA_NAME, schemaName);
+ options.put(OracleConstant.TABLE_NAME, tableName);
+ if (!options.containsKey(OracleConstant.TABLENAME_CASE_INSENSITIVE)) {
+ // Set a default value:false to avoid retrieving unlisted tables
in oracle 11g.
+ // You can set null to replace it if your oracle version is more
than oracle 11g.
+ options.put(OracleConstant.TABLENAME_CASE_INSENSITIVE, "false");
+ }
+ if (!options.containsKey(OracleConstant.LOG_MINING_STRATEGY)) {
+ // Set a default value:online_catalog to improve performance and
reduce latency
+ options.put(OracleConstant.LOG_MINING_STRATEGY, "online_catalog");
+ }
+ if (!options.containsKey(OracleConstant.LOG_MINING_CONTINUOUS_MINE)) {
+ // Set a default value:true to improve performance and reduce
latency
+ options.put(OracleConstant.LOG_MINING_CONTINUOUS_MINE, "true");
+ }
+ if (port != null) {
+ options.put(OracleConstant.PORT, port.toString());
+ }
+ if (scanStartupMode != null) {
+ options.put(OracleConstant.SCAN_STARTUP_MODE,
scanStartupMode.getValue());
+ }
+ return options;
+ }
+
+ @Override
+ public String genTableName() {
+ return String.format("node_%s", super.getId());
+ }
+}
diff --git
a/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNodeTest.java
b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNodeTest.java
new file mode 100644
index 000000000..cc80e8e94
--- /dev/null
+++
b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/extract/OracleExtractNodeTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.protocol.node.extract;
+
+import org.apache.inlong.sort.SerializeBaseTest;
+import org.apache.inlong.sort.formats.common.IntFormatInfo;
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.node.Node;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Test for {@link OracleExtractNode}
+ */
+public class OracleExtractNodeTest extends SerializeBaseTest<Node> {
+
+ @Override
+ public Node getTestObject() {
+ List<FieldInfo> fields = Arrays.asList(
+ new FieldInfo("NAME", new StringFormatInfo()),
+ new FieldInfo("AGE", new IntFormatInfo()));
+ return new OracleExtractNode("1", "oracle_input", fields,
+ null, null, "id", "localhost",
+ "username", "password", "database", "schema_name",
+ "table_name", null, null);
+ }
+}
diff --git a/inlong-sort/sort-connectors/pom.xml
b/inlong-sort/sort-connectors/pom.xml
index de116b316..64103e7bb 100644
--- a/inlong-sort/sort-connectors/pom.xml
+++ b/inlong-sort/sort-connectors/pom.xml
@@ -108,6 +108,11 @@
<groupId>com.ververica</groupId>
<artifactId>flink-connector-mongodb-cdc</artifactId>
</dependency>
+ <!--for oracle-->
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-oracle-cdc</artifactId>
+ </dependency>
<!--for debezium-->
<dependency>
<groupId>io.debezium</groupId>
diff --git
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java
index 6c151d9fa..396b4cc90 100644
---
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java
+++
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java
@@ -36,6 +36,7 @@ import org.apache.inlong.sort.protocol.node.LoadNode;
import org.apache.inlong.sort.protocol.node.Node;
import org.apache.inlong.sort.protocol.node.extract.KafkaExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MySqlExtractNode;
+import org.apache.inlong.sort.protocol.node.extract.OracleExtractNode;
import org.apache.inlong.sort.protocol.node.load.HbaseLoadNode;
import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
import org.apache.inlong.sort.protocol.node.transform.DistinctNode;
@@ -686,6 +687,8 @@ public class FlinkSqlParser implements Parser {
}
if (node instanceof MySqlExtractNode) {
sb.append(parseMySqlExtractNodeMetaField(metaField));
+ } else if (node instanceof OracleExtractNode) {
+ sb.append(parseOracleExtractNodeMetaField(metaField));
} else if (node instanceof KafkaExtractNode) {
sb.append(parseKafkaExtractNodeMetaField(metaField));
} else if (node instanceof KafkaLoadNode) {
@@ -701,21 +704,26 @@ public class FlinkSqlParser implements Parser {
String metaType;
switch (metaField.getBuiltInField()) {
case MYSQL_METADATA_TABLE:
+ case TABLE_NAME:
metaType = "STRING METADATA FROM 'value.table'";
break;
case MYSQL_METADATA_DATABASE:
+ case DATABASE_NAME:
metaType = "STRING METADATA FROM 'value.database'";
break;
case MYSQL_METADATA_EVENT_TIME:
+ case OP_TS:
metaType = "TIMESTAMP(3) METADATA FROM
'value.event-timestamp'";
break;
case MYSQL_METADATA_EVENT_TYPE:
+ case OP_TYPE:
metaType = "STRING METADATA FROM 'value.op-type'";
break;
case MYSQL_METADATA_DATA:
metaType = "STRING METADATA FROM 'value.data'";
break;
case MYSQL_METADATA_IS_DDL:
+ case IS_DDL:
metaType = "BOOLEAN METADATA FROM 'value.is-ddl'";
break;
case METADATA_TS:
@@ -746,9 +754,11 @@ public class FlinkSqlParser implements Parser {
String metaType;
switch (metaField.getBuiltInField()) {
case MYSQL_METADATA_TABLE:
+ case TABLE_NAME:
metaType = "STRING METADATA FROM 'value.table'";
break;
case MYSQL_METADATA_DATABASE:
+ case DATABASE_NAME:
metaType = "STRING METADATA FROM 'value.database'";
break;
case METADATA_SQL_TYPE:
@@ -761,13 +771,16 @@ public class FlinkSqlParser implements Parser {
metaType = "TIMESTAMP_LTZ(3) METADATA FROM
'value.ingestion-timestamp'";
break;
case MYSQL_METADATA_EVENT_TIME:
+ case OP_TS:
metaType = "TIMESTAMP_LTZ(3) METADATA FROM
'value.event-timestamp'";
break;
// additional metadata
case MYSQL_METADATA_EVENT_TYPE:
+ case OP_TYPE:
metaType = "STRING METADATA FROM 'value.op-type'";
break;
case MYSQL_METADATA_IS_DDL:
+ case IS_DDL:
metaType = "BOOLEAN METADATA FROM 'value.is-ddl'";
break;
case METADATA_MYSQL_TYPE:
@@ -789,21 +802,26 @@ public class FlinkSqlParser implements Parser {
String metaType;
switch (metaField.getBuiltInField()) {
case MYSQL_METADATA_TABLE:
+ case TABLE_NAME:
metaType = "STRING METADATA FROM 'meta.table_name' VIRTUAL";
break;
case MYSQL_METADATA_DATABASE:
+ case DATABASE_NAME:
metaType = "STRING METADATA FROM 'meta.database_name' VIRTUAL";
break;
case MYSQL_METADATA_EVENT_TIME:
+ case OP_TS:
metaType = "TIMESTAMP(3) METADATA FROM 'meta.op_ts' VIRTUAL";
break;
case MYSQL_METADATA_EVENT_TYPE:
+ case OP_TYPE:
metaType = "STRING METADATA FROM 'meta.op_type' VIRTUAL";
break;
case MYSQL_METADATA_DATA:
metaType = "STRING METADATA FROM 'meta.data' VIRTUAL";
break;
case MYSQL_METADATA_IS_DDL:
+ case IS_DDL:
metaType = "BOOLEAN METADATA FROM 'meta.is_ddl' VIRTUAL";
break;
case METADATA_TS:
@@ -830,6 +848,27 @@ public class FlinkSqlParser implements Parser {
return metaType;
}
+ private String parseOracleExtractNodeMetaField(BuiltInFieldInfo metaField)
{
+ String metaType;
+ switch (metaField.getBuiltInField()) {
+ case TABLE_NAME:
+ metaType = "STRING METADATA FROM 'table_name' VIRTUAL";
+ break;
+ case SCHEMA_NAME:
+ metaType = "STRING METADATA FROM 'schema_name' VIRTUAL";
+ break;
+ case DATABASE_NAME:
+ metaType = "STRING METADATA FROM 'database_name' VIRTUAL";
+ break;
+ case OP_TS:
+ metaType = "TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL";
+ break;
+ default:
+ metaType =
TableFormatUtils.deriveLogicalType(metaField.getFormatInfo()).asSummaryString();
+ }
+ return metaType;
+ }
+
/**
* Generate primary key format in sql
*
diff --git
a/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/OracleExtractSqlParseTest.java
b/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/OracleExtractSqlParseTest.java
new file mode 100644
index 000000000..c5106f6fb
--- /dev/null
+++
b/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/OracleExtractSqlParseTest.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.parser;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.inlong.sort.formats.common.IntFormatInfo;
+import org.apache.inlong.sort.formats.common.LongFormatInfo;
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.parser.impl.FlinkSqlParser;
+import org.apache.inlong.sort.parser.result.ParseResult;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.GroupInfo;
+import org.apache.inlong.sort.protocol.StreamInfo;
+import org.apache.inlong.sort.protocol.node.Node;
+import org.apache.inlong.sort.protocol.node.extract.OracleExtractNode;
+import org.apache.inlong.sort.protocol.node.format.CanalJsonFormat;
+import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import
org.apache.inlong.sort.protocol.transformation.relation.NodeRelationShip;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Test for {@link OracleExtractNode}
+ */
+public class OracleExtractSqlParseTest extends AbstractTestBase {
+
+ /**
+ * Build oracle extract node
+ *
+ * @return The oracle extract node
+ */
+ private OracleExtractNode buildOracleExtractNode() {
+ List<FieldInfo> fields = Arrays.asList(
+ new FieldInfo("ID", new LongFormatInfo()),
+ new FieldInfo("NAME", new StringFormatInfo()),
+ new FieldInfo("AGE", new IntFormatInfo()));
+ return new OracleExtractNode("1", "oracle_input", fields,
+ null, null, "ID", "localhost",
+ "flinkuser", "flinkpw", "xE",
+ "flinkuser", "table", 1521, null);
+ }
+
+ private Node buildKafkaLoadNode() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new
LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())
+ );
+ List<FieldRelationShip> relations = Arrays
+ .asList(new FieldRelationShip(new FieldInfo("ID", new
LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("NAME", new
StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("AGE", new
IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()))
+ );
+ return new KafkaLoadNode("2", "kafka_output", fields, relations, null,
+ null, "topic", "localhost:9092",
+ new CanalJsonFormat(), null,
+ null, null);
+ }
+
+ /**
+ * Build node relation
+ *
+ * @param inputs extract node
+ * @param outputs load node
+ * @return node relation
+ */
+ private NodeRelationShip buildNodeRelation(List<Node> inputs, List<Node>
outputs) {
+ List<String> inputIds =
inputs.stream().map(Node::getId).collect(Collectors.toList());
+ List<String> outputIds =
outputs.stream().map(Node::getId).collect(Collectors.toList());
+ return new NodeRelationShip(inputIds, outputIds);
+ }
+
+ /**
+ * Test flink sql task for extract is oracle {@link OracleExtractNode} and
load is kafka {@link KafkaLoadNode}
+ *
+ * @throws Exception The exception may be thrown when executing
+ */
+ @Test
+ public void testOracleExtractSqlParse() throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(60000);
+ env.disableOperatorChaining();
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,
settings);
+ Node inputNode = buildOracleExtractNode();
+ Node outputNode = buildKafkaLoadNode();
+ StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode,
outputNode),
+
Collections.singletonList(buildNodeRelation(Collections.singletonList(inputNode),
+ Collections.singletonList(outputNode))));
+ GroupInfo groupInfo = new GroupInfo("1",
Collections.singletonList(streamInfo));
+ FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv,
groupInfo);
+ ParseResult result = parser.parse();
+ Assert.assertTrue(result.tryExecute());
+ }
+
+}
diff --git a/licenses/inlong-sort/LICENSE b/licenses/inlong-sort/LICENSE
index 212ca0872..8a79d118c 100644
--- a/licenses/inlong-sort/LICENSE
+++ b/licenses/inlong-sort/LICENSE
@@ -688,6 +688,7 @@ The text of each license is also included at
licenses/LICENSE-[project].txt.
org.eclipse.jetty:apache-jstl:9.3.20.v20170531 - Apache :: JSTL module
(https://github.com/eclipse/jetty.project/tree/jetty-9.3.20.v20170531/apache-jstl),
(Apache Software License - Version 2.0; Eclipse Public License - Version 1.0)
com.ververica:flink-connector-mysql-cdc:2.2.1 - flink-connector-mysql-cdc
(https://github.com/ververica/flink-cdc-connectors/tree/master/flink-sql-connector-mysql-cdc/),
(The Apache Software License, Version 2.0)
com.ververica:flink-connector-mongodb-cdc:2.2.1 -
flink-connector-mongodb-cdc
(https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-mongodb-cdc/),
(The Apache Software License, Version 2.0)
+ com.ververica:flink-connector-oracle-cdc:2.2.1 - flink-connector-oracle-cdc
(https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-oracle-cdc/),
(The Apache Software License, Version 2.0)
com.ververica:flink-connector-postgres-cdc:2.2.1 -
flink-connector-postgres-cdc
(https://github.com/ververica/flink-cdc-connectors/tree/master/flink-sql-connector-postgres-cdc/),
(The Apache Software License, Version 2.0)
com.ververica:flink-connector-sqlserver-cdc:2.2.1 -
flink-connector-sqlserver-cdc
(https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-sqlserver-cdc/),
(The Apache Software License, Version 2.0)
org.apache.hadoop:hadoop-annotations:2.10.1 - Apache Hadoop Annotations
(https://github.com/apache/hadoop/tree/branch-2.10.1/hadoop-common-project/hadoop-annotations),
(Apache License, Version 2.0)
diff --git a/licenses/inlong-sort/NOTICE b/licenses/inlong-sort/NOTICE
index 9988bebcf..950360d52 100644
--- a/licenses/inlong-sort/NOTICE
+++ b/licenses/inlong-sort/NOTICE
@@ -616,6 +616,14 @@ This distribution has a binary dependency on jersey, which
is available under th
License. The source code of jersey can be found at
https://github.com/jersey/jersey/.
+========================================================================
+
+flink-connector-oracle-cdc
+Copyright 2022 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
========================================================================
diff --git a/pom.xml b/pom.xml
index 557b63b05..2e972e676 100644
--- a/pom.xml
+++ b/pom.xml
@@ -17,8 +17,8 @@
specific language governing permissions and limitations
under the License.
-->
-<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xmlns="http://maven.apache.org/POM/4.0.0";
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
@@ -202,6 +202,7 @@
<flink.connector.sqlserver.cdc.version>2.2.1</flink.connector.sqlserver.cdc.version>
<flink.pulsar.version>1.13.6.1-rc9</flink.pulsar.version>
<flink.connector.mongodb.cdc.version>2.2.1</flink.connector.mongodb.cdc.version>
+
<flink.connector.oracle.cdc.version>2.2.1</flink.connector.oracle.cdc.version>
<qcloud.flink.cos.fs.hadoop.version>1.10.0-0.1.10</qcloud.flink.cos.fs.hadoop.version>
<qcloud.chdfs.version>2.5</qcloud.chdfs.version>
@@ -989,6 +990,12 @@
<version>${flink.connector.postgres.cdc.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-oracle-cdc</artifactId>
+ <version>${flink.connector.oracle.cdc.version}</version>
+ </dependency>
+
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-sqlserver-cdc</artifactId>
