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

[ISSUE-1827][Mysql] Mysql connector support parallel snapshot when th… #2046

Closed
wants to merge 1 commit into from
Closed

[ISSUE-1827][Mysql] Mysql connector support parallel snapshot when th… #2046

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
74 changes: 64 additions & 10 deletions ...src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package com.ververica.cdc.connectors.mysql.debezium.reader;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FlinkRuntimeException;

import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.ThreadFactoryBuilder;
Expand Down Expand Up @@ -49,8 +50,10 @@
import javax.annotation.Nullable;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
Expand Down Expand Up @@ -248,7 +251,11 @@ public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
boolean reachBinlogEnd = false;
SourceRecord lowWatermark = null;
SourceRecord highWatermark = null;
Map<Struct, SourceRecord> snapshotRecords = new LinkedHashMap<>();

SnapshotRecords snapshotRecords =
containsPrimaryKey()
? new PrimaryKeySnapshotRecords()
: new NoPrimaryKeySnapshotRecords();
while (!reachBinlogEnd) {
checkReadException();
List<DataChangeEvent> batch = queue.poll();
Expand All @@ -273,22 +280,15 @@ public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
break;
}

if (!reachBinlogStart) {
snapshotRecords.put((Struct) record.key(), record);
} else {
if (isRequiredBinlogRecord(record)) {
// upsert binlog events through the record key
upsertBinlog(snapshotRecords, record);
}
}
snapshotRecords.collect(record, reachBinlogStart);
}
}
// snapshot split return its data once
hasNextElement.set(false);

final List<SourceRecord> normalizedRecords = new ArrayList<>();
normalizedRecords.add(lowWatermark);
normalizedRecords.addAll(formatMessageTimestamp(snapshotRecords.values()));
normalizedRecords.addAll(formatMessageTimestamp(snapshotRecords.getRecords()));
normalizedRecords.add(highWatermark);

final List<SourceRecords> sourceRecordsSet = new ArrayList<>();
Expand Down Expand Up @@ -403,4 +403,58 @@ public boolean isRunning() {
return currentTaskRunning;
}
}

private boolean containsPrimaryKey() {
return !CollectionUtil.isNullOrEmpty(currentSnapshotSplit.getSplitKeyType().getFields());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is better not to judge in this way. The chunk key will be set when we have to handle the big table in some cases to provide the at-least-once guarantee.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could init the SnapshotRecordCollector when the record.key() returns null.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is better not to judge in this way. The chunk key will be set when we have to handle the big table in some cases to provide the at-least-once guarantee.

Thanks for your review. You mean that the chunk key may be set even though the table has no primary keys?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. If the table contains too much data, there is a performance problem with this solution.
In this case, we may provide the at-least-once semantics instead of the exactly once semantics.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible that the exactly once semantic is required in some scenarios ? How about to provide an option to make it optional for users who can choose between better performance or exactly once semantic?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should control this by the experimental option scan.incremental.snapshot.chunk.key-column.
By default, we will return a single split. If users set the scan.incremental.snapshot.chunk.key-column, we will split the table into multi splits.

}

/** Collect records need to be sent, except low/high watermark. */
interface SnapshotRecords {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SnapshotRecords -> SnapshotRecordCollector

void collect(SourceRecord record, boolean reachBinlogStart);

Collection<SourceRecord> getRecords();
}

/** Collect records with primary key. May upsert binlog events. */
class PrimaryKeySnapshotRecords implements SnapshotRecords {
private final Map<Struct, SourceRecord> snapshotRecords = new LinkedHashMap<>();

@Override
public void collect(SourceRecord record, boolean reachBinlogStart) {
if (!reachBinlogStart) {
snapshotRecords.put((Struct) record.key(), record);
} else {
if (isRequiredBinlogRecord(record)) {
// upsert binlog events through the record key
upsertBinlog(snapshotRecords, record);
}
}
}

@Override
public Collection<SourceRecord> getRecords() {
return snapshotRecords.values();
}
}

