Skip to content
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

Jump to bottom

[minor][cdc-connector][mongodb] Add op_type metadata column #3615

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[minor][cdc-connector][mongodb] Add op_type metadata column #3615

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
32d3f1a
[minor][cdc-connector][mongodb] Add `op_type` metadata column
qg-lin Sep 20, 2024
9701250
Merge branch 'apache:master' into add-op-type-mata-column-for-mongo
qg-lin Sep 26, 2024
19e38b0
update doc and case
qg-lin Sep 26, 2024
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
6 changes: 6 additions & 0 deletions docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,6 +332,11 @@ upstart 流需要一个唯一的密钥，所以我们必须声明 `_id` 作为
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
<td>它指示在数据库中进行更改的时间。 <br>如果记录是从表的快照而不是改变流中读取的，该值将始终为0。</td>
</tr>
<tr>
<td>op_type</td>
<td>STRING NOT NULL</td>
<td>该行的操作类型，值为：insert、update、replace、delete。<br>由于DeduplicateFunctionHelper#processLastRowOnChangelog返回删除前的行，delete类型暂时缺失。</td>
</tr>
</tbody>
</table>

Expand All @@ -341,6 +346,7 @@ CREATE TABLE products (
db_name STRING METADATA FROM 'database_name' VIRTUAL,
collection_name STRING METADATA FROM 'collection_name' VIRTUAL,
operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
operation_type STRING METADATA FROM 'op_type' VIRTUAL,
_id STRING, // 必须声明
name STRING,
weight DECIMAL(10,3),
Expand Down
6 changes: 6 additions & 0 deletions docs/content/docs/connectors/flink-sources/mongodb-cdc.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,11 @@ The following format metadata can be exposed as read-only (VIRTUAL) columns in a
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
<td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
</tr>
<tr>
<td>op_type</td>
<td>STRING NOT NULL</td>
<td>Operation type of the row, the value is one of: insert,update,replace,delete. <br>Since DeduplicateFunctionHelper#processLastRowOnChangelog returns the row before deletion, the `delete` type is temporarily missing.</td>
</tr>
</tbody>
</table>

Expand All @@ -366,6 +371,7 @@ CREATE TABLE products (
db_name STRING METADATA FROM 'database_name' VIRTUAL,
collection_name STRING METADATA FROM 'collection_name' VIRTUAL,
operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
operation_type STRING METADATA FROM 'op_type' VIRTUAL,
_id STRING, // must be declared
name STRING,
weight DECIMAL(10,3),
Expand Down
14 changes: 14 additions & 0 deletions .../src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBReadableMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,20 @@ public Object read(SourceRecord record) {
return TimestampData.fromEpochMillis(
(Long) source.get(AbstractSourceInfo.TIMESTAMP_KEY));
}
}),

OP_TYPE(
"op_type",
DataTypes.STRING().notNull(),
new MetadataConverter() {
private static final long serialVersionUID = 1L;

@Override
public Object read(SourceRecord record) {
Struct value = (Struct) record.value();
String opType = value.getString(MongoDBEnvelope.OPERATION_TYPE_FIELD);
return StringData.fromString(opType);
}
});

