alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-2673 Read the raw bytes of a character-type field

Browse Source
This commit is contained in:
Arik Cohen 2020-10-22 13:40:14 -04:00 committed by Jiri Pechanec
parent c274c2b39b
commit e44b76fb9d
2 changed files with 21 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/SnapshotReader.java
View File

@ -171,6 +171,12 @@ else if (actualColumn.jdbcType() == Types.TINYINT || actualColumn.jdbcType() ==
// read it again to get correct scale
return rs.getObject(fieldNo) == null ? null : rs.getInt(fieldNo);
}
// DBZ-2673
else if (actualColumn.typeName().equals("CHAR") ||
actualColumn.typeName().equals("VARCHAR") ||
actualColumn.typeName().equals("TEXT")) {
return rs.getBytes(fieldNo);
}
else {
return rs.getObject(fieldNo);
}

15
debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/SnapshotReaderIT.java
View File

@ -190,6 +190,20 @@ private void snapshotOfSingleDatabase(boolean useGlobalLock, boolean storeOnlyMo
assertThat(customers.numberOfKeySchemaChanges()).isEqualTo(1);
assertThat(customers.numberOfValueSchemaChanges()).isEqualTo(1);
List<Struct> customerRecrods = new ArrayList<>();
customers.forEach(val -> {
customerRecrods.add(((Struct) val.value()).getStruct("after"));
});
Struct customer = customerRecrods.stream().sorted((a, b) -> a.getInt32("id").compareTo(b.getInt32("id"))).findFirst().get();
assertThat(customer.get("first_name")).isInstanceOf(String.class);
assertThat(customer.get("last_name")).isInstanceOf(String.class);
assertThat(customer.get("email")).isInstanceOf(String.class);
assertThat(customer.get("first_name")).isEqualTo("Sally");
assertThat(customer.get("last_name")).isEqualTo("Thomas");
assertThat(customer.get("email")).isEqualTo("sally.thomas@acme.com");
Collection orders = store.collection(DATABASE.getDatabaseName(), "orders");
assertThat(orders.numberOfCreates()).isEqualTo(5);
assertThat(orders.numberOfUpdates()).isEqualTo(0);
@ -250,6 +264,7 @@ public void snapshotWithBackupLocksShouldNotWaitForReads() throws Exception {
MySQLConnection db = MySQLConnection.forTestDatabase(DATABASE.getDatabaseName());
Thread t = new Thread() {
@Override
public void run() {
try {
JdbcConnection connection = db.connect();