Support metadata columns for mysql-cdc connector #496
Conversation
|
|
||
| // TODO: we execute an additional DELETE statement to make incremental-snapshot | ||
| // mode pass, in that mode, the first statement will be ignored which should be a bug! | ||
| statement.execute("DELETE FROM shopping_cart WHERE product_no=101;"); |
There was a problem hiding this comment.
This might be a bug in new source, we need to investigate in another issue.
|
The tests failed |
|
I did a review of your code, and I think your implementation is better. |
| /** {@link SourceRecord} metadata info converter. */ | ||
| @FunctionalInterface | ||
| public interface MetadataConverter extends Serializable { | ||
| Object read(SourceRecord record); | ||
| } | ||
|
|
||
| /** Emits a row with physical fields and metadata fields. */ | ||
| private static final class OutputProjectionCollector |
There was a problem hiding this comment.
Is it possible to define this MetadataConverter as a generic interface and separate MetadataConverter and OutputProjectionCollector from RowDataDebeziumDeserializeSchema so that other connectors can reuse this logic.
| @@ -101,12 +101,12 @@ public void setup() throws Exception { | |||
| } | |||
|
|
|||
| @Test | |||
| public void testMysqlServerInBerlin() throws Exception { | |||
| public void testMySqlServerInBerlin() throws Exception { | |||
| testTemporalTypesWithMySqlServerTimezone("Asia/Shanghai"); | |||
There was a problem hiding this comment.
testTemporalTypesWithMySqlServerTimezone("Asia/Shanghai") -> testTemporalTypesWithMySqlServerTimezone("Europe/Berlin")
...or-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/table/MySqlTimezoneITCase.java
Outdated
Show resolved
Hide resolved
wuchong
force-pushed
the
mysql-metadata
branch
from
81d6fcb
to
ad07a06
Compare
|
LGTM， When the CI Pass. |
| @@ -622,7 +726,7 @@ public void testPrimaryKeyWithSnowflakeAlgorithm() throws Exception { | |||
| } | |||
|
|
|||
| @Test | |||
| public void testInconsistentSchema() throws Exception { | |||
| public void testShadingTablesWithInconsistentSchema() throws Exception { | |||
There was a problem hiding this comment.
| public void testShadingTablesWithInconsistentSchema() throws Exception { | |
| public void testShardingTablesWithInconsistentSchema() throws Exception { |
| } | ||
| }), | ||
|
|
||
| TIMESTAMP( |
There was a problem hiding this comment.
Could we add a description for every metadata? And it's better to add the feature in document.
| @@ -479,6 +483,106 @@ public void testWideTable() throws Exception { | |||
| result.getJobClient().get().cancel().get(); | |||
| } | |||
|
|
|||
| @Test | |||
| public void testMetadataColumns() throws Exception { | |||
| userDatabase1.createAndInitialize(); | |||
There was a problem hiding this comment.
We need to cleanup the userDatabase1 resource, otherwise the test testShadingTablesWithInconsistentSchema will fail due to matching the database creating in this test.
|
Thanks all for the feedbacks, I have address all your comments. |
|
Thanks @wuchong, I'll open a PR to support mongodb metadata columns as well. |
We introduced following metadata columns in this PR: