github-actions[bot] commented on code in PR #64850:
URL: https://github.com/apache/doris/pull/64850#discussion_r3527074271
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/AbstractCdcSourceReader.java:
##########
@@ -184,44 +189,27 @@ public void loadTableSchemasFromJson(String json) throws
IOException {
* [{"i":"\"schema\".\"table\"","uc":false,"c":{...debeziumDoc...}},...]}.
*/
@Override
Review Comment:
This now fails the FE `commitOffset` when `tableSchemas` cannot be
serialized, but it happens after `writeRecords()` has already left the `try`
block and run `cleanupReaderResources()`. For PostgreSQL that cleanup calls
`commitSourceOffset()` and acknowledges the stream split's current offset to
the replication slot before this method is called. In an ADD Relation + DML
batch, the iterator has already advanced the split state to the DML offset, so
a serialization exception here leaves FE at the old offset/tableSchemas while
the upstream slot may have confirmed WAL past the Relation/DML that FE still
needs to retry. The new test only checks that the mock FE commit body is
unchanged; it does not catch that the source offset was already committed.
Please move schema serialization/durable FE commit ahead of the PostgreSQL slot
acknowledgement, or otherwise skip `commitSourceOffset()` when the batch's
schema baseline cannot be persisted.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeManager.java:
##########
@@ -125,25 +175,33 @@ private static String handleResponse(HttpPost request)
throws IOException {
* <li>DROP COLUMN — "Column does not exists": column was already
dropped.
* </ul>
*/
- private static void parseResponse(String sql, String responseBody) throws
IOException {
+ private static void parseResponse(SchemaChangeOperation operation, String
responseBody)
+ throws IOException {
JsonNode root = OBJECT_MAPPER.readTree(responseBody);
JsonNode code = root.get("code");
if (code != null && code.asInt() == 0) {
return;
}
String msg = root.path("msg").asText("");
+ String data = root.path("data").asText("");
Review Comment:
This `ADD COLUMN already exists` branch is only safe for a retry of the
exact same DDL. It also fires when the target table already has a same-name
column with a different definition. For example, if Doris already has `new_col
INT` and the PostgreSQL Relation adds `new_col TEXT`, FE returns this same
error, we return success here, `PipelineCoordinator` applies/persists the
PostgreSQL baseline, and later rows are written as if Doris had the generated
TEXT column. The fallback `_schema` check has the same problem because it only
compares the column name, even though `_schema` exposes type/nullability/key
metadata. Please verify that the existing column is compatible with the
generated operation before treating this as idempotent; otherwise fail the
batch so the baseline is not advanced over a mismatched target schema.
--
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]
