From 5ab01bc78d0963dadac13cb598300c6a5f07ca2a Mon Sep 17 00:00:00 2001 From: tanishq171 Date: Thu, 13 Aug 2026 21:26:01 +0530 Subject: [PATCH] source-mysql v10: preserve saved schema history so column additions don't break CDC The v7 schema-history rebuild unconditionally replaced the saved history with a fresh snapshot from live SHOW CREATE TABLE. That breaks in-place schema evolution: when a column is added on the source, the rebuild seeds the connector with the current (post-ALTER) N-column shape, but the connector resumes from a saved binlog offset that predates the ALTER. Debezium then validates old, pre-ALTER row events (N-1 columns) against the rebuilt N-column schema and aborts with 'internal schema size N, but row size N-1, restart connector with schema recovery mode' instead of auto-propagating the new column. v10 keeps the saved schema history whenever it is non-empty, letting Debezium replay the ALTER DDL from the binlog at its correct position and evolve the schema in lockstep (stock behavior). The SHOW CREATE TABLE rebuild now runs only when there is no saved history to resume from. Trade-off: a non-empty-but-incomplete history (a table missing after a swap + retention gap) is no longer force-rebuilt. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../connectors/source-mysql/metadata.yaml | 2 +- .../mysql/MySqlSourceDebeziumOperations.kt | 45 +++++++++++++------ 2 files changed, 33 insertions(+), 14 deletions(-) diff --git a/airbyte-integrations/connectors/source-mysql/metadata.yaml b/airbyte-integrations/connectors/source-mysql/metadata.yaml index b78f4afd4356..6da06df6a722 100644 --- a/airbyte-integrations/connectors/source-mysql/metadata.yaml +++ b/airbyte-integrations/connectors/source-mysql/metadata.yaml @@ -9,7 +9,7 @@ data: connectorSubtype: database connectorType: source definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad - dockerImageTag: 3.52.3-gtid-patch-v9 + dockerImageTag: 3.52.3-gtid-patch-v10 dockerRepository: tanishq171/source-mysql documentationUrl: https://docs.airbyte.com/integrations/sources/mysql githubIssueLabel: source-mysql diff --git a/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt b/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt index 85383b146415..85e74d6cb70c 100644 --- a/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt +++ b/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt @@ -279,20 +279,39 @@ class MySqlSourceDebeziumOperations( // // Safety: if information_schema or any SHOW CREATE TABLE call fails, fall back to // existing history rather than partial-replace. Worst-case v7 is no worse than v6. - val rebuiltSchemaHistory: DebeziumSchemaHistory? = - try { - rebuildSchemaHistoryFromSource( - existing = debeziumState.schemaHistory, - referencePosition = savedStateOffset.position, - ) - } catch (e: Exception) { - log.warn(e) { - "Schema history rebuild failed; proceeding with existing history. " + - "${e.message}" + // v10: only rebuild when there is no saved schema history to start from. + // + // The v7 full-replace above breaks in-place schema evolution. When a column is added on + // the source, the SHOW CREATE TABLE rebuild seeds the connector with the *current* + // (post-ALTER) table shape, but the connector resumes streaming from a saved binlog + // offset that predates the ALTER. Debezium then validates old, pre-ALTER row events + // (N-1 columns) against the rebuilt N-column schema and aborts with + // "internal schema size N, but row size N-1, restart connector with schema recovery + // mode". Stock Debezium instead replays the ALTER DDL from the binlog at its correct + // position and evolves the schema in lockstep — which is how a new column is normally + // auto-propagated. + // + // So: keep the saved schema history whenever it is non-empty (let binlog DDL replay + // drive schema evolution), and fall back to the SHOW CREATE TABLE rebuild only when + // there is no saved history to resume from — the empty-history case the v7 rebuild was + // originally meant to cover. (A non-empty-but-incomplete history — e.g. a table missing + // after a swap + retention gap — is no longer force-rebuilt; that trade is accepted here + // in favour of not breaking column additions.) + val schemaHistory: DebeziumSchemaHistory? = + debeziumState.schemaHistory?.takeIf { it.wrapped.isNotEmpty() } + ?: try { + rebuildSchemaHistoryFromSource( + existing = debeziumState.schemaHistory, + referencePosition = savedStateOffset.position, + ) + } catch (e: Exception) { + log.warn(e) { + "Schema history rebuild failed; proceeding with existing history. " + + "${e.message}" + } + debeziumState.schemaHistory } - debeziumState.schemaHistory - } - return ValidDebeziumWarmStartState(debeziumState.offset, rebuiltSchemaHistory) + return ValidDebeziumWarmStartState(debeziumState.offset, schemaHistory) } /**