github-actions[bot] commented on code in PR #65325:
URL: https://github.com/apache/doris/pull/65325#discussion_r3548702951
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -59,8 +75,175 @@ public DeserializeResult deserialize(Map<String, String>
context, SourceRecord r
}
private DeserializeResult handleSchemaChangeEvent(
- SourceRecord record, Map<String, String> context) {
- // todo: record has mysql ddl, need to convert doris ddl
- return DeserializeResult.empty();
+ SourceRecord record, Map<String, String> context) throws
IOException {
+ HistoryRecord history = RecordUtils.getHistoryRecord(record);
+ String database =
history.document().getString(HistoryRecord.Fields.DATABASE_NAME);
+ String ddl =
history.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
+ Map<TableId, TableChanges.TableChange> freshSchemas =
deserializeTableChanges(history);
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL deserializer received schema change,
baselineSchemas={}, freshSchemas={}, ddl={}",
+ tableSchemas == null ? 0 : tableSchemas.size(),
+ freshSchemas.size(),
+ ddl);
+ if (freshSchemas.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event has no
table schema "
+ + "change. DDL: {}",
+ ddl);
+ return DeserializeResult.empty();
+ }
+
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL schema baseline is empty, adopting
fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL: {}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (TableId tableId : freshSchemas.keySet()) {
+ if (!tableSchemas.containsKey(tableId)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {} has no baseline,
adopting fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL:
{}",
+ tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ Tables tables = toTables(tableSchemas);
+ List<MySqlSchemaChange> changes = Collections.emptyList();
+ try {
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser();
+ parser.setCurrentDatabase(database);
+ parser.parse(ddl, tables);
+ changes = parser.getAndClearParsedChanges();
+ } catch (Exception e) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change DDL parser
failed. No Doris DDL "
+ + "emitted, FE baseline will advance to the schema
carried by the "
+ + "history record. DDL: {}",
+ ddl,
+ e);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ if (changes.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event
produced no supported "
+ + "column change. No Doris DDL emitted, FE
baseline will advance to "
+ + "the schema carried by the history record. DDL:
{}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+
+ List<MySqlSchemaChange> supportedChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : changes) {
+ if (change.getType() == MySqlSchemaChange.Type.UNSUPPORTED) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL unsupported schema
change {} for table {}: {}."
+ + " Doris DDL will not be emitted for this
change, FE baseline will"
+ + " advance to the schema carried by the
history record. DDL: {}",
+ change.getSourceOperation(),
+ change.getTableId() == null
+ ? "<unknown>"
+ : change.getTableId().identifier(),
+ change.getReason(),
+ ddl);
+ continue;
+ }
+ supportedChanges.add(change);
+ }
+ if (supportedChanges.isEmpty()) {
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ if (tableId == null || !tableSchemas.containsKey(tableId)) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change table {}
has no baseline."
+ + " No Doris DDL emitted, FE baseline will
advance to the schema"
+ + " carried by the history record. DDL: {}",
+ tableId == null ? "<unknown>" : tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ List<SchemaChangeOperation> schemaChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ Set<String> excludedCols =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ String targetTable = resolveTargetTable(tableId.table());
+ String db = context.get(Constants.DORIS_TARGET_DB);
+ if (change.getType() == MySqlSchemaChange.Type.ADD) {
+ String columnName = change.getColumn().name();
+ if (excludedCols.contains(columnName)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {}: added column '{}'
is excluded, skipping ADD",
+ tableId.identifier(),
+ columnName);
+ continue;
+ }
+ String colType =
SchemaChangeHelper.mysqlColumnToDorisType(change.getColumn());
+ schemaChanges.add(
+ SchemaChangeOperation.addColumn(
+ targetTable,
+ columnName,
+ SchemaChangeHelper.buildAddColumnSql(
Review Comment:
This path loses the MySQL column-position part of supported ADD COLUMN
statements. The parser test/regression exercises `ADD COLUMN first_col ...
FIRST` and `ADD COLUMN after_name ... AFTER name`, but the emitted Doris DDL is
built as `ALTER TABLE ... ADD COLUMN name type` with only an optional comment.
Doris therefore appends the columns while `freshSchemas` advances the persisted
FE baseline to the source schema order. JSON stream load is name-based, but
user-visible schema order (`DESC`, `SHOW CREATE TABLE`, `SELECT *`) now
diverges from the source and from the stored baseline. Please carry the parsed
`FIRST`/`AFTER` position into `MySqlSchemaChange` and append the matching Doris
position clause, or reject/skip positional ADDs instead of treating them as
fully applied.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -59,8 +75,175 @@ public DeserializeResult deserialize(Map<String, String>
context, SourceRecord r
}
private DeserializeResult handleSchemaChangeEvent(
- SourceRecord record, Map<String, String> context) {
- // todo: record has mysql ddl, need to convert doris ddl
- return DeserializeResult.empty();
+ SourceRecord record, Map<String, String> context) throws
IOException {
+ HistoryRecord history = RecordUtils.getHistoryRecord(record);
+ String database =
history.document().getString(HistoryRecord.Fields.DATABASE_NAME);
+ String ddl =
history.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
+ Map<TableId, TableChanges.TableChange> freshSchemas =
deserializeTableChanges(history);
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL deserializer received schema change,
baselineSchemas={}, freshSchemas={}, ddl={}",
+ tableSchemas == null ? 0 : tableSchemas.size(),
+ freshSchemas.size(),
+ ddl);
+ if (freshSchemas.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event has no
table schema "
+ + "change. DDL: {}",
+ ddl);
+ return DeserializeResult.empty();
+ }
+
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL schema baseline is empty, adopting
fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL: {}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (TableId tableId : freshSchemas.keySet()) {
+ if (!tableSchemas.containsKey(tableId)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {} has no baseline,
adopting fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL:
{}",
+ tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ Tables tables = toTables(tableSchemas);
+ List<MySqlSchemaChange> changes = Collections.emptyList();
+ try {
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser();
+ parser.setCurrentDatabase(database);
+ parser.parse(ddl, tables);
+ changes = parser.getAndClearParsedChanges();
+ } catch (Exception e) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change DDL parser
failed. No Doris DDL "
+ + "emitted, FE baseline will advance to the schema
carried by the "
+ + "history record. DDL: {}",
+ ddl,
+ e);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ if (changes.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event
produced no supported "
+ + "column change. No Doris DDL emitted, FE
baseline will advance to "
+ + "the schema carried by the history record. DDL:
{}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+
+ List<MySqlSchemaChange> supportedChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : changes) {
+ if (change.getType() == MySqlSchemaChange.Type.UNSUPPORTED) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL unsupported schema
change {} for table {}: {}."
+ + " Doris DDL will not be emitted for this
change, FE baseline will"
+ + " advance to the schema carried by the
history record. DDL: {}",
+ change.getSourceOperation(),
+ change.getTableId() == null
+ ? "<unknown>"
+ : change.getTableId().identifier(),
+ change.getReason(),
+ ddl);
+ continue;
+ }
+ supportedChanges.add(change);
+ }
+ if (supportedChanges.isEmpty()) {
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ if (tableId == null || !tableSchemas.containsKey(tableId)) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change table {}
has no baseline."
+ + " No Doris DDL emitted, FE baseline will
advance to the schema"
+ + " carried by the history record. DDL: {}",
+ tableId == null ? "<unknown>" : tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ List<SchemaChangeOperation> schemaChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ Set<String> excludedCols =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ String targetTable = resolveTargetTable(tableId.table());
+ String db = context.get(Constants.DORIS_TARGET_DB);
Review Comment:
This lookup only works on the from-to write path, where `writeRecords`
injects `DORIS_TARGET_DB` before deserializing. The fetch/TVF paths still call
`sourceReader.deserialize(fetchRecord.getConfig(), element)` with only source
connection properties, and `CdcStreamTableValuedFunction` does not put a Doris
target DB into the `FetchRecordRequest`. Since this PR also enables MySQL
`includeSchemaChanges(true)` and returns schema-change events from
`FilteredRecordIterator`, a binlog TVF/fetch that sees `ADD`/`DROP` with an
existing baseline reaches this code with `db == null`;
`SchemaChangeHelper.quoteTableIdentifier` then dereferences it while building
DDL that the fetch path cannot execute anyway. Please keep automatic MySQL DDL
generation on the write path, or otherwise handle schema-change records without
requiring `DORIS_TARGET_DB` for `/api/fetchRecordStream` and debug fetch
callers.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -59,8 +75,175 @@ public DeserializeResult deserialize(Map<String, String>
context, SourceRecord r
}
private DeserializeResult handleSchemaChangeEvent(
- SourceRecord record, Map<String, String> context) {
- // todo: record has mysql ddl, need to convert doris ddl
- return DeserializeResult.empty();
+ SourceRecord record, Map<String, String> context) throws
IOException {
+ HistoryRecord history = RecordUtils.getHistoryRecord(record);
+ String database =
history.document().getString(HistoryRecord.Fields.DATABASE_NAME);
+ String ddl =
history.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
+ Map<TableId, TableChanges.TableChange> freshSchemas =
deserializeTableChanges(history);
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL deserializer received schema change,
baselineSchemas={}, freshSchemas={}, ddl={}",
+ tableSchemas == null ? 0 : tableSchemas.size(),
+ freshSchemas.size(),
+ ddl);
+ if (freshSchemas.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event has no
table schema "
+ + "change. DDL: {}",
+ ddl);
+ return DeserializeResult.empty();
+ }
+
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL schema baseline is empty, adopting
fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL: {}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (TableId tableId : freshSchemas.keySet()) {
+ if (!tableSchemas.containsKey(tableId)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {} has no baseline,
adopting fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL:
{}",
+ tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ Tables tables = toTables(tableSchemas);
+ List<MySqlSchemaChange> changes = Collections.emptyList();
+ try {
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser();
+ parser.setCurrentDatabase(database);
+ parser.parse(ddl, tables);
+ changes = parser.getAndClearParsedChanges();
+ } catch (Exception e) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change DDL parser
failed. No Doris DDL "
+ + "emitted, FE baseline will advance to the schema
carried by the "
+ + "history record. DDL: {}",
+ ddl,
+ e);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ if (changes.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event
produced no supported "
+ + "column change. No Doris DDL emitted, FE
baseline will advance to "
+ + "the schema carried by the history record. DDL:
{}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+
+ List<MySqlSchemaChange> supportedChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : changes) {
+ if (change.getType() == MySqlSchemaChange.Type.UNSUPPORTED) {
Review Comment:
Mixed ALTERs are not handled atomically here. If a single MySQL DDL contains
a supported change and an unsupported one, for example the new test's `ADD
COLUMN mixed_col INT, MODIFY COLUMN age BIGINT`, this loop drops the
`UNSUPPORTED` item but still emits the ADD. The coordinator then calls
`applySchemaChange(result.getUpdatedSchemas())` and commits `freshSchemas`, so
FE records the source post-DDL schema where `age` is BIGINT even though Doris
still has `age` as INT. Later rows that rely on the modified type can fail or
diverge, and retries start from a baseline that no longer matches the actual
Doris table. Please make each MySQL DDL event all-or-nothing with respect to
unsupported operations, or execute/validate every operation before advancing
the persisted schema baseline.
##########
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_sc_during_snapshot.groovy:
##########
@@ -0,0 +1,147 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_mysql_job_sc_during_snapshot",
"p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
+ def jobName = "test_streaming_mysql_job_sc_during_snapshot"
+ def currentDb = (sql "select database()")[0][0]
+ def table1 = "test_streaming_mysql_sc_snapshot_t"
+ def mysqlDb = "test_cdc_db"
+ def totalRows = 1000
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysqlPort = context.config.otherConfigs.get("mysql_57_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ def waitForColumn = { String column, boolean expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ (sql "DESC ${table1}").any { it[0] == column } == expected
+ })
+ }
+ def waitForValue = { int id, String column, String expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql "SELECT ${column} FROM ${table1} WHERE id=${id}"
+ rows.size() == 1 && String.valueOf(rows[0][0]) == expected
+ })
+ }
+ def waitForSucceedTasks = { int expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def cnt = sql """SELECT SucceedTaskCount FROM
jobs("type"="insert")
+ WHERE Name='${jobName}' AND
ExecuteType='STREAMING'"""
+ log.info("SucceedTaskCount before schema change: " + cnt)
+ cnt.size() == 1 && (cnt[0][0] as int) >= expected
+ })
+ }
+ def dumpJobState = {
+ log.info("jobs : " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks : " + sql("""SELECT * FROM tasks("type"="insert")
WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${table1}"""
+ sql """CREATE TABLE ${mysqlDb}.${table1} (
+ id INT NOT NULL,
+ tag VARCHAR(64),
+ version INT,
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+
+ StringBuilder sb = new StringBuilder()
+ sb.append("INSERT INTO ${mysqlDb}.${table1} (id, tag, version)
VALUES ")
+ for (int i = 1; i <= totalRows; i++) {
+ if (i > 1) {
+ sb.append(", ")
+ }
+ sb.append("(${i}, 'snap', 0)")
+ }
+ sql sb.toString()
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM MYSQL (
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysqlPort}?serverTimezone=UTC",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root",
+ "password" = "123456",
+ "database" = "${mysqlDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial",
+ "snapshot_split_size" = "10",
+ "snapshot_parallelism" = "1"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ // Issue the DDL only after at least one snapshot split has committed.
This keeps
+ // the later binlog split start offset before the DDL, while the
remaining snapshot
+ // splits may still exercise automatic task/reader retry if they see
the new
+ // ResultSet with the cached old source schema.
+ try {
+ waitForSucceedTasks(1)
Review Comment:
This does not make the "during snapshot" timing deterministic. The test
waits for `SucceedTaskCount >= 1`, then issues the DDL, but with a 2-second
poll interval a fast run can finish all snapshot work before the await observes
the first success. The rest of the assertions would still pass in that case:
the 1000 original rows have NULL `sc_note`, and the one row inserted after the
DDL has `during_snapshot`, which is also true for the normal post-snapshot
schema-change path. Please add a barrier/assertion that snapshot work is still
active before the ALTER, or otherwise force/block the snapshot reader so this
test proves the intended retry path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
