This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new dec5a9c68d6 branch-4.2: [improvement](fe) Improve streaming job schema
change error messages #69725 (#70167)
dec5a9c68d6 is described below
commit dec5a9c68d6a5abda0803f1f05844584ea846fab
Author: wudi <[email protected]>
AuthorDate: Wed Nov 4 17:36:03 2026 +0800
branch-4.2: [improvement](fe) Improve streaming job schema change error
messages #69725 (#70167)
Cherry-picked from #69725
---
.../insert/streaming/StreamingInsertJob.java | 4 +-
.../StreamingInsertJobStatusTransitionTest.java | 45 +++++++
.../cdcclient/service/PipelineCoordinator.java | 10 +-
.../source/deserialize/DeserializeResult.java | 17 ++-
.../deserialize/MySqlDebeziumJsonDeserializer.java | 20 ++-
.../PostgresDebeziumJsonDeserializer.java | 10 +-
.../PipelineCoordinatorSchemaChangeTest.java | 135 +++++++++++++++++++++
.../MySqlSchemaChangeDeserializeTest.java | 43 ++++++-
.../PostgresSchemaChangeDeserializeTest.java | 11 +-
.../test_streaming_mysql_job_sc_advanced.groovy | 34 ++++--
.../test_streaming_postgres_job_sc_advanced.groovy | 20 ++-
11 files changed, 315 insertions(+), 34 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index 58296c78ec8..ed6c6bcb5d7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -86,6 +86,7 @@ import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TxnStateChangeCallback;
import com.google.common.base.Preconditions;
+import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.annotations.SerializedName;
import lombok.Getter;
@@ -114,6 +115,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask,
Map<Object, Object>> implements
TxnStateChangeCallback, GsonPostProcessable {
public static final String JOB_FILE_CATALOG = "streaming_job";
+ private static final Gson ERROR_MSG_GSON =
GsonUtils.GSON.newBuilder().disableHtmlEscaping().create();
private long dbId;
// Streaming job statistics, all persisted in txn attachment
// when checkpoint image replay need Serialized
@@ -1273,7 +1275,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
// returning the plain FailMsg.msg string — they use a different error
taxonomy
// (FailMsg/CancelType) that is out of scope for this PR.
trow.addToColumnValue(new TCell().setStringVal(failureReason == null
- ? "" : GsonUtils.GSON.toJson(failureReason)));
+ ? "" : ERROR_MSG_GSON.toJson(failureReason)));
trow.addToColumnValue(new TCell().setStringVal(jobRuntimeMsg == null
? "" : jobRuntimeMsg));
trow.addToColumnValue(new TCell().setStringVal(getLag()));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
index 0cfb837bf91..096bc1ffae6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
@@ -17,17 +17,23 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.base.JobExecutionConfiguration;
import org.apache.doris.job.cdc.request.TaskFailureRequest;
import org.apache.doris.job.common.FailureReason;
import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.InsertJob;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
import org.apache.doris.transaction.TxnStateCallbackFactory;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
@@ -35,6 +41,7 @@ import org.mockito.Mockito;
import java.util.Collections;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
public class StreamingInsertJobStatusTransitionTest {
@@ -47,6 +54,44 @@ public class StreamingInsertJobStatusTransitionTest {
return job;
}
+ @Test
+ public void testErrorMsgKeepsReadableCharactersAndValidJson() {
+ String message = TaskFailureRequest.SCHEMA_CHANGE_UNSUPPORTED
+ + " Table: source.orders -> target.orders; columns
added=[city]."
+ + " Source DDL: [ALTER TABLE orders ADD city VARCHAR(30)
COMMENT 'city']"
+ + "\nDetails: \"quoted\", path=C:\\data";
+ String json = visibleErrorMsg(new FailureReason(message));
+
+ Assertions.assertTrue(json.contains("source.orders -> target.orders"),
json);
+ Assertions.assertTrue(json.contains("columns added=[city]"), json);
+ Assertions.assertTrue(json.contains("COMMENT 'city'"), json);
+ Assertions.assertFalse(json.contains("\\" + "u003e"), json);
+ Assertions.assertFalse(json.contains("\\" + "u003d"), json);
+ Assertions.assertFalse(json.contains("\\" + "u0027"), json);
+ JsonObject parsed = JsonParser.parseString(json).getAsJsonObject();
+ Assertions.assertEquals("CANNOT_RESUME_ERR",
parsed.get("code").getAsString());
+ Assertions.assertEquals(message, parsed.get("msg").getAsString());
+ }
+
+ @Test
+ public void testErrorMsgIsEmptyWithoutFailure() {
+ Assertions.assertEquals("", visibleErrorMsg(null));
+ }
+
+ private static String visibleErrorMsg(FailureReason reason) {
+ StreamingInsertJob job = newJob(JobStatus.PAUSED);
+ job.setCreateUser(UserIdentity.ROOT);
+ JobExecutionConfiguration config = new JobExecutionConfiguration();
+ config.setExecuteType(JobExecuteType.STREAMING);
+ job.setJobConfig(config);
+ job.setExecuteSql("SELECT 1");
+ job.setEncryptedSql("SELECT 1");
+ job.setFailureReason(reason);
+ int index = InsertJob.SCHEMA.stream().map(column -> column.getName())
+ .collect(Collectors.toList()).indexOf("ErrorMsg");
+ return job.getTvfInfo().getColumnValue().get(index).getStringVal();
+ }
+
@Test
public void testManualResumeToleratesSchemaChangesUntilSuccess() throws
Exception {
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
index a8439409c3d..a422012b63c 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
@@ -79,6 +79,9 @@ import
org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBo
public class PipelineCoordinator {
private static final Logger LOG =
LoggerFactory.getLogger(PipelineCoordinator.class);
private static final String SPLIT_ID = "splitId";
+ private static final String SCHEMA_CHANGE_RECOVERY_HINT =
+ "Review and adjust the target schema as needed. "
+ + "RESUME confirms acceptance of unsupported changes for
this recovery round.";
// jobId
private final Map<String, DorisBatchStreamLoad> batchStreamLoadMap = new
ConcurrentHashMap<>();
// taskId -> list of split offsets (accumulates all splits processed in
one task)
@@ -745,8 +748,8 @@ public class PipelineCoordinator {
TaskFailureRequest.SCHEMA_CHANGE_UNSUPPORTED
+ " "
+ result.getUnsupportedReason()
- + ". Review the target schema. RESUME confirms
acceptance of unsupported"
- + " changes for this recovery round.");
+ + ". "
+ + SCHEMA_CHANGE_RECOVERY_HINT);
}
try {
SchemaChangeManager.validateTargetSchemas(
@@ -764,7 +767,8 @@ public class PipelineCoordinator {
+ result.getUnsupportedReason()
+ ". Doris target: "
+ e.getMessage()
- + ". Fix the target schema, then RESUME
the job.");
+ + ". "
+ + SCHEMA_CHANGE_RECOVERY_HINT);
// The root message drives FE's pause classification.
failure.addSuppressed(e);
throw failure;
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/DeserializeResult.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/DeserializeResult.java
index c982e57a863..95aaa1e299a 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/DeserializeResult.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/DeserializeResult.java
@@ -87,7 +87,10 @@ public class DeserializeResult {
public static DeserializeResult unsupportedSchemaChange(
String reason,
Map<TableId, TableChanges.TableChange> previousSchemas,
- Map<TableId, TableChanges.TableChange> updatedSchemas) {
+ Map<TableId, TableChanges.TableChange> updatedSchemas,
+ String targetDb,
+ Map<String, String> targetTableMappings,
+ String sourceDdl) {
StringBuilder message = new StringBuilder(reason);
for (Map.Entry<TableId, TableChanges.TableChange> entry :
updatedSchemas.entrySet()) {
Table before = previousSchemas.get(entry.getKey()).getTable();
@@ -108,7 +111,14 @@ public class DeserializeResult {
.map(Column::name)
.filter(name -> after.columnWithName(name) == null)
.collect(Collectors.toList());
- message.append(". Table: ").append(entry.getKey().identifier());
+ String sourceTable = entry.getKey().table();
+ String targetTable = targetTableMappings.getOrDefault(sourceTable,
sourceTable);
+ message.append(". Table: ")
+ .append(entry.getKey().identifier())
+ .append(" -> ")
+ .append(targetDb)
+ .append(".")
+ .append(targetTable);
if (!added.isEmpty()) {
message.append("; columns added=").append(added);
}
@@ -116,6 +126,9 @@ public class DeserializeResult {
message.append("; removed=").append(removed);
}
}
+ if (sourceDdl != null) {
+ message.append(". Source DDL: [").append(sourceDdl).append("]");
+ }
return new DeserializeResult(
Type.SCHEMA_CHANGE,
Collections.emptyList(),
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java
index a7c63e43e30..6ce826f13a4 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java
@@ -124,6 +124,7 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
return DeserializeResult.schemaChange(Collections.emptyList(),
freshSchemas);
}
+ String db = context.get(Constants.DORIS_TARGET_DB);
for (Map.Entry<TableId, TableChanges.TableChange> entry :
freshSchemas.entrySet()) {
if (entry.getValue().getType() !=
TableChanges.TableChangeType.ALTER) {
continue;
@@ -135,7 +136,7 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
.primaryKeyColumnNames()
.equals(entry.getValue().getTable().primaryKeyColumnNames())) {
return unsupportedSchemaChange(
- record, ddl, "Primary key changes are not supported",
freshSchemas);
+ db, record, ddl, "Primary key changes are not
supported", freshSchemas);
}
}
@@ -148,7 +149,8 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
changes = parser.getAndClearParsedChanges();
} catch (Exception e) {
LOG.warn("Failed to parse MySQL schema change DDL: {}", ddl, e);
- return unsupportedSchemaChange(record, ddl, "Cannot parse schema
change", freshSchemas);
+ return unsupportedSchemaChange(
+ db, record, ddl, "Cannot parse schema change",
freshSchemas);
}
if (changes.isEmpty()) {
for (Map.Entry<TableId, TableChanges.TableChange> entry :
freshSchemas.entrySet()) {
@@ -170,7 +172,7 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
.collect(Collectors.toSet());
if (!before.equals(after)) {
return unsupportedSchemaChange(
- record, ddl, "Unrecognized column change",
freshSchemas);
+ db, record, ddl, "Unrecognized column change",
freshSchemas);
}
}
return DeserializeResult.schemaChange(Collections.emptyList(),
freshSchemas);
@@ -192,10 +194,15 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
if (excluded.contains(columnName) &&
excluded.contains(newColumnName)) {
continue;
}
+ String columnChange =
+ columnName.equals(newColumnName)
+ ? columnName
+ : columnName + " -> " + newColumnName;
return unsupportedSchemaChange(
+ db,
record,
ddl,
- type + " COLUMN is not supported (" + columnName + " -> "
+ newColumnName + ")",
+ type + " COLUMN is not supported (" + columnChange + ")",
freshSchemas);
}
if (supportedChanges.isEmpty()) {
@@ -220,7 +227,6 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
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)) {
@@ -285,6 +291,7 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
}
private DeserializeResult unsupportedSchemaChange(
+ String targetDb,
SourceRecord record,
String ddl,
String reason,
@@ -295,7 +302,8 @@ public class MySqlDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
reason,
ddl,
record.sourceOffset());
- return DeserializeResult.unsupportedSchemaChange(reason, tableSchemas,
freshSchemas);
+ return DeserializeResult.unsupportedSchemaChange(
+ reason, tableSchemas, freshSchemas, targetDb,
targetTableMappingsCache, ddl);
}
private Tables toTables(Map<TableId, TableChanges.TableChange> schemas) {
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java
index f0cc6dbaa2c..5e81cd0f23c 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java
@@ -141,12 +141,13 @@ public class PostgresDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
return DeserializeResult.schemaChange(Collections.emptyList(),
updatedSchemas);
}
+ String db = context.get(Constants.DORIS_TARGET_DB);
Set<String> excludedCols =
excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
String unsupportedReason =
unsupportedChangeReason(stored.getTable(), freshTable,
excludedCols);
if (unsupportedReason != null) {
- return unsupportedSchemaChange(updatedSchemas, unsupportedReason);
+ return unsupportedSchemaChange(db, updatedSchemas,
unsupportedReason);
}
List<Column> added = new ArrayList<>();
@@ -169,11 +170,11 @@ public class PostgresDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
dropped.removeIf(excludedCols::contains);
if (hasAddedAndDropped && (!added.isEmpty() || !dropped.isEmpty())) {
return unsupportedSchemaChange(
+ db,
updatedSchemas,
"Cannot distinguish a column rename from simultaneous
ADD/DROP");
}
- String db = context.get(Constants.DORIS_TARGET_DB);
List<SchemaChangeOperation> schemaChanges = new ArrayList<>();
String targetTable = resolveTargetTable(tableId.table());
@@ -212,8 +213,9 @@ public class PostgresDebeziumJsonDeserializer extends
DebeziumJsonDeserializer {
}
private DeserializeResult unsupportedSchemaChange(
- Map<TableId, TableChanges.TableChange> updatedSchemas, String
reason) {
- return DeserializeResult.unsupportedSchemaChange(reason, tableSchemas,
updatedSchemas);
+ String targetDb, Map<TableId, TableChanges.TableChange>
updatedSchemas, String reason) {
+ return DeserializeResult.unsupportedSchemaChange(
+ reason, tableSchemas, updatedSchemas, targetDb,
targetTableMappingsCache, null);
}
private static String unsupportedChangeReason(
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/service/PipelineCoordinatorSchemaChangeTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/service/PipelineCoordinatorSchemaChangeTest.java
new file mode 100644
index 00000000000..d64e7f62961
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/service/PipelineCoordinatorSchemaChangeTest.java
@@ -0,0 +1,135 @@
+// 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.
+
+package org.apache.doris.cdcclient.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.doris.cdcclient.source.deserialize.DeserializeResult;
+import org.apache.doris.job.cdc.request.WriteRecordRequest;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.sql.Types;
+import java.util.Map;
+
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+
+class PipelineCoordinatorSchemaChangeTest {
+ private static final String RECOVERY_HINT =
+ "Review and adjust the target schema as needed. "
+ + "RESUME confirms acceptance of unsupported changes for
this recovery round.";
+
+ @Test
+ void initialFailureKeepsReasonAndExplainsResumeConfirmation() throws
Exception {
+ DeserializeResult result = unsupportedChange();
+ IOException failure = validateFailure(new WriteRecordRequest(),
result);
+
+ assertThat(ExceptionUtils.getRootCauseMessage(failure))
+ .contains("[SCHEMA_CHANGE_UNSUPPORTED]",
result.getUnsupportedReason())
+ .endsWith(RECOVERY_HINT);
+ assertThat(failure.getSuppressed()).isEmpty();
+ }
+
+ @Test
+ void resumedFailureKeepsMissingColumnsAndConfirmation() throws Exception {
+ assertResumeFailure(
+
"{\"code\":0,\"data\":{\"status\":200,\"properties\":[{\"name\":\"id\"}]}}",
+ "target_db.mapped_t1: missing columns [city]");
+ }
+
+ @Test
+ void resumedFailureKeepsSchemaQueryErrorAndConfirmation() throws Exception
{
+ assertResumeFailure(
+ "{\"code\":1,\"msg\":\"schema unavailable\"}", "schema
unavailable");
+ }
+
+ private static void assertResumeFailure(String responseBody, String
detail) throws Exception {
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext(
+ "/api/streaming/schema/target_db/mapped_t1",
+ exchange -> {
+ byte[] bytes =
responseBody.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, bytes.length);
+ exchange.getResponseBody().write(bytes);
+ exchange.close();
+ });
+ server.start();
+ try {
+ WriteRecordRequest request = new WriteRecordRequest();
+ request.setFrontendAddress("127.0.0.1:" +
server.getAddress().getPort());
+ request.setTargetDb("target_db");
+ request.setToken("token");
+ request.setJobId("123");
+ request.setConfig(Map.of("table.t1.target_table", "mapped_t1"));
+ request.setTolerateSchemaChange(true);
+ DeserializeResult result = unsupportedChange();
+
+ IOException failure = validateFailure(request, result);
+
+ // FE classifies the root message, so a nested cause must not hide
the marker.
+ assertThat(ExceptionUtils.getRootCauseMessage(failure))
+ .contains("[SCHEMA_CHANGE_UNSUPPORTED]",
result.getUnsupportedReason(),
+ "Doris target:", detail)
+ .endsWith(RECOVERY_HINT);
+ assertThat(failure.getSuppressed()).hasSize(1);
+
assertThat(failure.getSuppressed()[0]).hasMessageContaining(detail);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private static IOException validateFailure(WriteRecordRequest request,
DeserializeResult result)
+ throws Exception {
+ Method validate = PipelineCoordinator.class.getDeclaredMethod(
+ "validateUnsupportedSchemaChange", WriteRecordRequest.class,
DeserializeResult.class);
+ validate.setAccessible(true);
+ InvocationTargetException failure = assertThrows(
+ InvocationTargetException.class,
+ () -> validate.invoke(new PipelineCoordinator(1), request,
result));
+ assertThat(failure.getCause()).isInstanceOf(IOException.class);
+ return (IOException) failure.getCause();
+ }
+
+ private static DeserializeResult unsupportedChange() {
+ TableId tableId = new TableId("source_db", null, "t1");
+ Table before = Table.editor().tableId(tableId)
+
.addColumn(Column.editor().name("id").type("INT").jdbcType(Types.INTEGER).create())
+ .create();
+ Table after = before.edit()
+
.addColumn(Column.editor().name("city").type("INT").jdbcType(Types.INTEGER).create())
+ .create();
+ return DeserializeResult.unsupportedSchemaChange(
+ "MODIFY COLUMN is not supported (id)",
+ Map.of(tableId, new
TableChanges.TableChange(TableChanges.TableChangeType.ALTER, before)),
+ Map.of(tableId, new
TableChanges.TableChange(TableChanges.TableChangeType.ALTER, after)),
+ "target_db", Map.of("t1", "mapped_t1"),
+ "ALTER TABLE t1 ADD COLUMN city INT, MODIFY COLUMN id BIGINT");
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/MySqlSchemaChangeDeserializeTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/MySqlSchemaChangeDeserializeTest.java
index e9655e1b8da..bde64b7246e 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/MySqlSchemaChangeDeserializeTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/MySqlSchemaChangeDeserializeTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.cdcclient.source.deserialize;
import static
org.apache.flink.cdc.connectors.mysql.debezium.dispatcher.EventDispatcherImpl.HISTORY_RECORD_FIELD;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -60,12 +61,14 @@ class MySqlSchemaChangeDeserializeTest {
DeserializeResult result =
deserializer.deserialize(
- Collections.emptyMap(),
+ CONTEXT,
schemaChangeRecord(null, table("id", "city")));
assertEquals(DeserializeResult.Type.SCHEMA_CHANGE, result.getType());
assertNotNull(result.getUnsupportedReason());
assertTrue(result.getUnsupportedReason().contains("Cannot parse schema
change"));
+ assertTrue(result.getUnsupportedReason().contains("Table: db1.t1 ->
doris_db.t1"));
+ assertFalse(result.getUnsupportedReason().contains("Source DDL:"));
assertTrue(result.getSchemaChanges().isEmpty());
assertEquals(1, result.getUpdatedSchemas().size());
assertEquals(table("id", "city"),
result.getUpdatedSchemas().get(TABLE).getTable());
@@ -190,7 +193,8 @@ class MySqlSchemaChangeDeserializeTest {
assertEquals(DeserializeResult.Type.SCHEMA_CHANGE,
result.getType());
if ("evolve".equals(behavior)) {
assertNotNull(result.getUnsupportedReason());
- assertTrue(result.getUnsupportedReason().contains("MODIFY
COLUMN"));
+ assertTrue(result.getUnsupportedReason().contains("MODIFY
COLUMN is not supported (city)"));
+ assertFalse(result.getUnsupportedReason().contains("city ->
city"));
} else {
assertNull(result.getUnsupportedReason());
}
@@ -207,23 +211,52 @@ class MySqlSchemaChangeDeserializeTest {
@Test
void schemaChangeMixedAddAndSameNameChangeRequiresConfirmation() throws
Exception {
- Table baseline = table("id", "age");
- Table fresh = table("id", "age", "city");
+ Table baseline = table("id", "age", "old_col");
+ Table fresh = table("id", "age", "city", "note");
for (String alter : new String[] {"MODIFY COLUMN age INT", "CHANGE
COLUMN age age INT"}) {
MySqlDebeziumJsonDeserializer deserializer =
newDeserializer(baseline);
+ deserializer.targetTableMappingsCache = Map.of("t1", "mapped_t1");
+ String ddl = "ALTER TABLE t1 ADD COLUMN city INT, " + alter
+ + ", DROP COLUMN old_col, ADD COLUMN note INT;";
DeserializeResult result =
deserializer.deserialize(
CONTEXT,
- schemaChangeRecord("ALTER TABLE t1 ADD COLUMN city
INT, " + alter, fresh));
+ schemaChangeRecord(ddl, fresh));
assertEquals(DeserializeResult.Type.SCHEMA_CHANGE,
result.getType());
assertNotNull(result.getUnsupportedReason());
+ String reason = result.getUnsupportedReason();
+ assertTrue(reason.contains("COLUMN is not supported (age)"));
+ assertFalse(reason.contains("age -> age"));
+ assertTrue(reason.contains("Table: db1.t1 -> doris_db.mapped_t1"));
+ assertTrue(reason.contains("columns added=[city, note]"));
+ assertTrue(reason.contains("removed=[old_col]"));
+ assertTrue(reason.contains("Source DDL: [" + ddl + "]"));
assertTrue(result.getSchemaChanges().isEmpty(), "mixed DDL must
not be partially applied");
assertEquals(fresh,
result.getUpdatedSchemas().get(TABLE).getTable());
assertEquals(baseline,
deserializer.getTableSchemas().get(TABLE).getTable());
}
}
+ @Test
+ void renamedColumnRetainsBothNamesAndSourceDdl() throws Exception {
+ Table baseline = table("id", "age");
+ Table fresh = table("id", "age2");
+ for (String alter : new String[] {"CHANGE COLUMN age age2 INT",
"RENAME COLUMN age TO age2"}) {
+ MySqlDebeziumJsonDeserializer deserializer =
newDeserializer(baseline);
+ String ddl = "ALTER TABLE t1 " + alter;
+
+ DeserializeResult result = deserializer.deserialize(CONTEXT,
schemaChangeRecord(ddl, fresh));
+
+ assertNotNull(result.getUnsupportedReason());
+ assertTrue(result.getUnsupportedReason().contains("COLUMN is not
supported (age -> age2)"));
+ assertTrue(result.getUnsupportedReason().contains("Table: db1.t1
-> doris_db.t1"));
+ assertTrue(result.getUnsupportedReason().contains("Source DDL: ["
+ ddl + "]"));
+ assertTrue(result.getSchemaChanges().isEmpty());
+ assertEquals(baseline,
deserializer.getTableSchemas().get(TABLE).getTable());
+ }
+ }
+
@Test
void ignoreReturnsUpdatedSchemasWithoutDdlOrConfirmation() throws
Exception {
Table baseline = table("id", "name");
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/PostgresSchemaChangeDeserializeTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/PostgresSchemaChangeDeserializeTest.java
index 12efc56c6d0..a3844777150 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/PostgresSchemaChangeDeserializeTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/deserialize/PostgresSchemaChangeDeserializeTest.java
@@ -114,12 +114,15 @@ class PostgresSchemaChangeDeserializeTest {
newDeserializer(
tableWith(column("id", "int4", true,
null).edit().nativeType(23).create()));
Table fresh = tableWith(column("id", "int8", true,
null).edit().nativeType(20).create());
+ deserializer.targetTableMappingsCache = Map.of("t1", "mapped_t1");
DeserializeResult result = deserializer.deserialize(CONTEXT,
schemaRecord(fresh));
assertEquals(DeserializeResult.Type.SCHEMA_CHANGE, result.getType());
assertNotNull(result.getUnsupportedReason());
- assertTrue(result.getUnsupportedReason().contains("int4 to int8"));
+ assertTrue(result.getUnsupportedReason().contains("int4 to int8 (OID
23 -> 20)"));
+ assertTrue(result.getUnsupportedReason().contains("Table: public.t1 ->
doris_db.mapped_t1"));
+ assertFalse(result.getUnsupportedReason().contains("Source DDL:"));
assertTrue(result.getDdls().isEmpty(), "type change must not emit
DDL");
assertEquals(fresh, result.getUpdatedSchemas().get(TABLE).getTable());
}
@@ -276,7 +279,7 @@ class PostgresSchemaChangeDeserializeTest {
@Test
void
relationSimultaneousAddAndDropRequiresConfirmationWithoutApplyingBaseline()
throws Exception {
- // stored [id,name]; Relation [id,nick] -> name dropped + nick added
-> treated as RENAME.
+ // Relation reports the final columns, not the original DDL or a
confirmed rename.
Table baseline = storedTable("id", "name");
Table fresh = storedTable("id", "nick");
PostgresDebeziumJsonDeserializer deserializer =
newDeserializer(baseline);
@@ -287,6 +290,10 @@ class PostgresSchemaChangeDeserializeTest {
assertEquals(DeserializeResult.Type.SCHEMA_CHANGE, result.getType());
assertNotNull(result.getUnsupportedReason());
assertTrue(result.getUnsupportedReason().contains("rename"));
+ assertTrue(result.getUnsupportedReason().contains("Table: public.t1 ->
doris_db.t1"));
+ assertTrue(result.getUnsupportedReason().contains("columns
added=[nick]; removed=[name]"));
+ assertFalse(result.getUnsupportedReason().contains("Source DDL:"));
+ assertFalse(result.getUnsupportedReason().contains("name -> nick"));
assertTrue(result.getDdls().isEmpty(), "rename must not emit DDL");
assertEquals(fresh, result.getUpdatedSchemas().get(TABLE).getTable());
assertEquals(baseline,
deserializer.getTableSchemas().get(TABLE).getTable());
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_sc_advanced.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_sc_advanced.groovy
index eabd23de1ea..748ea77fae1 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_sc_advanced.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_sc_advanced.groovy
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+import groovy.json.JsonSlurper
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
@@ -62,10 +63,24 @@ suite("test_streaming_mysql_job_sc_advanced",
"p0,external,mysql,external_docker
def waitForCommit = { long previous ->
Awaitility.await().atMost(180, SECONDS).pollInterval(1,
SECONDS).until({ succeedCount() > previous })
}
- def waitForPause = { String reason ->
+ def waitForPause = { String reason, List<String> details ->
Awaitility.await().atMost(180, SECONDS).pollInterval(1,
SECONDS).until({
def rows = sql """SELECT Status, ErrorMsg FROM
jobs("type"="insert") WHERE Name='${jobName}'"""
- rows.size() == 1 && rows[0][0] == "PAUSED" &&
rows[0][1].toString().contains(reason)
+ if (rows.size() != 1 || rows[0][0] != "PAUSED") {
+ return false
+ }
+ def rawError = rows[0][1].toString()
+ def error = new JsonSlurper().parseText(rawError)
+ return (error.code == "CANNOT_RESUME_ERR"
+ && rawError.contains(" -> ")
+ && error.msg.contains("[SCHEMA_CHANGE_UNSUPPORTED]")
+ && error.msg.contains(reason)
+ && error.msg.contains("Table: ${mysqlDb}.${table1} ->
${currentDb}.${table1}")
+ && error.msg.contains("Source DDL: [ALTER TABLE")
+ && error.msg.contains("Review and adjust the target
schema as needed.")
+ && error.msg.contains("RESUME confirms acceptance of
unsupported changes for this recovery round.")
+ && !error.msg.contains("age -> age)")
+ && details.every { error.msg.contains(it) })
})
}
@@ -161,13 +176,15 @@ suite("test_streaming_mysql_job_sc_advanced",
"p0,external,mysql,external_docker
(first_col, name, after_name, age, city, mixed_col) VALUES
('X', 'E1', 9, 5, 'sh', 100)"""
}
- waitForPause("SCHEMA_CHANGE_UNSUPPORTED")
+ def mixedDdlDetails = ["MODIFY COLUMN is not supported (age)",
+ "ADD COLUMN mixed_col INT", "MODIFY COLUMN age BIGINT"]
+ waitForPause("SCHEMA_CHANGE_UNSUPPORTED", mixedDdlDetails)
qt_mixed_paused """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
qt_mixed_not_applied """SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema='${currentDb}' AND table_name='${table1}' AND
column_name='mixed_col'"""
qt_mixed_row_not_written "SELECT COUNT(*) FROM ${table1} WHERE
name='E1'"
sql "RESUME JOB WHERE jobname='${jobName}'"
- waitForPause("missing columns")
+ waitForPause("Doris target:", mixedDdlDetails + ["missing columns
[mixed_col]"])
qt_missing_column_pause """SELECT Status FROM jobs("type"="insert")
WHERE Name='${jobName}'"""
sql "ALTER TABLE ${table1} ADD COLUMN mixed_col INT NULL"
long beforeMixedResume = succeedCount()
@@ -191,7 +208,8 @@ suite("test_streaming_mysql_job_sc_advanced",
"p0,external,mysql,external_docker
(first_col, name, after_name, age, city, mixed_col) VALUES
('M', 'F1', 10, 6, 'sz', 200)"""
}
- waitForPause("SCHEMA_CHANGE_UNSUPPORTED")
+ waitForPause("SCHEMA_CHANGE_UNSUPPORTED",
+ ["MODIFY COLUMN is not supported (age)", "MODIFY COLUMN age
SMALLINT"])
qt_modify_paused """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
long beforeModifyResume = succeedCount()
sql "RESUME JOB WHERE jobname='${jobName}'"
@@ -213,7 +231,8 @@ suite("test_streaming_mysql_job_sc_advanced",
"p0,external,mysql,external_docker
(first_col, name, after_name, age2, city, mixed_col) VALUES
('R', 'G1', 11, 7, 'bj', 300)"""
}
- waitForPause("SCHEMA_CHANGE_UNSUPPORTED")
+ waitForPause("SCHEMA_CHANGE_UNSUPPORTED",
+ ["CHANGE COLUMN is not supported (age -> age2)", "CHANGE
COLUMN age age2 INT"])
qt_rename_paused """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
// Keep age available for DML replayed from the last committed offset.
sql "ALTER TABLE ${table1} ADD COLUMN age2 INT NULL"
@@ -236,7 +255,8 @@ suite("test_streaming_mysql_job_sc_advanced",
"p0,external,mysql,external_docker
sql """INSERT INTO ${mysqlDb}.${table1}
(first_col, name, after_name, age2, city, mixed_col) VALUES
('S', 'H1', 12, 8, 'cd', 400)"""
}
- waitForPause("SCHEMA_CHANGE_UNSUPPORTED")
+ waitForPause("SCHEMA_CHANGE_UNSUPPORTED",
+ ["MODIFY COLUMN is not supported (age2)", "COMMENT 'comment
only'"])
qt_column_comment_paused """SELECT Status FROM jobs("type"="insert")
WHERE Name='${jobName}'"""
long beforeCommentResume = succeedCount()
sql "RESUME JOB WHERE jobname='${jobName}'"
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_sc_advanced.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_sc_advanced.groovy
index 33aac91d5cb..fd30dbb5d85 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_sc_advanced.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_sc_advanced.groovy
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+import groovy.json.JsonSlurper
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
@@ -109,10 +110,21 @@ suite("test_streaming_postgres_job_sc_advanced",
def waitForSchemaPause = { boolean missingColumn ->
Awaitility.await().atMost(180, SECONDS).pollInterval(1,
SECONDS).until({
def rows = sql """select Status, ErrorMsg from
jobs("type"="insert") where Name='${jobName}'"""
- (rows.size() == 1 && rows[0][0] == 'PAUSED'
- &&
rows[0][1].toString().contains('[SCHEMA_CHANGE_UNSUPPORTED]')
- && (!missingColumn ||
(rows[0][1].toString().contains('missing columns')
- && rows[0][1].toString().contains('c3'))))
+ if (rows.size() != 1 || rows[0][0] != 'PAUSED') {
+ return false
+ }
+ def rawError = rows[0][1].toString()
+ def error = new JsonSlurper().parseText(rawError)
+ return (error.code == 'CANNOT_RESUME_ERR'
+ && rawError.contains(" -> ")
+ && error.msg.contains('[SCHEMA_CHANGE_UNSUPPORTED]')
+ && error.msg.contains("Table: ${pgSchema}.${table1} ->
${currentDb}.${table1}")
+ && error.msg.contains('columns added=[c3];
removed=[c1]')
+ && !error.msg.contains('Source DDL:')
+ && error.msg.contains('Review and adjust the target
schema as needed.')
+ && error.msg.contains('RESUME confirms acceptance of
unsupported changes for this recovery round.')
+ && (!missingColumn || (error.msg.contains('Doris
target:')
+ && error.msg.contains('missing columns
[c3]'))))
})
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
