This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d2a3b48a9c8 [fix](streaming-job) Isolate schemas for CDC snapshot
splits (#65645)
d2a3b48a9c8 is described below
commit d2a3b48a9c8c3180e07031f28b903134134e3247
Author: wudi <[email protected]>
AuthorDate: Mon Jul 20 17:16:57 2026 +0800
[fix](streaming-job) Isolate schemas for CDC snapshot splits (#65645)
### What problem does this PR solve?
During snapshot backfill, a snapshot split carried schemas for every
captured table. A change record from another table could then be
evaluated with the current split key, causing an invalid-field failure
when the tables use different primary-key names.
This change limits each MySQL and PostgreSQL snapshot split to its own
table schema while retaining the reader-level schemas for the global
stream split. It also strengthens the multi-table concurrent-DML cases
by explicitly enabling snapshot backfill, using different primary-key
names, and keeping a source writer active across multiple successful
tasks.
---
.../source/reader/JdbcIncrementalSourceReader.java | 2 +-
.../source/reader/mysql/MySqlSourceReader.java | 2 +-
.../source/reader/mysql/MySqlSourceReaderTest.java | 73 +++++++++++++++++
.../reader/postgres/PostgresSourceReaderTest.java | 74 +++++++++++++++++
...ming_mysql_job_snapshot_with_concurrent_dml.out | 23 ++++++
...g_mysql_job_snapshot_with_concurrent_dml.groovy | 95 ++++++++++++++++++----
...snapshot_with_concurrent_dml_multi_table.groovy | 71 ++++++++++------
7 files changed, 298 insertions(+), 42 deletions(-)
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
index a6887ce8a96..b4c7dcada31 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
@@ -733,7 +733,7 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
splitStart,
splitEnd,
null,
- tableSchemas);
+ Collections.singletonMap(tableId, tableChange));
return split;
}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index 430ab7b6668..53c74af0cb6 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -743,7 +743,7 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
splitStart,
splitEnd,
null,
- tableSchemas);
+ Collections.singletonMap(tableId, tableChange));
return split;
}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
index fe17794f030..2422d6c7c98 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
@@ -24,16 +24,26 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.JobBaseConfig;
import com.github.shyiko.mysql.binlog.GtidSet;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
+import java.sql.Types;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableEditor;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+
public class MySqlSourceReaderTest {
private static final String SERVER_UUID =
"24bc7850-2c16-11e6-a073-0242ac110002";
@@ -115,6 +125,46 @@ public class MySqlSourceReaderTest {
assertFalse(config.isIncludeSchemaChanges());
}
+ @Test
+ void snapshotSplitContainsOnlyCurrentTableSchema() throws Exception {
+ TableId tableId = TableId.parse("testdb.orders");
+ TableId otherTableId = TableId.parse("testdb.other_orders");
+ TableChanges.TableChange tableChange = tableChange(tableId, "pk");
+ Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();
+ tableSchemas.put(tableId, tableChange);
+ tableSchemas.put(otherTableId, tableChange(otherTableId, "other_id"));
+
+ MySqlSourceReader reader =
+ new MySqlSourceReader() {
+ @Override
+ protected Class<?> probeSplitKeyClass(
+ TableId ignoredTableId,
+ Column ignoredSplitColumn,
+ JobBaseConfig ignoredJobConfig) {
+ return Integer.class;
+ }
+ };
+ reader.setTableSchemas(tableSchemas);
+
+ Method method =
+ MySqlSourceReader.class.getDeclaredMethod(
+ "createSnapshotSplit", Map.class, JobBaseConfig.class);
+ method.setAccessible(true);
+ MySqlSnapshotSplit split =
+ (MySqlSnapshotSplit)
+ method.invoke(
+ reader,
+ snapshotOffset("testdb.orders", "pk"),
+ new JobBaseConfig("job-1", "MYSQL", Map.of(),
null));
+
+ assertEquals(tableId, split.getTableId());
+ assertEquals(Map.of(tableId, tableChange), split.getTableSchemas());
+ assertFalse(split.getTableSchemas().containsKey(otherTableId));
+ assertEquals(2, reader.getTableSchemas().size());
+ assertTrue(reader.getTableSchemas().containsKey(tableId));
+ assertTrue(reader.getTableSchemas().containsKey(otherTableId));
+ }
+
// Drive the real generateMySqlConfig JSON-offset path and return the
rebuilt startup offset.
private BinlogOffset startupBinlogOffset(String offsetJson) throws
Exception {
return sourceConfig(offsetJson).getStartupOptions().binlogOffset;
@@ -140,4 +190,27 @@ public class MySqlSourceReaderTest {
m.setAccessible(true);
return (MySqlSourceConfig) m.invoke(new MySqlSourceReader(), cfg,
"job-1", 0);
}
+
+ private static Map<String, Object> snapshotOffset(String tableId, String
splitKey) {
+ Map<String, Object> offset = new HashMap<>();
+ offset.put("splitId", tableId + ":0");
+ offset.put("tableId", tableId);
+ offset.put("splitKey", List.of(splitKey));
+ offset.put("splitStart", new Object[] {1});
+ offset.put("splitEnd", new Object[] {10});
+ return offset;
+ }
+
+ private static TableChanges.TableChange tableChange(TableId tableId,
String splitKey) {
+ TableEditor editor = Table.editor().tableId(tableId);
+ editor.addColumns(
+ Column.editor()
+ .name(splitKey)
+ .type("INT")
+ .jdbcType(Types.INTEGER)
+ .optional(false)
+ .create());
+ editor.setPrimaryKeyNames(splitKey);
+ return new
TableChanges.TableChange(TableChanges.TableChangeType.CREATE, editor.create());
+ }
}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReaderTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReaderTest.java
index f9f0efe9a73..c5119bc8ab2 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReaderTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReaderTest.java
@@ -17,22 +17,33 @@
package org.apache.doris.cdcclient.source.reader.postgres;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.doris.cdcclient.source.reader.JdbcIncrementalSourceReader;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.JobBaseConfig;
import io.debezium.config.Configuration;
import io.debezium.connector.postgresql.connection.PostgresConnection;
import io.debezium.jdbc.JdbcConfiguration;
import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableEditor;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
import
org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.sql.Types;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
class PostgresSourceReaderTest {
@@ -72,6 +83,46 @@ class PostgresSourceReaderTest {
assertFalse(config.isIncludeSchemaChanges());
}
+ @Test
+ void snapshotSplitContainsOnlyCurrentTableSchema() throws Exception {
+ TableId tableId = TableId.parse("public.orders", false);
+ TableId otherTableId = TableId.parse("public.other_orders", false);
+ TableChanges.TableChange tableChange = tableChange(tableId, "pk");
+ Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();
+ tableSchemas.put(tableId, tableChange);
+ tableSchemas.put(otherTableId, tableChange(otherTableId, "other_id"));
+
+ PostgresSourceReader reader =
+ new PostgresSourceReader() {
+ @Override
+ protected Class<?> probeSplitKeyClass(
+ TableId ignoredTableId,
+ Column ignoredSplitColumn,
+ JobBaseConfig ignoredJobConfig) {
+ return Integer.class;
+ }
+ };
+ reader.setTableSchemas(tableSchemas);
+
+ Method method =
+ JdbcIncrementalSourceReader.class.getDeclaredMethod(
+ "createSnapshotSplit", Map.class, JobBaseConfig.class);
+ method.setAccessible(true);
+ SnapshotSplit split =
+ (SnapshotSplit)
+ method.invoke(
+ reader,
+ snapshotOffset("public.orders", "pk"),
+ new JobBaseConfig("job-1", "POSTGRES",
Map.of(), null));
+
+ assertEquals(tableId, split.getTableId());
+ assertEquals(Map.of(tableId, tableChange), split.getTableSchemas());
+ assertFalse(split.getTableSchemas().containsKey(otherTableId));
+ assertEquals(2, reader.getTableSchemas().size());
+ assertTrue(reader.getTableSchemas().containsKey(tableId));
+ assertTrue(reader.getTableSchemas().containsKey(otherTableId));
+ }
+
private static PostgresSourceConfig sourceConfig(Map<String, String>
overrides)
throws Exception {
Map<String, String> cfg = new HashMap<>();
@@ -97,4 +148,27 @@ class PostgresSourceReaderTest {
field.setAccessible(true);
return field.get(connection);
}
+
+ private static Map<String, Object> snapshotOffset(String tableId, String
splitKey) {
+ Map<String, Object> offset = new HashMap<>();
+ offset.put("splitId", tableId + ":0");
+ offset.put("tableId", tableId);
+ offset.put("splitKey", List.of(splitKey));
+ offset.put("splitStart", new Object[] {1});
+ offset.put("splitEnd", new Object[] {10});
+ return offset;
+ }
+
+ private static TableChanges.TableChange tableChange(TableId tableId,
String splitKey) {
+ TableEditor editor = Table.editor().tableId(tableId);
+ editor.addColumns(
+ Column.editor()
+ .name(splitKey)
+ .type("int4")
+ .jdbcType(Types.INTEGER)
+ .optional(false)
+ .create());
+ editor.setPrimaryKeyNames(splitKey);
+ return new
TableChanges.TableChange(TableChanges.TableChangeType.CREATE, editor.create());
+ }
}
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.out
index 7f1b97bd503..e97d6041b53 100644
---
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.out
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.out
@@ -2,15 +2,27 @@
-- !select_count --
1007
+-- !select_count_t2 --
+1007
+
-- !select_updates --
1 99
100 99
500 99
999 99
+-- !select_updates_t2 --
+1 99
+100 99
+500 99
+999 99
+
-- !select_deletes --
0
+-- !select_deletes_t2 --
+0
+
-- !select_inserts --
1001 concurrent_ins 1
1002 concurrent_ins 1
@@ -23,3 +35,14 @@
1009 concurrent_ins 1
1010 concurrent_ins 1
+-- !select_inserts_t2 --
+1001 concurrent_ins 1
+1002 concurrent_ins 1
+1003 concurrent_ins 1
+1004 concurrent_ins 1
+1005 concurrent_ins 1
+1006 concurrent_ins 1
+1007 concurrent_ins 1
+1008 concurrent_ins 1
+1009 concurrent_ins 1
+1010 concurrent_ins 1
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.groovy
index 597bea37320..b387fa5dd81 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_with_concurrent_dml.groovy
@@ -18,19 +18,22 @@
import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.MILLISECONDS
import static java.util.concurrent.TimeUnit.SECONDS
suite("test_streaming_mysql_job_snapshot_with_concurrent_dml",
"p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
def jobName = "test_streaming_mysql_job_snapshot_with_concurrent_dml_name"
def currentDb = (sql "select database()")[0][0]
def table1 = "streaming_snapshot_dml_mysql"
- def unrelated = "streaming_snapshot_dml_unrelated_mysql"
+ def table2 = "streaming_snapshot_dml_other_pk_mysql"
+ def excludedTable = "streaming_snapshot_dml_excluded_mysql"
def mysqlDb = "test_cdc_db"
def totalRows = 1000
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
sql """drop table if exists ${currentDb}.${table1} force"""
- sql """drop table if exists ${currentDb}.${unrelated} force"""
+ sql """drop table if exists ${currentDb}.${table2} force"""
+ sql """drop table if exists ${currentDb}.${excludedTable} force"""
String enabled = context.config.otherConfigs.get("enableJdbcTest")
if (enabled != null && enabled.equalsIgnoreCase("true")) {
@@ -59,17 +62,33 @@
suite("test_streaming_mysql_job_snapshot_with_concurrent_dml", "p0,external,mysq
}
sql sb.toString()
- sql """DROP TABLE IF EXISTS ${mysqlDb}.${unrelated}"""
- sql """CREATE TABLE ${mysqlDb}.${unrelated} (
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${table2}"""
+ sql """CREATE TABLE ${mysqlDb}.${table2} (
+ `other_id` int NOT NULL,
+ `tag` varchar(64),
+ `version` int,
+ PRIMARY KEY (`other_id`)
+ ) ENGINE=InnoDB"""
+
+ StringBuilder table2Rows = new StringBuilder()
+ table2Rows.append("INSERT INTO ${mysqlDb}.${table2} (other_id,
tag, version) VALUES ")
+ for (int i = 1; i <= totalRows; i++) {
+ if (i > 1) table2Rows.append(", ")
+ table2Rows.append("(${i}, 'snap', 0)")
+ }
+ sql table2Rows.toString()
+
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${excludedTable}"""
+ sql """CREATE TABLE ${mysqlDb}.${excludedTable} (
`id` int NOT NULL,
`tag` varchar(64),
PRIMARY KEY (`id`)
) ENGINE=InnoDB"""
- sql """INSERT INTO ${mysqlDb}.${unrelated} (id, tag) VALUES (1,
'pre_snap')"""
+ sql """INSERT INTO ${mysqlDb}.${excludedTable} (id, tag) VALUES
(1, 'pre_snap')"""
}
- // snapshot_split_size=10 + snapshot_parallelism=1 -> 100 serial
splits, slow enough that
- // the concurrent DML below actually overlaps with snapshot.
+ // Two tables with different primary-key names make a foreign-table
backfill record
+ // incompatible with the current split key. Serial small splits keep
the snapshot active.
sql """CREATE JOB ${jobName}
ON STREAMING
FROM MYSQL (
@@ -79,27 +98,54 @@
suite("test_streaming_mysql_job_snapshot_with_concurrent_dml", "p0,external,mysq
"user" = "root",
"password" = "123456",
"database" = "${mysqlDb}",
- "include_tables" = "${table1}",
+ "include_tables" = "${table1},${table2}",
"offset" = "initial",
"snapshot_split_size" = "10",
- "snapshot_parallelism" = "1"
+ "snapshot_parallelism" = "1",
+ "skip_snapshot_backfill" = "false"
)
TO DATABASE ${currentDb} (
"table.create.properties.replication_num" = "1"
)
"""
- // Concurrent DML on source while cdc-client is still snapshotting.
+ def succeedTaskCount = {
+ def rows = sql """select SucceedTaskCount from
jobs("type"="insert") where Name='${jobName}' and ExecuteType='STREAMING'"""
+ rows.size() == 1 ? (rows.get(0).get(0).toString() as long) : 0L
+ }
+ Awaitility.await().atMost(120, SECONDS).pollInterval(100,
MILLISECONDS).until({
+ succeedTaskCount() >= 1
+ })
+
+ def writeProbeDml = {
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
+ sql """UPDATE ${mysqlDb}.${table1} SET version=version+1 WHERE
id=500"""
+ sql """UPDATE ${mysqlDb}.${table2} SET version=version+1 WHERE
other_id=500"""
+ }
+ }
+
+ writeProbeDml()
+ long probeStartTaskCount = succeedTaskCount()
+ long probeTargetTaskCount = probeStartTaskCount + 10
+ Awaitility.await().atMost(120, SECONDS).pollInterval(200,
MILLISECONDS).until({
+ writeProbeDml()
+ succeedTaskCount() >= probeTargetTaskCount
+ })
+ log.info("MySQL probe DML covered tasks
${probeStartTaskCount}..${succeedTaskCount()}")
+
+ // Apply deterministic DML after the probe updates so final results
remain stable.
connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
for (int i = 1; i <= 10; i++) {
sql """INSERT INTO ${mysqlDb}.${table1} (id, tag, version)
VALUES (${totalRows + i}, 'concurrent_ins', 1)"""
+ sql """INSERT INTO ${mysqlDb}.${table2} (other_id, tag,
version) VALUES (${totalRows + i}, 'concurrent_ins', 1)"""
}
sql """UPDATE ${mysqlDb}.${table1} SET version=99 WHERE id IN (1,
100, 500, 999)"""
sql """DELETE FROM ${mysqlDb}.${table1} WHERE id IN (2, 200,
800)"""
+ sql """UPDATE ${mysqlDb}.${table2} SET version=99 WHERE other_id
IN (1, 100, 500, 999)"""
+ sql """DELETE FROM ${mysqlDb}.${table2} WHERE other_id IN (2, 200,
800)"""
- // DML on unrelated table - must NOT leak into Doris (not in
include_tables).
- sql """INSERT INTO ${mysqlDb}.${unrelated} (id, tag) VALUES (2,
'concurrent_unrelated_ins')"""
- sql """UPDATE ${mysqlDb}.${unrelated} SET
tag='concurrent_unrelated_upd' WHERE id=1"""
+ sql """INSERT INTO ${mysqlDb}.${excludedTable} (id, tag) VALUES
(2, 'concurrent_excluded_ins')"""
+ sql """UPDATE ${mysqlDb}.${excludedTable} SET
tag='concurrent_excluded_upd' WHERE id=1"""
}
def expectedRows = totalRows + 10 - 3
@@ -113,15 +159,26 @@
suite("test_streaming_mysql_job_snapshot_with_concurrent_dml", "p0,external,mysq
def del2 = sql """select count(1) from
${currentDb}.${table1} where id=2"""
def del800 = sql """select count(1) from
${currentDb}.${table1} where id=800"""
def ins1010 = sql """select count(1) from
${currentDb}.${table1} where id=${totalRows + 10}"""
+ def cnt2 = sql """select count(1) from
${currentDb}.${table2}"""
+ def upd2 = sql """select version from
${currentDb}.${table2} where other_id=999"""
+ def del2Table2 = sql """select count(1) from
${currentDb}.${table2} where other_id=2"""
+ def ins2 = sql """select count(1) from
${currentDb}.${table2} where other_id=${totalRows + 10}"""
def v1 = upd1.size() == 0 ? null : upd1.get(0).get(0)
def v999 = upd999.size() == 0 ? null :
upd999.get(0).get(0)
- log.info("incr cnt=${cnt} v1=${v1} v999=${v999}
del2=${del2} del800=${del800} ins1010=${ins1010}")
+ def v2 = upd2.size() == 0 ? null : upd2.get(0).get(0)
+ log.info("incr cnt=${cnt} cnt2=${cnt2} v1=${v1}
v999=${v999} v2=${v2} "
+ + "del2=${del2} del800=${del800}
del2Table2=${del2Table2} "
+ + "ins1010=${ins1010} ins2=${ins2}")
cnt.get(0).get(0) == expectedRows &&
+ cnt2.get(0).get(0) == expectedRows &&
v1 != null && v1.toString() == '99' &&
v999 != null && v999.toString() == '99' &&
+ v2 != null && v2.toString() == '99' &&
del2.get(0).get(0) == 0 &&
del800.get(0).get(0) == 0 &&
- ins1010.get(0).get(0) == 1
+ del2Table2.get(0).get(0) == 0 &&
+ ins1010.get(0).get(0) == 1 &&
+ ins2.get(0).get(0) == 1
}
)
} catch (Exception ex) {
@@ -132,13 +189,17 @@
suite("test_streaming_mysql_job_snapshot_with_concurrent_dml", "p0,external,mysq
throw ex
}
- def showUnrelated = sql """show tables from ${currentDb} like
'${unrelated}'"""
- assert showUnrelated.size() == 0
+ def showExcluded = sql """show tables from ${currentDb} like
'${excludedTable}'"""
+ assert showExcluded.size() == 0
qt_select_count """select count(1) from ${currentDb}.${table1}"""
+ qt_select_count_t2 """select count(1) from ${currentDb}.${table2}"""
qt_select_updates """select id, version from ${currentDb}.${table1}
where id in (1, 100, 500, 999) order by id"""
+ qt_select_updates_t2 """select other_id, version from
${currentDb}.${table2} where other_id in (1, 100, 500, 999) order by other_id"""
qt_select_deletes """select count(1) from ${currentDb}.${table1} where
id in (2, 200, 800)"""
+ qt_select_deletes_t2 """select count(1) from ${currentDb}.${table2}
where other_id in (2, 200, 800)"""
qt_select_inserts """select id, tag, version from
${currentDb}.${table1} where id > ${totalRows} order by id"""
+ qt_select_inserts_t2 """select other_id, tag, version from
${currentDb}.${table2} where other_id > ${totalRows} order by other_id"""
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table.groovy
index 6ecdffdea8d..6dd528a3a7e 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table.groovy
@@ -18,6 +18,7 @@
import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.MILLISECONDS
import static java.util.concurrent.TimeUnit.SECONDS
// Multi-table from-to snapshot + concurrent DML: with >=2 tables each
snapshot split flips the
@@ -31,6 +32,7 @@
suite("test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table",
def table1 = "streaming_snapshot_dml_multi_pg_t1"
def table2 = "streaming_snapshot_dml_multi_pg_t2"
def tables = [table1, table2]
+ def primaryKeys = [(table1): "pk", (table2): "other_id"]
def pgDB = "postgres"
def pgSchema = "cdc_test"
def pgUser = "postgres"
@@ -51,21 +53,22 @@
suite("test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table",
// ===== Prepare PG side: two tables, each 1000 snapshot rows =====
connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
tables.each { t ->
+ def primaryKey = primaryKeys[t]
sql """DROP TABLE IF EXISTS ${pgDB}.${pgSchema}.${t}"""
sql """
create table ${pgDB}.${pgSchema}.${t} (
- id integer PRIMARY KEY,
+ ${primaryKey} integer PRIMARY KEY,
tag varchar(64),
version integer
);
"""
- sql """INSERT INTO ${pgDB}.${pgSchema}.${t} (id, tag, version)
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${t} (${primaryKey},
tag, version)
SELECT g, 'snap', 0 FROM generate_series(1,
${totalRows}) g"""
}
}
- // snapshot_split_size=10 + snapshot_parallelism=1 -> 100 serial
splits per table, slow
- // enough that the concurrent DML overlaps snapshot while the
publication keeps flipping.
+ // Different primary-key names make a foreign-table record
incompatible with the current
+ // split key. Serial small splits keep the snapshot active during
concurrent DML.
sql """CREATE JOB ${jobName}
ON STREAMING
FROM POSTGRES (
@@ -79,28 +82,49 @@
suite("test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table",
"include_tables" = "${table1},${table2}",
"offset" = "initial",
"snapshot_split_size" = "10",
- "snapshot_parallelism" = "1"
+ "snapshot_parallelism" = "1",
+ "skip_snapshot_backfill" = "false"
)
TO DATABASE ${currentDb} (
"table.create.properties.replication_num" = "1"
)
"""
- // Wait until the first snapshot split commits (slot created, snapshot
in progress) so the
- // DML below lands inside the snapshot window and overlaps the
publication flipping.
- Awaitility.await().atMost(120, SECONDS).pollInterval(1,
SECONDS).until({
- def c = sql """select SucceedTaskCount from jobs("type"="insert")
where Name='${jobName}' and ExecuteType='STREAMING'"""
- c.size() == 1 && (c.get(0).get(0).toString() as long) >= 1
+ // Wait until snapshot processing starts, then keep writing while
subsequent splits run.
+ def succeedTaskCount = {
+ def rows = sql """select SucceedTaskCount from
jobs("type"="insert") where Name='${jobName}' and ExecuteType='STREAMING'"""
+ rows.size() == 1 ? (rows.get(0).get(0).toString() as long) : 0L
+ }
+ Awaitility.await().atMost(120, SECONDS).pollInterval(100,
MILLISECONDS).until({
+ succeedTaskCount() >= 1
+ })
+
+ def writeProbeDml = {
+ connect("${pgUser}", "${pgPassword}",
+ "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ sql """UPDATE ${pgDB}.${pgSchema}.${table1} SET
version=version+1 WHERE pk=500"""
+ sql """UPDATE ${pgDB}.${pgSchema}.${table2} SET
version=version+1 WHERE other_id=500"""
+ }
+ }
+
+ writeProbeDml()
+ long probeStartTaskCount = succeedTaskCount()
+ long probeTargetTaskCount = probeStartTaskCount + 10
+ Awaitility.await().atMost(120, SECONDS).pollInterval(200,
MILLISECONDS).until({
+ writeProbeDml()
+ succeedTaskCount() >= probeTargetTaskCount
})
+ log.info("PostgreSQL probe DML covered tasks
${probeStartTaskCount}..${succeedTaskCount()}")
- // Concurrent DML on BOTH tables while still snapshotting. Same DML
shape on each table.
+ // Apply deterministic DML after the probe updates so final results
remain stable.
connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
tables.each { t ->
+ def primaryKey = primaryKeys[t]
for (int i = 1; i <= 10; i++) {
- sql """INSERT INTO ${pgDB}.${pgSchema}.${t} (id, tag,
version) VALUES (${totalRows + i}, 'concurrent_ins', 1)"""
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${t}
(${primaryKey}, tag, version) VALUES (${totalRows + i}, 'concurrent_ins', 1)"""
}
- sql """UPDATE ${pgDB}.${pgSchema}.${t} SET version=99 WHERE id
IN (1, 100, 500, 999)"""
- sql """DELETE FROM ${pgDB}.${pgSchema}.${t} WHERE id IN (2,
200, 800)"""
+ sql """UPDATE ${pgDB}.${pgSchema}.${t} SET version=99 WHERE
${primaryKey} IN (1, 100, 500, 999)"""
+ sql """DELETE FROM ${pgDB}.${pgSchema}.${t} WHERE
${primaryKey} IN (2, 200, 800)"""
}
}
@@ -112,17 +136,18 @@
suite("test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table",
{
boolean allOk = true
for (def t : tables) {
+ def primaryKey = primaryKeys[t]
def showTbl = sql """show tables from ${currentDb}
like '${t}'"""
if (showTbl.size() == 0) {
allOk = false
break
}
def cnt = sql """select count(1) from
${currentDb}.${t}"""
- def upd1 = sql """select version from
${currentDb}.${t} where id=1"""
- def upd999 = sql """select version from
${currentDb}.${t} where id=999"""
- def del2 = sql """select count(1) from
${currentDb}.${t} where id=2"""
- def del800 = sql """select count(1) from
${currentDb}.${t} where id=800"""
- def ins = sql """select count(1) from
${currentDb}.${t} where id=${totalRows + 10}"""
+ def upd1 = sql """select version from
${currentDb}.${t} where ${primaryKey}=1"""
+ def upd999 = sql """select version from
${currentDb}.${t} where ${primaryKey}=999"""
+ def del2 = sql """select count(1) from
${currentDb}.${t} where ${primaryKey}=2"""
+ def del800 = sql """select count(1) from
${currentDb}.${t} where ${primaryKey}=800"""
+ def ins = sql """select count(1) from
${currentDb}.${t} where ${primaryKey}=${totalRows + 10}"""
def v1 = upd1.size() == 0 ? null :
upd1.get(0).get(0)
def v999 = upd999.size() == 0 ? null :
upd999.get(0).get(0)
log.info("table=${t} cnt=${cnt} v1=${v1}
v999=${v999} del2=${del2} del800=${del800} ins=${ins}")
@@ -150,10 +175,10 @@
suite("test_streaming_postgres_job_snapshot_with_concurrent_dml_multi_table",
qt_select_count_t1 """select count(1) from ${currentDb}.${table1}"""
qt_select_count_t2 """select count(1) from ${currentDb}.${table2}"""
- qt_select_updates_t1 """select id, version from ${currentDb}.${table1}
where id in (1, 100, 500, 999) order by id"""
- qt_select_updates_t2 """select id, version from ${currentDb}.${table2}
where id in (1, 100, 500, 999) order by id"""
- qt_select_inserts_t1 """select id, tag, version from
${currentDb}.${table1} where id > ${totalRows} order by id"""
- qt_select_inserts_t2 """select id, tag, version from
${currentDb}.${table2} where id > ${totalRows} order by id"""
+ qt_select_updates_t1 """select pk, version from ${currentDb}.${table1}
where pk in (1, 100, 500, 999) order by pk"""
+ qt_select_updates_t2 """select other_id, version from
${currentDb}.${table2} where other_id in (1, 100, 500, 999) order by other_id"""
+ qt_select_inserts_t1 """select pk, tag, version from
${currentDb}.${table1} where pk > ${totalRows} order by pk"""
+ qt_select_inserts_t2 """select other_id, tag, version from
${currentDb}.${table2} where other_id > ${totalRows} order by other_id"""
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
