Skip to content

[Feature][Oracle-CDC] Support read no primary key table #6209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hailin0 merged 1 commit into from
Jan 15, 2024
Merged

[Feature][Oracle-CDC] Support read no primary key table #6209

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...connectors/seatunnel/cdc/oracle/source/reader/fetch/scan/OracleSnapshotSplitReadTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,7 +218,7 @@ private void createDataEventsForTable(
snapshotSplit.getSplitEnd() == null,
snapshotSplit.getSplitStart(),
snapshotSplit.getSplitEnd(),
snapshotSplit.getSplitKeyType().getTotalFields(),
snapshotSplit.getSplitKeyType(),
connectorConfig.getSnapshotFetchSize());
ResultSet rs = selectStatement.executeQuery()) {

Expand Down
3 changes: 2 additions & 1 deletion ...src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -293,13 +293,14 @@ public static PreparedStatement readTableSplitDataStatement(
boolean isLastSplit,
Object[] splitStart,
Object[] splitEnd,
int primaryKeyNum,
SeaTunnelRowType splitKeyType,
int fetchSize) {
try {
final PreparedStatement statement = initStatement(jdbc, sql, fetchSize);
if (isFirstSplit && isLastSplit) {
return statement;
}
int primaryKeyNum = splitKeyType.getTotalFields();
if (isFirstSplit) {
for (int i = 0; i < primaryKeyNum; i++) {
statement.setObject(i + 1, splitEnd[i]);
Expand Down
46 changes: 46 additions & 0 deletions ...e-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ public class OracleCDCIT extends TestSuiteBase implements TestResource {
private static final String DATABASE = "DEBEZIUM";
private static final String SOURCE_TABLE1 = "FULL_TYPES";
private static final String SOURCE_TABLE2 = "FULL_TYPES2";
private static final String SOURCE_TABLE_NO_PRIMARY_KEY = "FULL_TYPES_NO_PRIMARY_KEY";

private static final String SINK_TABLE1 = "SINK_FULL_TYPES";
private static final String SINK_TABLE2 = "SINK_FULL_TYPES2";
Expand Down Expand Up @@ -175,6 +176,51 @@ public void testOracleCdcCheckDataE2e(TestContainer container) throws Exception
});
}

@TestTemplate
public void testOracleCdcCheckDataWithNoPrimaryKey(TestContainer container) throws Exception {

clearTable(DATABASE, SOURCE_TABLE_NO_PRIMARY_KEY);
clearTable(DATABASE, SINK_TABLE1);

insertSourceTable(DATABASE, SOURCE_TABLE_NO_PRIMARY_KEY);

CompletableFuture.supplyAsync(
() -> {
try {
container.executeJob("/oraclecdc_to_oracle_with_no_primary_key.conf");
} catch (Exception e) {
log.error("Commit task exception :" + e.getMessage());
throw new RuntimeException(e);
}
return null;
});

// snapshot stage
await().atMost(600000, TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
Assertions.assertIterableEquals(
querySql(
getSourceQuerySQL(
DATABASE, SOURCE_TABLE_NO_PRIMARY_KEY)),
querySql(getSourceQuerySQL(DATABASE, SINK_TABLE1)));
});

// insert update delete
updateSourceTable(DATABASE, SOURCE_TABLE_NO_PRIMARY_KEY);

// stream stage
await().atMost(600000, TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
Assertions.assertIterableEquals(
querySql(
getSourceQuerySQL(
DATABASE, SOURCE_TABLE_NO_PRIMARY_KEY)),
querySql(getSourceQuerySQL(DATABASE, SINK_TABLE1)));
});
}

@TestTemplate
@DisabledOnContainer(
value = {},
Expand Down
53 changes: 53 additions & 0 deletions ...nel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/resources/ddl/column_type_test.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,46 @@ create table DEBEZIUM.FULL_TYPES2 (
primary key (ID)
);

create table DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY (
ID NUMBER(9) not null,
VAL_VARCHAR VARCHAR2(1000),
VAL_VARCHAR2 VARCHAR2(1000),
VAL_NVARCHAR2 NVARCHAR2(1000),
VAL_CHAR CHAR(3),
VAL_NCHAR NCHAR(3),
VAL_BF BINARY_FLOAT,
VAL_BD BINARY_DOUBLE,
VAL_F FLOAT,
VAL_F_10 FLOAT(10),
VAL_NUM NUMBER(10, 6),
VAL_DP FLOAT,
VAL_R FLOAT(63),
VAL_DECIMAL NUMBER(10, 6),
VAL_NUMERIC NUMBER(10, 6),
VAL_NUM_VS NUMBER,
VAL_INT NUMBER,
VAL_INTEGER NUMBER,
VAL_SMALLINT NUMBER,
VAL_NUMBER_38_NO_SCALE NUMBER(38),
VAL_NUMBER_38_SCALE_0 NUMBER(38),
VAL_NUMBER_1 NUMBER(1),
VAL_NUMBER_2 NUMBER(2),
VAL_NUMBER_4 NUMBER(4),
VAL_NUMBER_9 NUMBER(9),
VAL_NUMBER_18 NUMBER(18),
VAL_NUMBER_2_NEGATIVE_SCALE NUMBER(1, -1),
VAL_NUMBER_4_NEGATIVE_SCALE NUMBER(2, -2),
VAL_NUMBER_9_NEGATIVE_SCALE NUMBER(8, -1),
VAL_NUMBER_18_NEGATIVE_SCALE NUMBER(16, -2),
VAL_NUMBER_36_NEGATIVE_SCALE NUMBER(36, -2),
VAL_DATE DATE,
VAL_TS TIMESTAMP(6),
VAL_TS_PRECISION2 TIMESTAMP(2),
VAL_TS_PRECISION4 TIMESTAMP(4),
VAL_TS_PRECISION9 TIMESTAMP(6),
VAL_TSLTZ TIMESTAMP(6) WITH LOCAL TIME ZONE
);

INSERT INTO DEBEZIUM.FULL_TYPES VALUES (
1, 'vc2', 'vc2', 'nvc2', 'c', 'nc',
1.1, 2.22, 3.33, 8.888, 4.4444, 5.555, 6.66, 1234.567891, 1234.567891, 77.323,
Expand All @@ -128,6 +168,19 @@ INSERT INTO DEBEZIUM.FULL_TYPES2 VALUES (
TO_TIMESTAMP_TZ('2022-10-30 01:34:56.00789', 'yyyy-mm-dd HH24:MI:SS.FF5')
);

INSERT INTO DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY VALUES (
1, 'vc2', 'vc2', 'nvc2', 'c', 'nc',
1.1, 2.22, 3.33, 8.888, 4.4444, 5.555, 6.66, 1234.567891, 1234.567891, 77.323,
1, 22, 333, 4444, 5555, 1, 99, 9999, 999999999, 999999999999999999,
94, 9949, 999999994, 999999999999999949, 99999999999999999999999999999999999949,
TO_DATE('2022-10-30', 'yyyy-mm-dd'),
TO_TIMESTAMP('2022-10-30 12:34:56.00789', 'yyyy-mm-dd HH24:MI:SS.FF5'),
TO_TIMESTAMP('2022-10-30 12:34:56.12545', 'yyyy-mm-dd HH24:MI:SS.FF5'),
TO_TIMESTAMP('2022-10-30 12:34:56.12545', 'yyyy-mm-dd HH24:MI:SS.FF5'),
TO_TIMESTAMP('2022-10-30 12:34:56.125456789', 'yyyy-mm-dd HH24:MI:SS.FF9'),
TO_TIMESTAMP_TZ('2022-10-30 01:34:56.00789', 'yyyy-mm-dd HH24:MI:SS.FF5')
);

create table DEBEZIUM.SINK_FULL_TYPES (
ID NUMBER(9) not null,
VAL_VARCHAR VARCHAR2(1000),
Expand Down
66 changes: 66 additions & 0 deletions .../connector-cdc-oracle-e2e/src/test/resources/oraclecdc_to_oracle_with_no_primary_key.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
#
# 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.
#
######
###### This config file is a demonstration of streaming processing in seatunnel config
######

env {
# You can set engine configuration here
execution.parallelism = 1
job.mode = "STREAMING"
checkpoint.interval = 5000
}

source {
# This is a example source plugin **only for test and demonstrate the feature source plugin**
Oracle-CDC {
result_table_name = "customers"
username = "system"
password = "oracle"
database-names = ["XE"]
schema-names = ["DEBEZIUM"]
table-names = ["XE.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY"]
base-url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
source.reader.close.timeout = 120000
connection.pool.size = 1
debezium {
# log.mining.strategy = "online_catalog"
# log.mining.continuous.mine = true
database.oracle.jdbc.timezoneAsRegion = "false"
}

exactly_once = false
}
}

transform {
}

sink {
Jdbc {
source_table_name = "customers"
driver = "oracle.jdbc.driver.OracleDriver"
url = "jdbc:oracle:thin:system/oracle@oracle-host:1521:xe"
user = "system"
password = "oracle"
generate_sink_sql = true
database = "XE"
table = "DEBEZIUM.SINK_FULL_TYPES"
batch_size = 1
primary_keys = ["ID"]
}
}