private final String key;
Expand Down
67 changes: 34 additions & 33 deletions .../test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBFullChangelogITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -648,6 +648,7 @@ private void testMongoDBParallelSource(
+ " name STRING,"
+ " address STRING,"
+ " phone_number STRING,"
+ " op_type STRING METADATA FROM 'op_type' VIRTUAL,"
qg-lin marked this conversation as resolved.
Show resolved Hide resolved
+ " primary key (_id) not enforced"
+ ") WITH ("
+ " 'connector' = 'mongodb-cdc',"
Expand All @@ -674,31 +675,31 @@ private void testMongoDBParallelSource(
// first step: check the snapshot data
String[] snapshotForSingleTable =
new String[] {
"+I[101, user_1, Shanghai, 123567891234]",
"+I[102, user_2, Shanghai, 123567891234]",
"+I[103, user_3, Shanghai, 123567891234]",
"+I[109, user_4, Shanghai, 123567891234]",
"+I[110, user_5, Shanghai, 123567891234]",
"+I[111, user_6, Shanghai, 123567891234]",
"+I[118, user_7, Shanghai, 123567891234]",
"+I[121, user_8, Shanghai, 123567891234]",
"+I[123, user_9, Shanghai, 123567891234]",
"+I[1009, user_10, Shanghai, 123567891234]",
"+I[1010, user_11, Shanghai, 123567891234]",
"+I[1011, user_12, Shanghai, 123567891234]",
"+I[1012, user_13, Shanghai, 123567891234]",
"+I[1013, user_14, Shanghai, 123567891234]",
"+I[1014, user_15, Shanghai, 123567891234]",
"+I[1015, user_16, Shanghai, 123567891234]",
"+I[1016, user_17, Shanghai, 123567891234]",
"+I[1017, user_18, Shanghai, 123567891234]",
"+I[1018, user_19, Shanghai, 123567891234]",
"+I[1019, user_20, Shanghai, 123567891234]",
"+I[2000, user_21, Shanghai, 123567891234]"
"+I[101, user_1, Shanghai, 123567891234, insert]",
"+I[102, user_2, Shanghai, 123567891234, insert]",
"+I[103, user_3, Shanghai, 123567891234, insert]",
"+I[109, user_4, Shanghai, 123567891234, insert]",
"+I[110, user_5, Shanghai, 123567891234, insert]",
"+I[111, user_6, Shanghai, 123567891234, insert]",
"+I[118, user_7, Shanghai, 123567891234, insert]",
"+I[121, user_8, Shanghai, 123567891234, insert]",
"+I[123, user_9, Shanghai, 123567891234, insert]",
"+I[1009, user_10, Shanghai, 123567891234, insert]",
"+I[1010, user_11, Shanghai, 123567891234, insert]",
"+I[1011, user_12, Shanghai, 123567891234, insert]",
"+I[1012, user_13, Shanghai, 123567891234, insert]",
"+I[1013, user_14, Shanghai, 123567891234, insert]",
"+I[1014, user_15, Shanghai, 123567891234, insert]",
"+I[1015, user_16, Shanghai, 123567891234, insert]",
"+I[1016, user_17, Shanghai, 123567891234, insert]",
"+I[1017, user_18, Shanghai, 123567891234, insert]",
"+I[1018, user_19, Shanghai, 123567891234, insert]",
"+I[1019, user_20, Shanghai, 123567891234, insert]",
"+I[2000, user_21, Shanghai, 123567891234, insert]"
};
tEnv.executeSql(sourceDDL);
TableResult tableResult =
tEnv.executeSql("select cid, name, address, phone_number from customers");
tEnv.executeSql("select cid, name, address, phone_number, op_type from customers");
CloseableIterator<Row> iterator = tableResult.collect();
JobID jobId = tableResult.getJobClient().get().getJobID();
List<String> expectedSnapshotData = new ArrayList<>();
Expand Down Expand Up @@ -731,17 +732,17 @@ private void testMongoDBParallelSource(

String[] changeEventsForSingleTable =
new String[] {
"-U[101, user_1, Shanghai, 123567891234]",
"+U[101, user_1, Hangzhou, 123567891234]",
"-D[102, user_2, Shanghai, 123567891234]",
"+I[102, user_2, Shanghai, 123567891234]",
"-U[103, user_3, Shanghai, 123567891234]",
"+U[103, user_3, Hangzhou, 123567891234]",
"-U[1010, user_11, Shanghai, 123567891234]",
"+U[1010, user_11, Hangzhou, 123567891234]",
"+I[2001, user_22, Shanghai, 123567891234]",
"+I[2002, user_23, Shanghai, 123567891234]",
"+I[2003, user_24, Shanghai, 123567891234]"
"-U[101, user_1, Shanghai, 123567891234, update]",
"+U[101, user_1, Hangzhou, 123567891234, update]",
"-D[102, user_2, Shanghai, 123567891234, delete]",
"+I[102, user_2, Shanghai, 123567891234, insert]",
"-U[103, user_3, Shanghai, 123567891234, update]",
"+U[103, user_3, Hangzhou, 123567891234, update]",
"-U[1010, user_11, Shanghai, 123567891234, update]",
"+U[1010, user_11, Hangzhou, 123567891234, update]",
"+I[2001, user_22, Shanghai, 123567891234, insert]",
"+I[2002, user_23, Shanghai, 123567891234, insert]",
"+I[2003, user_24, Shanghai, 123567891234, insert]"
};
List<String> expectedChangeStreamData = new ArrayList<>();
for (int i = 0; i < captureCustomerCollections.length; i++) {
Expand Down
36 changes: 20 additions & 16 deletions ...c/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBConnectorITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -482,6 +482,7 @@ public void testMetadataColumns() throws Exception {
+ " weight DECIMAL(10,3),"
+ " db_name STRING METADATA FROM 'database_name' VIRTUAL,"
+ " collection_name STRING METADATA VIRTUAL,"
+ " op_type STRING METADATA VIRTUAL,"
+ " PRIMARY KEY (_id) NOT ENFORCED"
+ ") WITH ("
+ " 'connector' = 'mongodb-cdc',"
Expand All @@ -508,6 +509,7 @@ public void testMetadataColumns() throws Exception {
+ " weight DECIMAL(10,3),"
+ " database_name STRING,"
+ " collection_name STRING,"
+ " op_type STRING,"
+ " PRIMARY KEY (_id) NOT ENFORCED"
+ ") WITH ("
+ " 'connector' = 'values',"
Expand Down Expand Up @@ -562,22 +564,24 @@ public void testMetadataColumns() throws Exception {

List<String> expected =
Stream.of(
"+I(100000000000000000000101,scooter,Small 2-wheel scooter,3.140,%s,products)",
"+I(100000000000000000000102,car battery,12V car battery,8.100,%s,products)",
"+I(100000000000000000000103,12-pack drill bits,12-pack of drill bits with sizes ranging from #40 to #3,0.800,%s,products)",
"+I(100000000000000000000104,hammer,12oz carpenter''s hammer,0.750,%s,products)",
"+I(100000000000000000000105,hammer,12oz carpenter''s hammer,0.875,%s,products)",
"+I(100000000000000000000106,hammer,12oz carpenter''s hammer,1.000,%s,products)",
"+I(100000000000000000000107,rocks,box of assorted rocks,5.300,%s,products)",
"+I(100000000000000000000108,jacket,water resistent black wind breaker,0.100,%s,products)",
"+I(100000000000000000000109,spare tire,24 inch spare tire,22.200,%s,products)",
"+I(100000000000000000000110,jacket,water resistent white wind breaker,0.200,%s,products)",
"+I(100000000000000000000111,scooter,Big 2-wheel scooter,5.180,%s,products)",
"+U(100000000000000000000106,hammer,18oz carpenter hammer,1.000,%s,products)",
"+U(100000000000000000000107,rocks,box of assorted rocks,5.100,%s,products)",
"+U(100000000000000000000110,jacket,new water resistent white wind breaker,0.500,%s,products)",
"+U(100000000000000000000111,scooter,Big 2-wheel scooter,5.170,%s,products)",
"-D(100000000000000000000111,scooter,Big 2-wheel scooter,5.170,%s,products)")
"+I(100000000000000000000101,scooter,Small 2-wheel scooter,3.140,%s,products,insert)",
"+I(100000000000000000000102,car battery,12V car battery,8.100,%s,products,insert)",
"+I(100000000000000000000103,12-pack drill bits,12-pack of drill bits with sizes ranging from #40 to #3,0.800,%s,products,insert)",
"+I(100000000000000000000104,hammer,12oz carpenter''s hammer,0.750,%s,products,insert)",
"+I(100000000000000000000105,hammer,12oz carpenter''s hammer,0.875,%s,products,insert)",
"+I(100000000000000000000106,hammer,12oz carpenter''s hammer,1.000,%s,products,insert)",
"+I(100000000000000000000107,rocks,box of assorted rocks,5.300,%s,products,insert)",
"+I(100000000000000000000108,jacket,water resistent black wind breaker,0.100,%s,products,insert)",
"+I(100000000000000000000109,spare tire,24 inch spare tire,22.200,%s,products,insert)",
"+I(100000000000000000000110,jacket,water resistent white wind breaker,0.200,%s,products,insert)",
"+I(100000000000000000000111,scooter,Big 2-wheel scooter,5.180,%s,products,insert)",
"+U(100000000000000000000106,hammer,18oz carpenter hammer,1.000,%s,products,update)",
"+U(100000000000000000000107,rocks,box of assorted rocks,5.100,%s,products,update)",
"+U(100000000000000000000110,jacket,new water resistent white wind breaker,0.500,%s,products,update)",
"+U(100000000000000000000111,scooter,Big 2-wheel scooter,5.170,%s,products,update)",
// NOTE: DeduplicateFunctionHelper#processLastRowOnChangelog returns
// the preRow before deletion, so the `op_type` is 'update' here.
"-D(100000000000000000000111,scooter,Big 2-wheel scooter,5.170,%s,products,update)")
.map(s -> String.format(s, database))
.sorted()
.collect(Collectors.toList());
Expand Down
Loading