Skip to content

[FLINK-39406][oracle-cdc] Add support for reading from partitioned tables in Oracle connector #4365

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

Open
Hisoka-X wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-39406][oracle-cdc] Add support for reading from partitioned tables in Oracle connector #4365

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
190 changes: 190 additions & 0 deletions ...cle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1800,6 +1800,196 @@ private List<Event> executeAlterAndProvideExpected(TableId tableId, Statement st
return expected;
}

@Test
@Order(10)
public void testPartitionedTable() throws Exception {
createAndInitialize("partition_table.sql");
Map<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), ORACLE_CONTAINER.getHost());
options.put(PORT.key(), String.valueOf(ORACLE_CONTAINER.getOraclePort()));
options.put(USERNAME.key(), CONNECTOR_USER);
options.put(PASSWORD.key(), CONNECTOR_PWD);
options.put(TABLES.key(), "debezium.partitioned_customers");
options.put(DATABASE.key(), ORACLE_CONTAINER.getDatabaseName());
options.put(SCAN_STARTUP_MODE.key(), "initial");
Factory.Context context = new MockContext(Configuration.fromMap(options));
Properties dbzProperties = new Properties();
dbzProperties.setProperty("database.connection.adapter", "logminer");
dbzProperties.setProperty("log.mining.strategy", "online_catalog");
dbzProperties.setProperty("snapshot.locking.mode", "none");
dbzProperties.setProperty("database.history.store.only.captured.tables.ddl", "true");
dbzProperties.setProperty("include.schema.changes", "true");
OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
configFactory
.username(CONNECTOR_USER)
.password(CONNECTOR_PWD)
.port(ORACLE_CONTAINER.getOraclePort())
.databaseList(ORACLE_CONTAINER.getDatabaseName())
.hostname(ORACLE_CONTAINER.getHost())
.tableList("DEBEZIUM.PARTITIONED_CUSTOMERS")
.startupOptions(StartupOptions.initial())
.debeziumProperties(dbzProperties)
.includeSchemaChanges(true);
FlinkSourceProvider sourceProvider =
(FlinkSourceProvider)
new OracleDataSource(
configFactory,
context.getFactoryConfiguration(),
new ArrayList<>())
.getEventSourceProvider();
CloseableIterator<Event> events =
env.fromSource(
sourceProvider.getSource(),
WatermarkStrategy.noWatermarks(),
OracleDataSourceFactory.IDENTIFIER,
new EventTypeInfo())
.executeAndCollect();
Thread.sleep(150_000);

TableId tableId = TableId.tableId("DEBEZIUM", "PARTITIONED_CUSTOMERS");

// Build expected CreateTableEvent
CreateTableEvent createTableEvent =
new CreateTableEvent(
tableId,
Schema.newBuilder()
.physicalColumn("ID", DataTypes.BIGINT().notNull())
.physicalColumn("NAME", DataTypes.VARCHAR(255).notNull())
.physicalColumn("ADDRESS", DataTypes.VARCHAR(1024))
.physicalColumn("PHONE_NUMBER", DataTypes.VARCHAR(512))
.primaryKey(Collections.singletonList("ID"))
.build());

// Build expected snapshot data
RowType rowType =
RowType.of(
new DataType[] {
DataTypes.BIGINT().notNull(),
DataTypes.VARCHAR(255).notNull(),
DataTypes.VARCHAR(1024),
DataTypes.VARCHAR(512)
},
new String[] {"ID", "NAME", "ADDRESS", "PHONE_NUMBER"});
BinaryRecordDataGenerator generator = new BinaryRecordDataGenerator(rowType);

List<Event> expectedSnapshot = new ArrayList<>();
expectedSnapshot.add(createTableEvent);
// 21 rows across 4 partitions
long[][] snapshotIds = {
{101L},
{102L},
{103L},
{109L},
{110L},
{111L}, // partition p1 (< 500)
{501L},
{502L},
{513L}, // partition p2 (< 1000)
{1009L},
{1010L},
{1011L},
{1012L},
{1013L},
{1014L},
{1015L},
{1016L},
{1017L},
{1018L},
{1019L}, // partition p3 (< 2000)
{2000L} // partition p4 (MAXVALUE)
};
String[] names = {
"user_1", "user_2", "user_3", "user_4", "user_5", "user_6", "user_7", "user_8",
"user_9", "user_10", "user_11", "user_12", "user_13", "user_14", "user_15", "user_16",
"user_17", "user_18", "user_19", "user_20", "user_21"
};
for (int i = 0; i < snapshotIds.length; i++) {
expectedSnapshot.add(
DataChangeEvent.insertEvent(
tableId,
generator.generate(
new Object[] {
snapshotIds[i][0],
BinaryStringData.fromString(names[i]),
BinaryStringData.fromString("Shanghai"),
BinaryStringData.fromString("123567891234")
})));
}