/** Collect records without primary key. There are no binlog events. */
static class NoPrimaryKeySnapshotRecords implements SnapshotRecords {
private final List<SourceRecord> snapshotRecords = new LinkedList<>();

@Override
public void collect(SourceRecord record, boolean reachBinlogStart) {
if (!reachBinlogStart) {
snapshotRecords.add(record);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How to handle the delete and update events?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMHO, the snapshot stage for the table without primary key only read snapshot data, then binlog data would be read only in the incremental stage. So there are no delete and update events in SnapshotSplitReader if the table has no primary key. Please correct me.

Copy link
Contributor

@ruanhang1993 ruanhang1993 Apr 21, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I understand right, you want to skip the snapshot backfill task.
It is a good idea to skip the snapshot backfill task when there is a single split. I think you should set this in source but I do not see the relative changes.

} else {
throw new IllegalStateException(
"a table without primary key can not have binlog splits"
+ " in initialization stage");
}
}

@Override
public Collection<SourceRecord> getRecords() {
return snapshotRecords;
}
}
}
7 changes: 6 additions & 1 deletion ...src/main/java/com/ververica/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,12 @@ private void analyzeTable(TableId tableId) {
ChunkUtils.getChunkKeyColumn(
currentSplittingTable, sourceConfig.getChunkKeyColumn());
splitType = ChunkUtils.getChunkKeyColumnType(splitColumn);
minMaxOfSplitColumn = queryMinMax(jdbcConnection, tableId, splitColumn.name());
// split column would be null if table has no primary key.
if (splitColumn != null) {
minMaxOfSplitColumn = queryMinMax(jdbcConnection, tableId, splitColumn.name());
} else {
minMaxOfSplitColumn = new Object[2];
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The chunk splitter should return a total split immediately when the table does not have the primary key. This part should not be changed.

approximateRowCnt = queryApproximateRowCnt(jdbcConnection, tableId);
} catch (Exception e) {
throw new RuntimeException("Fail to analyze table in chunk splitter.", e);
Expand Down
12 changes: 6 additions & 6 deletions ...r-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/ChunkUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,20 +42,20 @@ public static RowType getChunkKeyColumnType(Table table, @Nullable String chunkK
return getChunkKeyColumnType(getChunkKeyColumn(table, chunkKeyColumn));
}

public static RowType getChunkKeyColumnType(Column chunkKeyColumn) {
public static RowType getChunkKeyColumnType(@Nullable Column chunkKeyColumn) {
if (chunkKeyColumn == null) {
return (RowType) ROW().getLogicalType();
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should not be changed as this. I think we could use the first column as the chunk key if the chunk key is not set in the table options.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean using the first column as chunk key to split the table if it has no primary key ?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I replied about this part at the first comment.

return (RowType)
ROW(FIELD(chunkKeyColumn.name(), MySqlTypeUtils.fromDbzColumn(chunkKeyColumn)))
.getLogicalType();
}

@Nullable
public static Column getChunkKeyColumn(Table table, @Nullable String chunkKeyColumn) {
List<Column> primaryKeys = table.primaryKeyColumns();
if (primaryKeys.isEmpty()) {
throw new ValidationException(
String.format(
"Incremental snapshot for tables requires primary key,"
+ " but table %s doesn't have primary key.",
table.id()));
return null;
}

if (chunkKeyColumn != null) {
Expand Down
79 changes: 45 additions & 34 deletions ...-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/source/MySqlSourceITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@
import org.apache.flink.types.RowUtils;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.Collector;
import org.apache.flink.util.ExceptionUtils;

import com.ververica.cdc.connectors.mysql.debezium.DebeziumUtils;
import com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory;
Expand All @@ -57,15 +56,19 @@
import io.debezium.jdbc.JdbcConnection;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.Assume;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.sql.SQLException;
import java.time.Duration;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
Expand All @@ -89,11 +92,14 @@
import static org.junit.Assert.assertTrue;

/** IT tests for {@link MySqlSource}. */
@RunWith(Parameterized.class)
public class MySqlSourceITCase extends MySqlSourceTestBase {

@Rule public final Timeout timeoutPerTest = Timeout.seconds(300);

private static final String DEFAULT_SCAN_STARTUP_MODE = "initial";
private static final String TABLE_WITH_PRIMARY_KEY = "customers";
private static final String TABLE_WITHOUT_PRIMARY_KEY = "customers_no_pk";
private final UniqueDatabase customDatabase =
new UniqueDatabase(MYSQL_CONTAINER, "customer", "mysqluser", "mysqlpw");

Expand Down Expand Up @@ -143,119 +149,122 @@ public class MySqlSourceITCase extends MySqlSourceTestBase {
"+I[2003, user_24, Shanghai, 123567891234]",
"+U[1010, user_11, Hangzhou, 123567891234]");

@Parameterized.Parameter public String tableName;

@Parameterized.Parameters(name = "tableName: {0}")
public static Collection<String> parameters() {
return Arrays.asList(TABLE_WITH_PRIMARY_KEY, TABLE_WITHOUT_PRIMARY_KEY);
}

@Test
public void testReadSingleTableWithSingleParallelism() throws Exception {
testMySqlParallelSource(
1, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"customers"});
1, FailoverType.NONE, FailoverPhase.NEVER, new String[] {tableName});
}

@Test
public void testReadSingleTableWithMultipleParallelism() throws Exception {
testMySqlParallelSource(
4, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"customers"});
4, FailoverType.NONE, FailoverPhase.NEVER, new String[] {tableName});
}

@Test
public void testReadMultipleTableWithSingleParallelism() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
1,
FailoverType.NONE,
FailoverPhase.NEVER,
new String[] {"customers", "customers_1"});
new String[] {"customers", "customers_no_pk"});
}

