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 1 commit
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
5 changes: 5 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>该行的操作类型。</td>
qg-lin marked this conversation as resolved.
Show resolved Hide resolved
</tr>
</tbody>
</table>

Expand Down
5 changes: 5 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.</td>
</tr>
</tbody>
</table>

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
Loading