// Perform DML operations across partitions
List<Event> expectedBinlog = new ArrayList<>();
try (Connection connection = getJdbcConnection();
Statement statement = connection.createStatement()) {
// INSERT into partition p4
statement.execute(
"INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (2001, 'user_22', 'Shanghai', '123567891234')");
expectedBinlog.add(
DataChangeEvent.insertEvent(
tableId,
generator.generate(
new Object[] {
2001L,
BinaryStringData.fromString("user_22"),
BinaryStringData.fromString("Shanghai"),
BinaryStringData.fromString("123567891234")
})));
// UPDATE in partition p1
statement.execute(
"UPDATE DEBEZIUM.PARTITIONED_CUSTOMERS SET ADDRESS = 'Hangzhou' WHERE ID = 103");
expectedBinlog.add(
DataChangeEvent.updateEvent(
tableId,
generator.generate(
new Object[] {
103L,
BinaryStringData.fromString("user_3"),
BinaryStringData.fromString("Shanghai"),
BinaryStringData.fromString("123567891234")
}),
generator.generate(
new Object[] {
103L,
BinaryStringData.fromString("user_3"),
BinaryStringData.fromString("Hangzhou"),
BinaryStringData.fromString("123567891234")
})));
// DELETE from partition p2
statement.execute("DELETE FROM DEBEZIUM.PARTITIONED_CUSTOMERS WHERE ID = 502");
expectedBinlog.add(
DataChangeEvent.deleteEvent(
tableId,
generator.generate(
new Object[] {
502L,
BinaryStringData.fromString("user_8"),
BinaryStringData.fromString("Shanghai"),
BinaryStringData.fromString("123567891234")
})));
}

// Fetch and verify snapshot events
List<Event> actual = fetchResults(events, expectedSnapshot.size() + expectedBinlog.size());

Map<TableId, List<RecordData.FieldGetter>> fieldGetterMaps = new HashMap<>();
Schema schema = createTableEvent.getSchema();
fieldGetterMaps.put(tableId, SchemaUtils.createFieldGetters(schema));

StringBuilder actualSnapshotStr =
getResultString(actual.subList(0, expectedSnapshot.size()), fieldGetterMaps, false);
StringBuilder expectedSnapshotStr =
getResultString(expectedSnapshot, fieldGetterMaps, false);
assertThat(actualSnapshotStr.toString()).isEqualTo(expectedSnapshotStr.toString());

StringBuilder actualBinlogStr =
getResultString(
actual.subList(expectedSnapshot.size(), actual.size()),
fieldGetterMaps,
false);
StringBuilder expectedBinlogStr = getResultString(expectedBinlog, fieldGetterMaps, false);
assertThat(actualBinlogStr.toString()).isEqualTo(expectedBinlogStr.toString());
env.close();
}

public static <T> List<T> fetchResults(Iterator<T> iter, int size) {
List<T> result = new ArrayList<>(size);
while (size > 0 && iter.hasNext()) {
Expand Down
2 changes: 1 addition & 1 deletion ...cle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleSourceTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ protected static List<TableId> listTables(Connection connection) {
Set<TableId> tableIdSet = new HashSet<>();
String queryTablesSql =
"SELECT OWNER ,TABLE_NAME,TABLESPACE_NAME FROM ALL_TABLES \n"
+ "WHERE TABLESPACE_NAME IS NOT NULL AND TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX') "
+ "WHERE (PARTITIONED = 'YES' OR (TABLESPACE_NAME IS NOT NULL AND TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX'))) "
+ "AND NESTED = 'NO' AND TABLE_NAME NOT IN (SELECT PARENT_TABLE_NAME FROM ALL_NESTED_TABLES)";
try {
ResultSet resultSet = connection.createStatement().executeQuery(queryTablesSql);
Expand Down
57 changes: 57 additions & 0 deletions ...connectors/flink-cdc-pipeline-connector-oracle/src/test/resources/ddl/partition_table.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
-- 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.

-- ----------------------------------------------------------------------------------------------------------------
-- DATABASE: partition table test
-- ----------------------------------------------------------------------------------------------------------------

-- Create a range-partitioned table for CDC partition table testing
CREATE TABLE DEBEZIUM.PARTITIONED_CUSTOMERS (
ID NUMBER(10) NOT NULL,
NAME VARCHAR2(255) NOT NULL,
ADDRESS VARCHAR2(1024),
PHONE_NUMBER VARCHAR2(512),
PRIMARY KEY(ID)
)
PARTITION BY RANGE (ID) (
PARTITION p1 VALUES LESS THAN (500),
PARTITION p2 VALUES LESS THAN (1000),
PARTITION p3 VALUES LESS THAN (2000),
PARTITION p4 VALUES LESS THAN (MAXVALUE)
);

ALTER TABLE DEBEZIUM.PARTITIONED_CUSTOMERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (101,'user_1','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (102,'user_2','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (103,'user_3','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (109,'user_4','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (110,'user_5','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (111,'user_6','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (501,'user_7','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (502,'user_8','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (513,'user_9','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1009,'user_10','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1010,'user_11','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1011,'user_12','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1012,'user_13','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1013,'user_14','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1014,'user_15','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1015,'user_16','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1016,'user_17','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1017,'user_18','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1018,'user_19','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (1019,'user_20','Shanghai','123567891234');
INSERT INTO DEBEZIUM.PARTITIONED_CUSTOMERS VALUES (2000,'user_21','Shanghai','123567891234');
2 changes: 1 addition & 1 deletion .../main/java/org/apache/flink/cdc/connectors/oracle/source/utils/OracleConnectionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ public static List<TableId> listTables(
Set<TableId> tableIdSet = new HashSet<>();
String queryTablesSql =
"SELECT OWNER ,TABLE_NAME,TABLESPACE_NAME FROM ALL_TABLES \n"
+ "WHERE TABLESPACE_NAME IS NOT NULL AND TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX') AND NESTED = 'NO' AND TABLE_NAME NOT IN (SELECT PARENT_TABLE_NAME FROM ALL_NESTED_TABLES)";
+ "WHERE (PARTITIONED = 'YES' OR (TABLESPACE_NAME IS NOT NULL AND TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX'))) AND NESTED = 'NO' AND TABLE_NAME NOT IN (SELECT PARENT_TABLE_NAME FROM ALL_NESTED_TABLES)";
try {
jdbcConnection.query(
queryTablesSql,
Expand Down
Loading
Loading