@Test
public void testReadMultipleTableWithMultipleParallelism() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
4,
FailoverType.NONE,
FailoverPhase.NEVER,
new String[] {"customers", "customers_1"});
new String[] {"customers", "customers_no_pk"});
}

// Failover tests
@Test
public void testTaskManagerFailoverInSnapshotPhase() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {"customers", "customers_1"});
FailoverType.TM,
FailoverPhase.SNAPSHOT,
new String[] {"customers", "customers_no_pk"});
}

@Test
public void testTaskManagerFailoverInBinlogPhase() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
FailoverType.TM, FailoverPhase.BINLOG, new String[] {"customers", "customers_1"});
FailoverType.TM,
FailoverPhase.BINLOG,
new String[] {"customers", "customers_no_pk"});
}

@Test
public void testTaskManagerFailoverFromLatestOffset() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
DEFAULT_PARALLELISM,
"latest-offset",
FailoverType.TM,
FailoverPhase.BINLOG,
new String[] {"customers", "customers_1"},
new String[] {"customers", "customers_no_pk"},
RestartStrategies.fixedDelayRestart(1, 0));
}

@Test
public void testJobManagerFailoverInSnapshotPhase() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"customers", "customers_1"});
FailoverType.JM,
FailoverPhase.SNAPSHOT,
new String[] {"customers", "customers_no_pk"});
}

@Test
public void testJobManagerFailoverInBinlogPhase() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
FailoverType.JM, FailoverPhase.BINLOG, new String[] {"customers", "customers_1"});
FailoverType.JM,
FailoverPhase.BINLOG,
new String[] {"customers", "customers_no_pk"});
}

@Test
public void testJobManagerFailoverFromLatestOffset() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testMySqlParallelSource(
DEFAULT_PARALLELISM,
"latest-offset",
FailoverType.JM,
FailoverPhase.BINLOG,
new String[] {"customers", "customers_1"},
new String[] {"customers", "customers_no_pk"},
RestartStrategies.fixedDelayRestart(1, 0));
}

@Test
public void testTaskManagerFailoverSingleParallelism() throws Exception {
testMySqlParallelSource(
1, FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {"customers"});
1, FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {tableName});
}

@Test
public void testJobManagerFailoverSingleParallelism() throws Exception {
testMySqlParallelSource(
1, FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"customers"});
}

@Test
public void testConsumingTableWithoutPrimaryKey() {
try {
testMySqlParallelSource(
1,
DEFAULT_SCAN_STARTUP_MODE,
FailoverType.NONE,
FailoverPhase.NEVER,
new String[] {"customers_no_pk"},
RestartStrategies.noRestart());
} catch (Exception e) {
assertTrue(
ExceptionUtils.findThrowableWithMessage(
e,
String.format(
"Incremental snapshot for tables requires primary key, but table %s doesn't have primary key",
customDatabase.getDatabaseName() + ".customers_no_pk"))
.isPresent());
}
1, FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {tableName});
}

@Test
@SuppressWarnings({"rawtypes", "unchecked"})
public void testSnapshotSplitReadingFailCrossCheckpoints() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
customDatabase.createAndInitialize();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(DEFAULT_PARALLELISM);
Expand Down Expand Up @@ -329,6 +338,7 @@ public void testSnapshotSplitReadingFailCrossCheckpoints() throws Exception {

@Test
public void testStartFromEarliestOffset() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
List<String> expected = new ArrayList<>();
expected.addAll(initialChanges);
expected.addAll(firstPartBinlogEvents);
Expand All @@ -337,6 +347,7 @@ public void testStartFromEarliestOffset() throws Exception {

@Test
public void testStartFromLatestOffset() throws Exception {
Assume.assumeTrue(TABLE_WITH_PRIMARY_KEY.equals(tableName));
testStartingOffset(StartupOptions.latest(), Collections.emptyList());
}

Expand Down