This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ba55ef965e [Feature][Connector-V2][CDC] Support comment-related schema
change events (#11025)
ba55ef965e is described below
commit ba55ef965ee3c8c489f5a95d4a739f553c14c439
Author: cloverdew <[email protected]>
AuthorDate: Sun Aug 16 10:31:22 2026 +0800
[Feature][Connector-V2][CDC] Support comment-related schema change events
(#11025)
---
.../org/apache/seatunnel/api/event/EventType.java | 2 +
.../apache/seatunnel/api/table/catalog/Column.java | 10 +
.../api/table/catalog/MetadataColumn.java | 5 +
.../api/table/catalog/PhysicalColumn.java | 19 ++
.../api/table/schema/SchemaChangeType.java | 6 +-
.../schema/event/AlterColumnCommentEvent.java | 71 +++++++
.../table/schema/event/AlterTableCommentEvent.java | 65 ++++++
.../handler/AlterTableSchemaEventHandler.java | 27 +++
.../handler/TableSchemaChangeEventDispatcher.java | 4 +
.../api/table/schema/event/EventTest.java | 27 +++
.../base/schema/AbstractSchemaChangeResolver.java | 225 ++++++++++++++++-----
.../row/SeaTunnelRowDebeziumDeserializeSchema.java | 13 +-
.../schema/AbstractSchemaChangeResolverTest.java | 158 +++++++++++++--
.../cdc/mysql/source/MySqlIncrementalSource.java | 4 +-
.../mysql/source/MySqlSchemaChangeResolver.java | 8 +-
.../parser/CustomAlterTableParserListener.java | 26 ++-
.../source/parser/CustomMySqlAntlrDdlParser.java | 19 +-
.../parser/CustomMySqlAntlrDdlParserListener.java | 4 +-
.../seatunnel/cdc/mysql/utils/MySqlTypeUtils.java | 3 +-
.../parser/CustomMySqlAntlrDdlParserTest.java | 20 +-
.../oracle/source/OracleSchemaChangeResolver.java | 7 +-
.../parser/CustomAlterTableParserListener.java | 6 +-
.../source/parser/CustomOracleAntlrDdlParser.java | 8 +-
.../parser/CustomOracleAntlrDdlParserListener.java | 4 +-
.../oracle/source/parser/OracleDdlParserTest.java | 38 ++--
.../databend/schema/SchemaChangeManager.java | 9 +
.../doris/schema/SchemaChangeManager.java | 9 +
.../sink/ElasticsearchSinkWriter.java | 14 +-
.../sink/ElasticsearchSinkWriterTest.java | 57 ++++++
.../jdbc/internal/dialect/JdbcDialect.java | 9 +
.../AlterPaimonTableSchemaEventHandler.java | 12 ++
.../seatunnel/starrocks/util/SchemaUtils.java | 9 +
.../cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java | 81 ++++++++
.../cdc/mysql/MysqlCDCWithSchemaChangeIT.java | 55 +++++
.../src/test/resources/ddl/comment_changes.sql | 63 ++++++
.../translation/flink/schema/SchemaOperator.java | 7 +-
.../flink/schema/SchemaOperatorTest.java | 47 ++++-
37 files changed, 1031 insertions(+), 120 deletions(-)
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/event/EventType.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/event/EventType.java
index d02e7809b8..999d321c8d 100644
--- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/event/EventType.java
+++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/event/EventType.java
@@ -25,6 +25,8 @@ public enum EventType {
SCHEMA_CHANGE_CHANGE_COLUMN,
SCHEMA_CHANGE_UPDATE_COLUMNS,
SCHEMA_CHANGE_RENAME_TABLE,
+ SCHEMA_CHANGE_ALTER_TABLE_COMMENT,
+ SCHEMA_CHANGE_ALTER_COLUMN_COMMENT,
LIFECYCLE_ENUMERATOR_OPEN,
LIFECYCLE_ENUMERATOR_CLOSE,
LIFECYCLE_READER_OPEN,
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java
index 844469164a..b644a18083 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java
@@ -229,4 +229,14 @@ public abstract class Column implements Serializable {
/** Returns a copy of the column with a replaced sourceType. */
public abstract Column reSourceType(String sourceType);
+
+ /**
+ * Returns a copy of the column with a replaced comment.
+ *
+ * <p>Subclasses should override this method when they support comment
mutation.
+ */
+ public Column copyWithComment(String comment) {
+ throw new UnsupportedOperationException(
+ String.format("%s does not support replacing comments",
getClass().getName()));
+ }
}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/MetadataColumn.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/MetadataColumn.java
index eab4b36578..3efff60ab0 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/MetadataColumn.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/MetadataColumn.java
@@ -79,4 +79,9 @@ public class MetadataColumn extends Column {
public Column reSourceType(String sourceType) {
throw new UnsupportedOperationException("Not implemented");
}
+
+ @Override
+ public Column copyWithComment(String newComment) {
+ return MetadataColumn.of(name, dataType, columnLength, nullable,
defaultValue, newComment);
+ }
}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/PhysicalColumn.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/PhysicalColumn.java
index c779cafc4d..01c10f800a 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/PhysicalColumn.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/PhysicalColumn.java
@@ -410,4 +410,23 @@ public class PhysicalColumn extends Column {
bitLen,
longColumnLength);
}
+
+ @Override
+ public Column copyWithComment(String newComment) {
+ return new PhysicalColumn(
+ name,
+ dataType,
+ columnLength,
+ scale,
+ nullable,
+ defaultValue,
+ newComment,
+ sourceType,
+ sinkType,
+ options,
+ isUnsigned,
+ isZeroFill,
+ bitLen,
+ longColumnLength);
+ }
}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/SchemaChangeType.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/SchemaChangeType.java
index e2a08c4e3a..3748d3f53d 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/SchemaChangeType.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/SchemaChangeType.java
@@ -25,5 +25,9 @@ public enum SchemaChangeType {
/** Update column in table. */
UPDATE_COLUMN,
/** Rename column in table. */
- RENAME_COLUMN;
+ RENAME_COLUMN,
+ /** Alter table comment. */
+ ALTER_TABLE_COMMENT,
+ /** Alter column comment. */
+ ALTER_COLUMN_COMMENT;
}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterColumnCommentEvent.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterColumnCommentEvent.java
new file mode 100644
index 0000000000..0466f112c4
--- /dev/null
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterColumnCommentEvent.java
@@ -0,0 +1,71 @@
+/*
+ * 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.seatunnel.api.table.schema.event;
+
+import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ * Represents an event where a column comment is altered.
+ *
+ * <p>This event is triggered by DDL statements like: {@code ALTER TABLE t
MODIFY COLUMN c INT
+ * COMMENT 'new comment'} when only the comment is changed (no type or other
attribute changes).
+ */
+@Getter
+@ToString(callSuper = true)
+public class AlterColumnCommentEvent extends AlterTableColumnEvent {
+
+ /** The name of the column whose comment is being altered. */
+ private final String column;
+
+ /** The old comment of the column. May be null if the column had no
comment before. */
+ private final String oldComment;
+
+ /** The new comment of the column. May be null if the comment is being
removed. */
+ private final String newComment;
+
+ public AlterColumnCommentEvent(
+ TableIdentifier tableIdentifier, String column, String oldComment,
String newComment) {
+ super(tableIdentifier);
+ this.column = column;
+ this.oldComment = oldComment;
+ this.newComment = newComment;
+ }
+
+ /**
+ * Creates an AlterColumnCommentEvent.
+ *
+ * @param tableIdentifier the identifier of the table
+ * @param column the name of the column
+ * @param oldComment the old comment (may be null)
+ * @param newComment the new comment (may be null)
+ * @return the event instance
+ */
+ public static AlterColumnCommentEvent of(
+ TableIdentifier tableIdentifier, String column, String oldComment,
String newComment) {
+ return new AlterColumnCommentEvent(tableIdentifier, column,
oldComment, newComment);
+ }
+
+ @Override
+ public EventType getEventType() {
+ return EventType.SCHEMA_CHANGE_ALTER_COLUMN_COMMENT;
+ }
+}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterTableCommentEvent.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterTableCommentEvent.java
new file mode 100644
index 0000000000..8af67f136c
--- /dev/null
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/event/AlterTableCommentEvent.java
@@ -0,0 +1,65 @@
+/*
+ * 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.seatunnel.api.table.schema.event;
+
+import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ * Represents an event where the table comment is altered.
+ *
+ * <p>This event is triggered by DDL statements like: {@code ALTER TABLE t
COMMENT = 'new comment'}
+ */
+@Getter
+@ToString(callSuper = true)
+public class AlterTableCommentEvent extends AlterTableEvent {
+
+ /** The old comment of the table. May be null if the table had no comment
before. */
+ private final String oldComment;
+
+ /** The new comment of the table. May be null if the comment is being
removed. */
+ private final String newComment;
+
+ public AlterTableCommentEvent(
+ TableIdentifier tableIdentifier, String oldComment, String
newComment) {
+ super(tableIdentifier);
+ this.oldComment = oldComment;
+ this.newComment = newComment;
+ }
+
+ /**
+ * Creates an AlterTableCommentEvent.
+ *
+ * @param tableIdentifier the identifier of the table
+ * @param oldComment the old comment (may be null)
+ * @param newComment the new comment (may be null)
+ * @return the event instance
+ */
+ public static AlterTableCommentEvent of(
+ TableIdentifier tableIdentifier, String oldComment, String
newComment) {
+ return new AlterTableCommentEvent(tableIdentifier, oldComment,
newComment);
+ }
+
+ @Override
+ public EventType getEventType() {
+ return EventType.SCHEMA_CHANGE_ALTER_TABLE_COMMENT;
+ }
+}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/AlterTableSchemaEventHandler.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/AlterTableSchemaEventHandler.java
index fd49df88ca..197948511c 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/AlterTableSchemaEventHandler.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/AlterTableSchemaEventHandler.java
@@ -21,10 +21,12 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
@@ -81,6 +83,13 @@ public class AlterTableSchemaEventHandler implements
TableSchemaChangeEventHandl
}
return newSchema;
}
+ if (alterTableEvent instanceof AlterTableCommentEvent) {
+ // Table comment changes do not affect the physical schema
structure
+ return schema;
+ }
+ if (alterTableEvent instanceof AlterColumnCommentEvent) {
+ return applyColumnComment(schema, (AlterColumnCommentEvent)
alterTableEvent);
+ }
throw new UnsupportedOperationException(
"Unsupported alter table event: " + alterTableEvent);
@@ -205,4 +214,22 @@ public class AlterTableSchemaEventHandler implements
TableSchemaChangeEventHandl
.constraintKey(schema.getConstraintKeys())
.build();
}
+
+ private TableSchema applyColumnComment(
+ TableSchema schema, AlterColumnCommentEvent columnCommentEvent) {
+ List<String> fieldNames = Arrays.asList(schema.getFieldNames());
+ if (!fieldNames.contains(columnCommentEvent.getColumn())) {
+ return schema;
+ }
+ int columnIndex = fieldNames.indexOf(columnCommentEvent.getColumn());
+ LinkedList<Column> columns = new LinkedList<>(schema.getColumns());
+ columns.set(
+ columnIndex,
+
columns.get(columnIndex).copyWithComment(columnCommentEvent.getNewComment()));
+ return TableSchema.builder()
+ .columns(columns)
+ .primaryKey(schema.getPrimaryKey())
+ .constraintKey(schema.getConstraintKeys())
+ .build();
+ }
}
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/TableSchemaChangeEventDispatcher.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/TableSchemaChangeEventDispatcher.java
index 37cef6c5c3..8daf1ca4fd 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/TableSchemaChangeEventDispatcher.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/schema/handler/TableSchemaChangeEventDispatcher.java
@@ -18,9 +18,11 @@
package org.apache.seatunnel.api.table.schema.handler;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
@@ -74,6 +76,8 @@ public class TableSchemaChangeEventDispatcher implements
TableSchemaChangeEventH
handlers.put(AlterTableModifyColumnEvent.class,
alterTableEventHandler);
handlers.put(AlterTableDropColumnEvent.class, alterTableEventHandler);
handlers.put(AlterTableChangeColumnEvent.class,
alterTableEventHandler);
+ handlers.put(AlterTableCommentEvent.class, alterTableEventHandler);
+ handlers.put(AlterColumnCommentEvent.class, alterTableEventHandler);
return handlers;
}
}
diff --git
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/schema/event/EventTest.java
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/schema/event/EventTest.java
index 77fad92d66..a19c1979c5 100644
---
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/schema/event/EventTest.java
+++
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/schema/event/EventTest.java
@@ -21,6 +21,8 @@ import org.apache.seatunnel.api.event.EventType;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import
org.apache.seatunnel.api.table.schema.handler.AlterTableSchemaEventHandler;
import org.apache.seatunnel.api.table.type.BasicType;
import org.junit.jupiter.api.Assertions;
@@ -65,6 +67,31 @@ public class EventTest {
Assertions.assertEquals(EventType.SCHEMA_CHANGE_DROP_COLUMN,
getEventType(dropColumnEvent));
}
+ @Test
+ public void testAlterColumnCommentEventUpdatesSchemaMetadata() {
+ TableSchema schema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "description",
+ BasicType.STRING_TYPE,
+ 512L,
+ true,
+ null,
+ "old comment"))
+ .build();
+ AlterColumnCommentEvent event =
+ AlterColumnCommentEvent.of(
+ TableIdentifier.of("", TablePath.DEFAULT),
+ "description",
+ "old comment",
+ "new comment");
+
+ TableSchema changeAfter = new
AlterTableSchemaEventHandler().reset(schema).apply(event);
+
+ Assertions.assertEquals("new comment",
changeAfter.getColumn("description").getComment());
+ }
+
private EventType getEventType(AlterTableColumnEvent event) {
if (event instanceof AlterTableAddColumnEvent) {
return EventType.SCHEMA_CHANGE_ADD_COLUMN;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolver.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolver.java
index 49fe1ce7e7..83b4880d38 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolver.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolver.java
@@ -24,10 +24,15 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils;
@@ -87,9 +92,45 @@ public abstract class AbstractSchemaChangeResolver
implements SchemaChangeResolv
ddlParser.setCurrentSchema(tablePath.getSchemaName());
// Parse DDL statement using Debezium's Antlr parser
ddlParser.parse(ddl, tables);
- List<AlterTableColumnEvent> parsedEvents = getAndClearParsedEvents();
- parsedEvents = completionEvent(parsedEvents, catalogTables);
+ List<AlterTableEvent> parsedEvents =
getAndClearParsedSchemaChangeEvents();
+ parsedEvents = completeSchemaChangeEvents(parsedEvents, catalogTables,
tablePath);
parsedEvents.forEach(e ->
e.setSourceDialectName(getSourceDialectName()));
+
+ if (parsedEvents.isEmpty()) {
+ return null;
+ }
+
+ // If there's a single table-level comment event, return it directly
+ if (parsedEvents.size() == 1 && parsedEvents.get(0) instanceof
AlterTableCommentEvent) {
+ AlterTableCommentEvent commentEvent = (AlterTableCommentEvent)
parsedEvents.get(0);
+ commentEvent.setStatement(ddl);
+ return commentEvent;
+ }
+
+ // Filter column events for AlterTableColumnsEvent
+ List<AlterTableColumnEvent> columnEvents =
+ parsedEvents.stream()
+ .filter(e -> e instanceof AlterTableColumnEvent)
+ .map(e -> (AlterTableColumnEvent) e)
+ .collect(Collectors.toList());
+
+ if (columnEvents.isEmpty()) {
+ return null;
+ }
+
+ // Warn if non-column events (e.g. table comment changes) are present
alongside column
+ // events, since only column events can be batched in
AlterTableColumnsEvent and the
+ // others will not be propagated for this DDL.
+ long droppedCount = parsedEvents.size() - columnEvents.size();
+ if (droppedCount > 0) {
+ log.warn(
+ "DDL '{}' produced {} non-column event(s) alongside column
events; "
+ + "only column changes will be propagated. "
+ + "Non-column events (e.g. table comment changes)
in mixed DDL are not yet supported.",
+ ddl,
+ droppedCount);
+ }
+
AlterTableColumnsEvent alterTableColumnsEvent =
new AlterTableColumnsEvent(
TableIdentifier.of(
@@ -97,69 +138,163 @@ public abstract class AbstractSchemaChangeResolver
implements SchemaChangeResolv
tablePath.getDatabaseName(),
tablePath.getSchemaName(),
tablePath.getTableName()),
- parsedEvents);
+ columnEvents);
alterTableColumnsEvent.setStatement(ddl);
alterTableColumnsEvent.setSourceDialectName(getSourceDialectName());
- return parsedEvents.isEmpty() ? null : alterTableColumnsEvent;
+ return alterTableColumnsEvent;
}
List<AlterTableColumnEvent> completionEvent(
List<AlterTableColumnEvent> events, List<CatalogTable>
catalogTables) {
+ return completeSchemaChangeEvents(
+ Lists.newArrayList(events),
+ catalogTables,
+ events.isEmpty() ? null : events.get(0).getTablePath())
+ .stream()
+ .filter(event -> event instanceof AlterTableColumnEvent)
+ .map(event -> (AlterTableColumnEvent) event)
+ .collect(Collectors.toList());
+ }
+
+ List<AlterTableEvent> completeSchemaChangeEvents(
+ List<AlterTableEvent> events, List<CatalogTable> catalogTables,
TablePath tablePath) {
return events.stream()
.map(
- columnEvent -> {
-
columnEvent.setSourceDialectName(getSourceDialectName());
+ event -> {
+ event.setSourceDialectName(getSourceDialectName());
if (catalogTables == null ||
catalogTables.isEmpty()) {
- return columnEvent;
+ return event;
}
- if (!(columnEvent instanceof
AlterTableChangeColumnEvent)) {
- return columnEvent;
+
+ CatalogTable table =
findCatalogTable(catalogTables, tablePath);
+
+ // Handle table comment event - fill in old comment
+ if (event instanceof AlterTableCommentEvent) {
+ AlterTableCommentEvent commentEvent =
+ (AlterTableCommentEvent) event;
+ if (table != null &&
commentEvent.getOldComment() == null) {
+ String oldComment = table.getComment();
+ AlterTableCommentEvent newEvent =
+ AlterTableCommentEvent.of(
+
commentEvent.getTableIdentifier(),
+ oldComment,
+
commentEvent.getNewComment());
+
newEvent.setSourceDialectName(getSourceDialectName());
+ return newEvent;
+ }
+ return event;
}
- AlterTableChangeColumnEvent changeColumnEvent =
- (AlterTableChangeColumnEvent) columnEvent;
- if (changeColumnEvent.getColumn().getDataType() !=
null) {
- return columnEvent;
+ // Handle column change event - complete type info
+ if (event instanceof AlterTableChangeColumnEvent) {
+ AlterTableChangeColumnEvent changeColumnEvent =
+ (AlterTableChangeColumnEvent) event;
+ if
(changeColumnEvent.getColumn().getDataType() != null) {
+ return event;
+ }
+ if (table != null) {
+ Column oldColumn =
+ table.getTableSchema()
+
.getColumn(changeColumnEvent.getOldColumn());
+ Column newColumn =
+ oldColumn.rename(
+
changeColumnEvent.getColumn().getName());
+ AlterTableChangeColumnEvent newEvent =
+ new AlterTableChangeColumnEvent(
+
changeColumnEvent.getTableIdentifier(),
+
changeColumnEvent.getOldColumn(),
+ newColumn,
+
changeColumnEvent.isFirst(),
+
changeColumnEvent.getAfterColumn());
+
newEvent.setSourceDialectName(getSourceDialectName());
+ return newEvent;
+ } else {
+ log.warn(
+ "Ignoring rename column {} type
completion for table {}",
+ changeColumnEvent.getOldColumn(),
+ changeColumnEvent.getTablePath());
+ }
}
- CatalogTable table =
- catalogTables.stream()
- .filter(
- catalogTable ->
- catalogTable
-
.getTablePath()
- .equals(
-
columnEvent
-
.getTablePath()))
- .findFirst()
- .orElse(null);
- if (table != null) {
- Column oldColumn =
- table.getTableSchema()
-
.getColumn(changeColumnEvent.getOldColumn());
- Column newColumn =
-
oldColumn.rename(changeColumnEvent.getColumn().getName());
- AlterTableChangeColumnEvent newEvent =
- new AlterTableChangeColumnEvent(
-
changeColumnEvent.getTableIdentifier(),
-
changeColumnEvent.getOldColumn(),
- newColumn,
- changeColumnEvent.isFirst(),
-
changeColumnEvent.getAfterColumn());
-
newEvent.setSourceDialectName(getSourceDialectName());
- return newEvent;
- } else {
- log.warn(
- "Ignoring rename column {} type
completion for table {}",
- changeColumnEvent.getOldColumn(),
- changeColumnEvent.getTablePath());
+ if (event instanceof AlterTableModifyColumnEvent
&& table != null) {
+ AlterTableModifyColumnEvent modifyColumnEvent =
+ (AlterTableModifyColumnEvent) event;
+ if (table.getTableSchema()
+
.contains(modifyColumnEvent.getColumn().getName())) {
+ Column oldColumn =
+ table.getTableSchema()
+ .getColumn(
+ modifyColumnEvent
+
.getColumn()
+
.getName());
+ AlterColumnCommentEvent columnCommentEvent
=
+
convertToColumnCommentEventIfOnlyCommentChanged(
+ modifyColumnEvent,
oldColumn);
+ if (columnCommentEvent != null) {
+
columnCommentEvent.setSourceDialectName(
+ getSourceDialectName());
+ return columnCommentEvent;
+ }
+ }
}
- return columnEvent;
+ return event;
})
.collect(Collectors.toList());
}
+ private CatalogTable findCatalogTable(List<CatalogTable> catalogTables,
TablePath tablePath) {
+ if (tablePath == null) {
+ return null;
+ }
+ return catalogTables.stream()
+ .filter(catalogTable ->
catalogTable.getTablePath().equals(tablePath))
+ .findFirst()
+ .orElse(null);
+ }
+
+ private AlterColumnCommentEvent
convertToColumnCommentEventIfOnlyCommentChanged(
+ AlterTableModifyColumnEvent modifyColumnEvent, Column oldColumn) {
+ if (oldColumn == null
+ || modifyColumnEvent.isFirst()
+ || StringUtils.isNotBlank(modifyColumnEvent.getAfterColumn()))
{
+ return null;
+ }
+ Column newColumn = modifyColumnEvent.getColumn();
+ if (isSameColumnExceptComment(oldColumn, newColumn)
+ && !StringUtils.equals(oldColumn.getComment(),
newColumn.getComment())) {
+ return AlterColumnCommentEvent.of(
+ modifyColumnEvent.getTableIdentifier(),
+ newColumn.getName(),
+ oldColumn.getComment(),
+ newColumn.getComment());
+ }
+ return null;
+ }
+
+ private boolean isSameColumnExceptComment(Column oldColumn, Column
newColumn) {
+ return StringUtils.equals(oldColumn.getName(), newColumn.getName())
+ && isSameDataType(oldColumn.getDataType(),
newColumn.getDataType())
+ && Objects.equals(oldColumn.getColumnLength(),
newColumn.getColumnLength())
+ && Objects.equals(oldColumn.getScale(), newColumn.getScale())
+ && oldColumn.isNullable() == newColumn.isNullable()
+ && Objects.equals(oldColumn.getDefaultValue(),
newColumn.getDefaultValue())
+ && StringUtils.equals(oldColumn.getSourceType(),
newColumn.getSourceType())
+ && Objects.equals(oldColumn.getOptions(),
newColumn.getOptions());
+ }
+
+ private boolean isSameDataType(
+ SeaTunnelDataType<?> oldDataType, SeaTunnelDataType<?>
newDataType) {
+ return Objects.equals(oldDataType, newDataType)
+ || (oldDataType != null
+ && newDataType != null
+ && oldDataType.getSqlType() ==
newDataType.getSqlType());
+ }
+
protected abstract DdlParser createDdlParser(TablePath tablePath);
+ protected List<AlterTableEvent> getAndClearParsedSchemaChangeEvents() {
+ return Lists.newArrayList(getAndClearParsedEvents());
+ }
+
protected abstract List<AlterTableColumnEvent> getAndClearParsedEvents();
protected abstract String getSourceDialectName();
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
index 4c237b1bcd..0732a121a9 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
@@ -24,6 +24,7 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandler;
@@ -180,7 +181,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
changeAfterSchema,
changeBefore.getOptions(),
changeBefore.getPartitionKeys(),
- changeBefore.getComment());
+ getChangeAfterTableComment(changeBefore,
event));
event.setChangeAfter(changeAfter);
changeBefore = changeAfter;
@@ -196,7 +197,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
changeAfterSchema,
changeBefore.getOptions(),
changeBefore.getPartitionKeys(),
- changeBefore.getComment());
+ getChangeAfterTableComment(changeBefore,
schemaChangeEvent));
}
tables.set(i, changeAfter);
schemaChangeEvent.setChangeAfter(changeAfter);
@@ -217,6 +218,14 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
collector.collect(schemaChangeEvent);
}
+ private String getChangeAfterTableComment(
+ CatalogTable changeBefore, SchemaChangeEvent schemaChangeEvent) {
+ if (schemaChangeEvent instanceof AlterTableCommentEvent) {
+ return ((AlterTableCommentEvent)
schemaChangeEvent).getNewComment();
+ }
+ return changeBefore.getComment();
+ }
+
private void deserializeDataChangeRecord(SourceRecord record,
Collector<SeaTunnelRow> collector)
throws Exception {
Envelope.Operation operation = Envelope.operationFor(record);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolverTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolverTest.java
index 1565a25eb9..f69f8338e4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolverTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolverTest.java
@@ -22,10 +22,13 @@ import
org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.type.BasicType;
-import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -36,30 +39,11 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import static org.mockito.Mockito.mock;
-
public class AbstractSchemaChangeResolverTest {
@Test
void testCompletionEvent() {
- JdbcSourceConfig config = mock(JdbcSourceConfig.class);
- AbstractSchemaChangeResolver resolver =
- new AbstractSchemaChangeResolver(config) {
- @Override
- protected DdlParser createDdlParser(TablePath tablePath) {
- return null;
- }
-
- @Override
- protected List<AlterTableColumnEvent>
getAndClearParsedEvents() {
- return Collections.emptyList();
- }
-
- @Override
- protected String getSourceDialectName() {
- return "mysql";
- }
- };
+ AbstractSchemaChangeResolver resolver = createResolver();
AlterTableChangeColumnEvent changeColumnEvent =
AlterTableChangeColumnEvent.change(
@@ -92,4 +76,136 @@ public class AbstractSchemaChangeResolverTest {
Assertions.assertEquals(1L,
changeColumnEvent.getColumn().getColumnLength());
Assertions.assertEquals("column comment",
changeColumnEvent.getColumn().getComment());
}
+
+ @Test
+ void testCompletionEventFillsTableComment() {
+ AbstractSchemaChangeResolver resolver = createResolver();
+ AlterTableCommentEvent tableCommentEvent =
+ AlterTableCommentEvent.of(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ null,
+ "new table comment");
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ TableSchema.builder().build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "old table comment",
+ null);
+
+ List<AlterTableEvent> events =
+ resolver.completeSchemaChangeEvents(
+ Collections.singletonList(tableCommentEvent),
+ Collections.singletonList(catalogTable),
+ TablePath.of("test_db", "test_table"));
+
+ AlterTableCommentEvent completedEvent = (AlterTableCommentEvent)
events.get(0);
+ Assertions.assertEquals("mysql",
completedEvent.getSourceDialectName());
+ Assertions.assertEquals("old table comment",
completedEvent.getOldComment());
+ Assertions.assertEquals("new table comment",
completedEvent.getNewComment());
+ }
+
+ @Test
+ void testCompletionEventConvertsModifyColumnToColumnCommentEvent() {
+ AbstractSchemaChangeResolver resolver = createResolver();
+ AlterTableModifyColumnEvent modifyColumnEvent =
+ AlterTableModifyColumnEvent.modify(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ PhysicalColumn.builder()
+ .name("description")
+ .dataType(BasicType.STRING_TYPE)
+ .columnLength(512L)
+ .nullable(true)
+ .comment("new column comment")
+ .build());
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.builder()
+ .name("description")
+
.dataType(BasicType.STRING_TYPE)
+ .columnLength(512L)
+ .nullable(true)
+ .comment("old column comment")
+ .build())
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null,
+ null);
+
+ List<AlterTableEvent> events =
+ resolver.completeSchemaChangeEvents(
+ Collections.singletonList(modifyColumnEvent),
+ Collections.singletonList(catalogTable),
+ TablePath.of("test_db", "test_table"));
+
+ AlterColumnCommentEvent completedEvent = (AlterColumnCommentEvent)
events.get(0);
+ Assertions.assertEquals("mysql",
completedEvent.getSourceDialectName());
+ Assertions.assertEquals("description", completedEvent.getColumn());
+ Assertions.assertEquals("old column comment",
completedEvent.getOldComment());
+ Assertions.assertEquals("new column comment",
completedEvent.getNewComment());
+ }
+
+ @Test
+ void testCompletionEventKeepsStructuralModifyColumnEvent() {
+ AbstractSchemaChangeResolver resolver = createResolver();
+ AlterTableModifyColumnEvent modifyColumnEvent =
+ AlterTableModifyColumnEvent.modify(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ PhysicalColumn.builder()
+ .name("description")
+ .dataType(BasicType.STRING_TYPE)
+ .columnLength(1024L)
+ .nullable(true)
+ .comment("new column comment")
+ .build());
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of(null, "test_db", "test_table"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.builder()
+ .name("description")
+
.dataType(BasicType.STRING_TYPE)
+ .columnLength(512L)
+ .nullable(true)
+ .comment("old column comment")
+ .build())
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null,
+ null);
+
+ List<AlterTableEvent> events =
+ resolver.completeSchemaChangeEvents(
+ Collections.singletonList(modifyColumnEvent),
+ Collections.singletonList(catalogTable),
+ TablePath.of("test_db", "test_table"));
+
+ Assertions.assertSame(modifyColumnEvent, events.get(0));
+ }
+
+ private AbstractSchemaChangeResolver createResolver() {
+ return new AbstractSchemaChangeResolver(null) {
+ @Override
+ protected DdlParser createDdlParser(TablePath tablePath) {
+ return null;
+ }
+
+ @Override
+ protected List<AlterTableColumnEvent> getAndClearParsedEvents() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected String getSourceDialectName() {
+ return "mysql";
+ }
+ };
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
index 7a12494447..20aef0f110 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
@@ -322,7 +322,9 @@ public class MySqlIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceCo
SchemaChangeType.ADD_COLUMN,
SchemaChangeType.DROP_COLUMN,
SchemaChangeType.RENAME_COLUMN,
- SchemaChangeType.UPDATE_COLUMN);
+ SchemaChangeType.UPDATE_COLUMN,
+ SchemaChangeType.ALTER_TABLE_COMMENT,
+ SchemaChangeType.ALTER_COLUMN_COMMENT);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlSchemaChangeResolver.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlSchemaChangeResolver.java
index de0e621c8b..ffd7dc6b46 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlSchemaChangeResolver.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlSchemaChangeResolver.java
@@ -19,6 +19,7 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
import
org.apache.seatunnel.connectors.cdc.base.schema.AbstractSchemaChangeResolver;
@@ -41,10 +42,15 @@ public class MySqlSchemaChangeResolver extends
AbstractSchemaChangeResolver {
}
@Override
- protected List<AlterTableColumnEvent> getAndClearParsedEvents() {
+ protected List<AlterTableEvent> getAndClearParsedSchemaChangeEvents() {
return ((CustomMySqlAntlrDdlParser)
ddlParser).getAndClearParsedEvents();
}
+ @Override
+ protected List<AlterTableColumnEvent> getAndClearParsedEvents() {
+ return ((CustomMySqlAntlrDdlParser)
ddlParser).getAndClearParsedColumnEvents();
+ }
+
@Override
protected String getSourceDialectName() {
return DatabaseIdentifier.MYSQL;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java
index 55dcb88abc..60c391f7be 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java
@@ -22,8 +22,9 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import
org.apache.seatunnel.connectors.cdc.base.source.parser.SeatunnelDDLParser;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlTypeUtils;
@@ -46,7 +47,7 @@ public class CustomAlterTableParserListener extends
MySqlParserBaseListener
private static final int STARTING_INDEX = 1;
private final MySqlAntlrDdlParser parser;
private final List<ParseTreeListener> listeners;
- private final LinkedList<AlterTableColumnEvent> changes;
+ private final LinkedList<AlterTableEvent> changes;
private List<ColumnEditor> columnEditors;
private TableIdentifier tableIdentifier;
@@ -60,7 +61,7 @@ public class CustomAlterTableParserListener extends
MySqlParserBaseListener
RelationalDatabaseConnectorConfig dbzConnectorConfig,
MySqlAntlrDdlParser parser,
List<ParseTreeListener> listeners,
- LinkedList<AlterTableColumnEvent> changes) {
+ LinkedList<AlterTableEvent> changes) {
this.dbzConnectorConfig = dbzConnectorConfig;
this.parser = parser;
this.listeners = listeners;
@@ -219,6 +220,25 @@ public class CustomAlterTableParserListener extends
MySqlParserBaseListener
super.enterAlterByDropColumn(ctx);
}
+ @Override
+ public void enterTableOptionComment(MySqlParser.TableOptionCommentContext
ctx) {
+ if (!parser.skipComments()) {
+ parser.runIfNotNull(
+ () -> {
+ if (ctx.COMMENT() != null && ctx.STRING_LITERAL() !=
null) {
+ String newComment =
+
parser.withoutQuotes(ctx.STRING_LITERAL().getText());
+ // Old comment is not available from DDL parsing
alone;
+ // it will be filled in by the resolver from the
catalog table
+ changes.add(
+ AlterTableCommentEvent.of(tableIdentifier,
null, newComment));
+ }
+ },
+ tableIdentifier);
+ }
+ super.enterTableOptionComment(ctx);
+ }
+
@Override
public org.apache.seatunnel.api.table.catalog.Column
toSeatunnelColumn(Column column) {
return MySqlTypeUtils.convertToSeaTunnelColumn(column,
dbzConnectorConfig);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParser.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParser.java
index d5f8a59254..d484ae6200 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParser.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParser.java
@@ -21,28 +21,32 @@ import
org.apache.seatunnel.shade.com.google.common.collect.Lists;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import io.debezium.antlr.AntlrDdlParserListener;
import io.debezium.antlr.DataTypeResolver;
+import io.debezium.connector.mysql.MySqlValueConverters;
import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
import io.debezium.ddl.parser.mysql.generated.MySqlParser;
import io.debezium.relational.RelationalDatabaseConnectorConfig;
import io.debezium.relational.TableId;
+import io.debezium.relational.Tables;
import java.sql.Types;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
+import java.util.stream.Collectors;
/** A ddl parser that will use custom listener. */
public class CustomMySqlAntlrDdlParser extends MySqlAntlrDdlParser {
- private final LinkedList<AlterTableColumnEvent> parsedEvents;
+ private final LinkedList<AlterTableEvent> parsedEvents;
private RelationalDatabaseConnectorConfig dbzConnectorConfig;
public CustomMySqlAntlrDdlParser(RelationalDatabaseConnectorConfig
dbzConnectorConfig) {
- super();
+ super(false, false, true, (MySqlValueConverters) null,
Tables.TableFilter.includeAll());
this.parsedEvents = new LinkedList<>();
this.dbzConnectorConfig = dbzConnectorConfig;
}
@@ -306,9 +310,16 @@ public class CustomMySqlAntlrDdlParser extends
MySqlAntlrDdlParser {
return new CustomMySqlAntlrDdlParserListener(dbzConnectorConfig, this,
parsedEvents);
}
- public List<AlterTableColumnEvent> getAndClearParsedEvents() {
- List<AlterTableColumnEvent> result = Lists.newArrayList(parsedEvents);
+ public List<AlterTableEvent> getAndClearParsedEvents() {
+ List<AlterTableEvent> result = Lists.newArrayList(parsedEvents);
parsedEvents.clear();
return result;
}
+
+ public List<AlterTableColumnEvent> getAndClearParsedColumnEvents() {
+ return getAndClearParsedEvents().stream()
+ .filter(event -> event instanceof AlterTableColumnEvent)
+ .map(event -> (AlterTableColumnEvent) event)
+ .collect(Collectors.toList());
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java
index c8be8460a3..2ed14ee942 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java
@@ -17,7 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.parser;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.antlr.v4.runtime.ParserRuleContext;
import org.antlr.v4.runtime.tree.ErrorNode;
@@ -61,7 +61,7 @@ public class CustomMySqlAntlrDdlParserListener extends
MySqlParserBaseListener
public CustomMySqlAntlrDdlParserListener(
RelationalDatabaseConnectorConfig dbzConnectorConfig,
MySqlAntlrDdlParser parser,
- LinkedList<AlterTableColumnEvent> parsedEvents) {
+ LinkedList<AlterTableEvent> parsedEvents) {
// Currently only DDL statements that modify the table structure are
supported, so add
// custom listeners to handle these events.
listeners.add(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java
index 2ff84a440b..54f352ab73 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java
@@ -91,7 +91,8 @@ public class MySqlTypeUtils {
.dataType(column.typeName())
.scale(column.scale().orElse(0))
.nullable(column.isOptional())
- .defaultValue(defaultValue);
+ .defaultValue(defaultValue)
+ .comment(column.comment());
if (column.length() >= 0) {
builder.length((long) column.length()).precision((long)
column.length());
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java
index 671ee42852..5f209b7e87 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java
@@ -18,6 +18,8 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.parser;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfigFactory;
@@ -50,7 +52,7 @@ public class CustomMySqlAntlrDdlParserTest {
parser.parse(
"ALTER TABLE products ADD COLUMN add_column1 VARCHAR(64) NOT
NULL DEFAULT 'db-a'",
new Tables());
- List<AlterTableColumnEvent> firstDatabaseEvents =
parser.getAndClearParsedEvents();
+ List<AlterTableColumnEvent> firstDatabaseEvents =
parser.getAndClearParsedColumnEvents();
Assertions.assertEquals(1, firstDatabaseEvents.size());
Assertions.assertEquals(
"multi_schema_shop_a",
@@ -61,7 +63,7 @@ public class CustomMySqlAntlrDdlParserTest {
parser.setCurrentDatabase("multi_schema_shop_b");
parser.parse(
"ALTER TABLE products ADD COLUMN add_column2 INT NOT NULL
DEFAULT 1", new Tables());
- List<AlterTableColumnEvent> secondDatabaseEvents =
parser.getAndClearParsedEvents();
+ List<AlterTableColumnEvent> secondDatabaseEvents =
parser.getAndClearParsedColumnEvents();
Assertions.assertEquals(1, secondDatabaseEvents.size());
Assertions.assertEquals(
"multi_schema_shop_b",
@@ -69,4 +71,18 @@ public class CustomMySqlAntlrDdlParserTest {
Assertions.assertEquals(
"products",
secondDatabaseEvents.get(0).getTablePath().getTableName());
}
+
+ @Test
+ void testParseAlterTableCommentEvent() {
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser(null);
+
+ parser.setCurrentDatabase("test_db");
+ parser.parse("ALTER TABLE products COMMENT = 'Product catalog table'",
new Tables());
+
+ List<AlterTableEvent> events = parser.getAndClearParsedEvents();
+ Assertions.assertEquals(1, events.size());
+ Assertions.assertTrue(events.get(0) instanceof AlterTableCommentEvent);
+ AlterTableCommentEvent event = (AlterTableCommentEvent) events.get(0);
+ Assertions.assertEquals("Product catalog table",
event.getNewComment());
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSchemaChangeResolver.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSchemaChangeResolver.java
index 5be13ffc39..2429317269 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSchemaChangeResolver.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleSchemaChangeResolver.java
@@ -28,6 +28,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseI
import io.debezium.relational.ddl.DdlParser;
import java.util.List;
+import java.util.stream.Collectors;
public class OracleSchemaChangeResolver extends AbstractSchemaChangeResolver {
public OracleSchemaChangeResolver(SourceConfig.Factory<JdbcSourceConfig>
sourceConfigFactory) {
@@ -41,7 +42,11 @@ public class OracleSchemaChangeResolver extends
AbstractSchemaChangeResolver {
@Override
protected List<AlterTableColumnEvent> getAndClearParsedEvents() {
- return ((CustomOracleAntlrDdlParser)
ddlParser).getAndClearParsedEvents();
+ return ((CustomOracleAntlrDdlParser) ddlParser)
+ .getAndClearParsedEvents().stream()
+ .filter(event -> event instanceof
AlterTableColumnEvent)
+ .map(event -> (AlterTableColumnEvent) event)
+ .collect(Collectors.toList());
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomAlterTableParserListener.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomAlterTableParserListener.java
index 61abeec602..24753bc984 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomAlterTableParserListener.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomAlterTableParserListener.java
@@ -23,8 +23,8 @@ import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import
org.apache.seatunnel.connectors.cdc.base.source.parser.SeatunnelDDLParser;
import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.utils.OracleTypeUtils;
@@ -52,13 +52,13 @@ public class CustomAlterTableParserListener extends
BaseParserListener
private List<ColumnEditor> columnEditors;
private int parsingColumnIndex = STARTING_INDEX;
- private final LinkedList<AlterTableColumnEvent> changes;
+ private final LinkedList<AlterTableEvent> changes;
private TableIdentifier tableIdentifier;
public CustomAlterTableParserListener(
CustomOracleAntlrDdlParser parser,
List<ParseTreeListener> listeners,
- LinkedList<AlterTableColumnEvent> changes) {
+ LinkedList<AlterTableEvent> changes) {
this.parser = parser;
this.listeners = listeners;
this.changes = changes;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParser.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParser.java
index f8146dc8f2..c6238e5646 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParser.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParser.java
@@ -20,7 +20,7 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.parser;
import org.apache.seatunnel.shade.com.google.common.collect.Lists;
import org.apache.seatunnel.api.table.catalog.TablePath;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import io.debezium.antlr.AntlrDdlParserListener;
import io.debezium.connector.oracle.antlr.OracleDdlParser;
@@ -32,7 +32,7 @@ import java.util.List;
/** A ddl parser that will use custom listener. */
public class CustomOracleAntlrDdlParser extends OracleDdlParser {
- private final LinkedList<AlterTableColumnEvent> parsedEvents;
+ private final LinkedList<AlterTableEvent> parsedEvents;
private final TablePath tablePath;
@@ -52,8 +52,8 @@ public class CustomOracleAntlrDdlParser extends
OracleDdlParser {
return new CustomOracleAntlrDdlParserListener(this, parsedEvents);
}
- public List<AlterTableColumnEvent> getAndClearParsedEvents() {
- List<AlterTableColumnEvent> result = Lists.newArrayList(parsedEvents);
+ public List<AlterTableEvent> getAndClearParsedEvents() {
+ List<AlterTableEvent> result = Lists.newArrayList(parsedEvents);
parsedEvents.clear();
return result;
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParserListener.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParserListener.java
index eb7bbb5264..b650158dca 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParserListener.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/CustomOracleAntlrDdlParserListener.java
@@ -17,7 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.parser;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.antlr.v4.runtime.ParserRuleContext;
import org.antlr.v4.runtime.tree.ParseTreeListener;
@@ -40,7 +40,7 @@ public class CustomOracleAntlrDdlParserListener extends
BaseParserListener
private final Collection<ParsingException> errors = new ArrayList<>();
public CustomOracleAntlrDdlParserListener(
- CustomOracleAntlrDdlParser parser,
LinkedList<AlterTableColumnEvent> parsedEvents) {
+ CustomOracleAntlrDdlParser parser, LinkedList<AlterTableEvent>
parsedEvents) {
// Currently only DDL statements that modify the table structure are
supported, so add
// custom listeners to handle these events.
listeners.add(new CustomAlterTableParserListener(parser, listeners,
parsedEvents));
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/OracleDdlParserTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/OracleDdlParserTest.java
index 5bcf98938b..8d814e86bf 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/OracleDdlParserTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/parser/OracleDdlParserTest.java
@@ -21,8 +21,8 @@ import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
-import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter;
@@ -57,7 +57,7 @@ public class OracleDdlParserTest {
+ "\" add ("
+ "\"col21\" varchar2(20), col22 number(19));";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> addEvent1 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> addEvent1 = parser.getAndClearParsedEvents();
Assertions.assertEquals(2, addEvent1.size());
testColumn(addEvent1.get(0), "col21", "varchar2(20)", "STRING", 20 *
4L, null, true, null);
testColumn(
@@ -72,7 +72,7 @@ public class OracleDdlParserTest {
ddl = "alter table " + TABLE_NAME + " add (col23 varchar2(20) not
null);";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> addEvent2 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> addEvent2 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, addEvent2.size());
testColumn(
addEvent2.get(0),
@@ -105,7 +105,7 @@ public class OracleDdlParserTest {
+ "col15 timestamp(9) default to_timestamp('20190101
00:00:00.000000','yyyymmdd hh24:mi:ss.ff6') not null,\n"
+ "col16 date default sysdate not null);";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> addEvent3 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> addEvent3 = parser.getAndClearParsedEvents();
Assertions.assertEquals(16, addEvent3.size());
// Special default values are handled for reference:
//
io.debezium.connector.oracle.OracleDefaultValueConverter.castTemporalFunctionCall
@@ -261,7 +261,7 @@ public class OracleDdlParserTest {
+ TABLE_NAME
+ "\" ADD \"ADD_COL2\" TIMESTAMP(6) DEFAULT
current_timestamp(6) NOT NULL ";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> addEvent4 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> addEvent4 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, addEvent4.size());
testColumn(
addEvent4.get(0),
@@ -278,14 +278,14 @@ public class OracleDdlParserTest {
public void testParseDDLForDropColumn() {
String ddl = "ALTER TABLE \"" + SCHEMA_NAME + "\".\"" + TABLE_NAME +
"\" DROP (T_VARCHAR2)";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> dropEvent1 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> dropEvent1 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, dropEvent1.size());
Assertions.assertEquals(
"T_VARCHAR2", ((AlterTableDropColumnEvent)
dropEvent1.get(0)).getColumn());
ddl = "alter table " + TABLE_NAME + " drop (col22, col23);";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> dropEvent2 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> dropEvent2 = parser.getAndClearParsedEvents();
Assertions.assertEquals(2, dropEvent2.size());
Assertions.assertEquals(
"col22".toUpperCase(), ((AlterTableDropColumnEvent)
dropEvent2.get(0)).getColumn());
@@ -294,7 +294,7 @@ public class OracleDdlParserTest {
ddl = "alter table " + TABLE_NAME + " drop (\"col22\");";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> dropEvent3 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> dropEvent3 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, dropEvent3.size());
Assertions.assertEquals(
"col22", ((AlterTableDropColumnEvent)
dropEvent3.get(0)).getColumn());
@@ -304,7 +304,7 @@ public class OracleDdlParserTest {
public void testParseDDLForRenameColumn() {
String ddl = "alter table " + TABLE_NAME + " rename column
STUDENT_NAME to STUDENT_NAME1";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> renameEvent1 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> renameEvent1 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, renameEvent1.size());
Assertions.assertEquals(
"STUDENT_NAME", ((AlterTableChangeColumnEvent)
renameEvent1.get(0)).getOldColumn());
@@ -321,7 +321,7 @@ public class OracleDdlParserTest {
+ "\" rename column CLASS_ID to CLASS_ID1\n";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> renameEvent2 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> renameEvent2 = parser.getAndClearParsedEvents();
Assertions.assertEquals(2, renameEvent2.size());
Assertions.assertEquals(
"STUDENT_ID", ((AlterTableChangeColumnEvent)
renameEvent2.get(0)).getOldColumn());
@@ -339,14 +339,14 @@ public class OracleDdlParserTest {
public void testParseDDLForModifyColumn() {
String ddl = "ALTER TABLE " + TABLE_NAME + " MODIFY COL1 varchar2(50)
not null;";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> modifyEvent1 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> modifyEvent1 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, modifyEvent1.size());
testColumn(
modifyEvent1.get(0), "COL1", "varchar2(50)", "STRING", 50 *
4L, null, false, null);
ddl = "alter table " + TABLE_NAME + " modify sex char(2) default 'M'
not null ;";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> modifyEvent2 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> modifyEvent2 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, modifyEvent2.size());
testColumn(
modifyEvent2.get(0),
@@ -364,7 +364,7 @@ public class OracleDdlParserTest {
+ TABLE_NAME
+ "\" MODIFY (ID NUMBER(*,0) NULL);";
parser.parse(ddl, new Tables());
- List<AlterTableColumnEvent> modifyEvent3 =
parser.getAndClearParsedEvents();
+ List<AlterTableEvent> modifyEvent3 = parser.getAndClearParsedEvents();
Assertions.assertEquals(1, modifyEvent3.size());
testColumn(
modifyEvent3.get(0),
@@ -378,7 +378,7 @@ public class OracleDdlParserTest {
}
private void testColumn(
- AlterTableColumnEvent alterTableColumnEvent,
+ AlterTableEvent alterTableEvent,
String columnName,
String sourceType,
String dataType,
@@ -387,17 +387,17 @@ public class OracleDdlParserTest {
boolean isNullable,
Object defaultValue) {
Column column;
- switch (alterTableColumnEvent.getEventType()) {
+ switch (alterTableEvent.getEventType()) {
case SCHEMA_CHANGE_ADD_COLUMN:
- column = ((AlterTableAddColumnEvent)
alterTableColumnEvent).getColumn();
+ column = ((AlterTableAddColumnEvent)
alterTableEvent).getColumn();
break;
case SCHEMA_CHANGE_MODIFY_COLUMN:
- column = ((AlterTableModifyColumnEvent)
alterTableColumnEvent).getColumn();
+ column = ((AlterTableModifyColumnEvent)
alterTableEvent).getColumn();
break;
default:
throw new UnsupportedOperationException(
- "Unsupported method named getColumn() for the
AlterTableColumnEvent: "
- + alterTableColumnEvent.getEventType().name());
+ "Unsupported method named getColumn() for the
AlterTableEvent: "
+ + alterTableEvent.getEventType().name());
}
Assertions.assertEquals(columnName, column.getName());
Assertions.assertEquals(sourceType.toUpperCase(),
column.getSourceType());
diff --git
a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/schema/SchemaChangeManager.java
b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/schema/SchemaChangeManager.java
index 50727d1775..5a533bb8bd 100644
---
a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/schema/SchemaChangeManager.java
+++
b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/schema/SchemaChangeManager.java
@@ -19,10 +19,12 @@ package
org.apache.seatunnel.connectors.seatunnel.databend.schema;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
@@ -71,6 +73,13 @@ public class SchemaChangeManager implements Serializable {
}
} else if (event instanceof AlterTableColumnEvent) {
applySchemaChange(connection, tablePath,
(AlterTableColumnEvent) event);
+ } else if (event instanceof AlterTableCommentEvent
+ || event instanceof AlterColumnCommentEvent) {
+ // Comment-only changes are not supported by Databend sink,
safely ignore
+ log.info(
+ "Ignoring comment change event for table {} - Databend
sink does not support comment sync: {}",
+ tablePath.getFullName(),
+ event.getClass().getSimpleName());
} else {
throw new SeaTunnelException(
"Unsupported schemaChangeEvent: " +
event.getClass().getName());
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/schema/SchemaChangeManager.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/schema/SchemaChangeManager.java
index 8c1a77f8f8..cd8ab76606 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/schema/SchemaChangeManager.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/schema/SchemaChangeManager.java
@@ -27,10 +27,12 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
@@ -133,6 +135,13 @@ public class SchemaChangeManager implements Serializable {
return;
}
applySchemaChange(tablePath, dropColumnEvent);
+ } else if (event instanceof AlterTableCommentEvent
+ || event instanceof AlterColumnCommentEvent) {
+ // Comment-only changes are not supported by Doris sink,
safely ignore
+ log.info(
+ "Ignoring comment change event for table {} - Doris
sink does not support comment sync: {}",
+ tablePath.getFullName(),
+ event.getEventType());
} else {
throw new SeaTunnelException(
"Unsupported schemaChangeEvent : " +
event.getEventType());
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java
index 50f779d85e..1b39d7708b 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java
@@ -25,9 +25,11 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandler;
@@ -132,7 +134,9 @@ public class ElasticsearchSinkWriter
@Override
public void applySchemaChange(SchemaChangeEvent event) throws IOException {
- if (event instanceof AlterTableColumnsEvent) {
+ if (isCommentOnlyEvent(event)) {
+ log.debug("Ignore comment-only schema change event: {}", event);
+ } else if (event instanceof AlterTableColumnsEvent) {
for (AlterTableColumnEvent columnEvent : ((AlterTableColumnsEvent)
event).getEvents()) {
applySingleSchemaChangeEvent(columnEvent);
}
@@ -159,8 +163,14 @@ public class ElasticsearchSinkWriter
vectorDimension);
}
+ static boolean isCommentOnlyEvent(SchemaChangeEvent event) {
+ return event instanceof AlterTableCommentEvent || event instanceof
AlterColumnCommentEvent;
+ }
+
private void applySingleSchemaChangeEvent(SchemaChangeEvent event) {
- if (event instanceof AlterTableAddColumnEvent) {
+ if (isCommentOnlyEvent(event)) {
+ log.debug("Ignore comment-only schema change event: {}", event);
+ } else if (event instanceof AlterTableAddColumnEvent) {
AlterTableAddColumnEvent addColumnEvent =
(AlterTableAddColumnEvent) event;
Column column = addColumnEvent.getColumn();
BasicTypeDefine<EsType> reconvert =
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriterTest.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriterTest.java
new file mode 100644
index 0000000000..672343eb22
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriterTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.elasticsearch.sink;
+
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ElasticsearchSinkWriterTest {
+
+ private static final TableIdentifier TABLE_IDENTIFIER =
+ TableIdentifier.of("", TablePath.DEFAULT);
+
+ @Test
+ void commentOnlySchemaChangeEventsAreNoOpForElasticsearch() {
+ Assertions.assertTrue(
+ ElasticsearchSinkWriter.isCommentOnlyEvent(
+ AlterTableCommentEvent.of(TABLE_IDENTIFIER, "old",
"new")));
+ Assertions.assertTrue(
+ ElasticsearchSinkWriter.isCommentOnlyEvent(
+ AlterColumnCommentEvent.of(TABLE_IDENTIFIER, "name",
"old", "new")));
+ }
+
+ @Test
+ void physicalSchemaChangeEventsStillRequireElasticsearchMappingChanges() {
+ Assertions.assertFalse(
+ ElasticsearchSinkWriter.isCommentOnlyEvent(
+ AlterTableAddColumnEvent.add(
+ TABLE_IDENTIFIER,
+ PhysicalColumn.builder()
+ .name("name")
+ .dataType(BasicType.STRING_TYPE)
+ .build())));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
index 95055a7d70..282bf70733 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/JdbcDialect.java
@@ -24,10 +24,12 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
import org.apache.seatunnel.api.table.converter.TypeConverter;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
@@ -556,6 +558,13 @@ public interface JdbcDialect extends Serializable {
return;
}
applySchemaChange(connection, tablePath, dropColumnEvent);
+ } else if (event instanceof AlterTableCommentEvent
+ || event instanceof AlterColumnCommentEvent) {
+ // Comment-only changes are not supported by JDBC sink, safely
ignore
+ log.info(
+ "Ignoring comment change event for table {} - JDBC
sink does not support comment sync: {}",
+ tablePath.getFullName(),
+ event.getClass().getSimpleName());
} else {
throw new UnsupportedOperationException("Unsupported
schemaChangeEvent: " + event);
}
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/handler/AlterPaimonTableSchemaEventHandler.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/handler/AlterPaimonTableSchemaEventHandler.java
index e0faa687b2..be8b799184 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/handler/AlterPaimonTableSchemaEventHandler.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/handler/AlterPaimonTableSchemaEventHandler.java
@@ -23,10 +23,12 @@ import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
@@ -85,6 +87,9 @@ public class AlterPaimonTableSchemaEventHandler {
}
} else if (event instanceof AlterTableColumnEvent) {
applySingleSchemaChangeEvent(event);
+ } else if (event instanceof AlterTableCommentEvent) {
+ // Table-level comment changes, handle gracefully
+ applySingleSchemaChangeEvent(event);
} else {
throw new UnsupportedOperationException("Unsupported alter table
event: " + event);
}
@@ -128,6 +133,13 @@ public class AlterPaimonTableSchemaEventHandler {
paimonCatalog.alterTable(
identifier, SchemaChange.renameColumn(oldColumn,
column.getName()), false);
}
+ } else if (event instanceof AlterTableCommentEvent
+ || event instanceof AlterColumnCommentEvent) {
+ // Comment-only changes are not supported by Paimon sink, safely
ignore
+ log.info(
+ "Ignoring comment change event for table {} - Paimon sink
does not support comment sync: {}",
+ paimonTablePath.getFullName(),
+ event.getClass().getSimpleName());
} else {
throw new UnsupportedOperationException("Unsupported alter table
event: " + event);
}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/SchemaUtils.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/SchemaUtils.java
index 18df8f4c77..2cb9933438 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/SchemaUtils.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/SchemaUtils.java
@@ -21,10 +21,12 @@ package
org.apache.seatunnel.connectors.seatunnel.starrocks.util;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.api.table.schema.event.AlterColumnCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
@@ -106,6 +108,13 @@ public class SchemaUtils {
return;
}
applySchemaChange(connection, tablePath, dropColumnEvent);
+ } else if (event instanceof AlterTableCommentEvent
+ || event instanceof AlterColumnCommentEvent) {
+ // Comment-only changes are not supported by StarRocks sink,
safely ignore
+ log.info(
+ "Ignoring comment change event for table {} -
StarRocks sink does not support comment sync: {}",
+ tablePath.getFullName(),
+ event.getEventType());
} else {
throw new SeaTunnelException(
"Unsupported schemaChangeEvent : " +
event.getEventType());
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
index a060f17ce5..edca21a634 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
@@ -143,6 +143,16 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
// waiting for case3/case4 completed
assertTableStructureAndData(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE);
+
+ // case5 comment changes with cdc data at same time
+ shopDatabase.setTemplateName("comment_changes").createAndInitialize();
+
+ // waiting for case5 completed – data must continue flowing after
comment DDL
+ assertTableStructureAndData(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE);
+
+ // verify the final source table comment was applied (tests comment
update idempotency)
+ assertSourceTableComment(
+ MYSQL_DATABASE, SOURCE_TABLE, "Updated product catalog with
sports equipment");
}
@Order(2)
@@ -170,6 +180,7 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
() ->
assertTableDataEqualsBySourceColumnOrder(
database, sourceTable, sinkTable,
null));
+ awaitIncrementalStreamingReady(database, sourceTable, sinkTable);
// case1 add columns with cdc data at same time
shopDatabase.setTemplateName("add_columns").createAndInitialize();
@@ -223,6 +234,11 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
// case4 modify column data type with cdc data at same time
assertCaseByDdlName("modify_columns", database, sourceTable,
sinkTable);
+
+ // case5 comment changes with cdc data at same time
+ assertCaseByDdlName("comment_changes", database, sourceTable,
sinkTable);
+ assertSourceTableComment(
+ database, sourceTable, "Updated product catalog with sports
equipment");
}
private void assertCaseByDdlName(
@@ -249,6 +265,7 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
() ->
assertTableDataEqualsBySourceColumnOrder(
database, sourceTable, sinkTable,
null));
+ awaitIncrementalStreamingReady(database, sourceTable, sinkTable);
// case1 add columns with cdc data at same time
shopDatabase.setTemplateName("add_columns").createAndInitialize();
@@ -309,6 +326,36 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
database, sourceTable, sinkTable,
null));
}
+ /**
+ * Waits until the incremental binlog reader is active before issuing
schema changes.
+ *
+ * <p>Snapshot rows can reach the sink before the binlog reader connects.
If DDL is issued in
+ * that gap, Debezium can initialize from the post-DDL table schema and
then replay pre-DDL row
+ * events with fewer columns.
+ */
+ private void awaitIncrementalStreamingReady(
+ String database, String sourceTable, String sinkTable) {
+ String source = quoteIdentifier(database) + "." +
quoteIdentifier(sourceTable);
+ executeUpdate(
+ "insert into "
+ + source
+ + " (id, name, description, weight)"
+ + " values (100, 'streaming-ready', 'streaming-ready',
0)");
+
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ assertTableDataEqualsBySourceColumnOrder(
+ database, sourceTable, sinkTable, "id
= 100"));
+
+ executeUpdate("delete from " + source + " where id = 100");
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ assertTableDataEqualsBySourceColumnOrder(
+ database, sourceTable, sinkTable, "id
= 100"));
+ }
+
private Connection getJdbcConnection() throws SQLException {
return DriverManager.getConnection(
MYSQL_CONTAINER.getJdbcUrl(),
@@ -404,6 +451,40 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
}
}
+ /**
+ * Asserts that the given source table's TABLE_COMMENT in {@code
information_schema.TABLES}
+ * matches {@code expectedComment}. This check is performed directly
against the source MySQL
+ * instance; it is intentionally not compared with the sink because the
JDBC sink treats comment
+ * events as no-ops.
+ */
+ private void assertSourceTableComment(String database, String table,
String expectedComment) {
+ String sql =
+ String.format(
+ "SELECT TABLE_COMMENT FROM information_schema.TABLES"
+ + " WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME =
'%s'",
+ database, table);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery(sql)) {
+ Assertions.assertTrue(rs.next(), "Table " + table + " not found in
information_schema");
+ Assertions.assertEquals(
+ expectedComment,
+ rs.getString("TABLE_COMMENT"),
+ "Source table comment should have been updated by the
ALTER TABLE COMMENT DDL");
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void executeUpdate(String sql) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.executeUpdate(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private List<List<Object>> query(String sql) {
try (Connection connection = getJdbcConnection()) {
ResultSet resultSet =
connection.createStatement().executeQuery(sql);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
index d6bc70bd0e..4e306917ea 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
@@ -219,6 +219,31 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
// waiting for case3/case4 completed
assertTableStructureAndData(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE);
+
+ // savepoint 3
+ Assertions.assertEquals(0,
container.savepointJob(jobId).getExitCode());
+
+ // case5 table comment change with cdc data at same time
+ shopDatabase.setTemplateName("comment_changes").createAndInitialize();
+
+ // restore 3
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.restoreJob(jobConfigFile, jobId);
+ } catch (Exception e) {
+ log.error("Commit task exception :" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ return null;
+ });
+
+ // waiting for case5 completed – data must continue flowing after
comment DDL
+ assertTableStructureAndData(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE);
+
+ // verify the final source table comment was applied (tests comment
update idempotency)
+ assertSourceTableComment(
+ MYSQL_DATABASE, SOURCE_TABLE, "Updated product catalog with
sports equipment");
}
@Order(2)
@@ -476,6 +501,11 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
// case4 modify column data type with cdc data at same time
assertCaseByDdlName("modify_columns", database, sourceTable,
sinkTable);
+
+ // case5 comment changes with cdc data at same time
+ assertCaseByDdlName("comment_changes", database, sourceTable,
sinkTable);
+ assertSourceTableComment(
+ database, sourceTable, "Updated product catalog with sports
equipment");
}
private void assertCaseByDdlName(
@@ -777,6 +807,31 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
}
}
+ /**
+ * Asserts that the given source table's TABLE_COMMENT in {@code
information_schema.TABLES}
+ * matches {@code expectedComment}. This check is performed directly
against the source MySQL
+ * instance; it is intentionally not compared with the sink because the
JDBC sink treats comment
+ * events as no-ops.
+ */
+ private void assertSourceTableComment(String database, String table,
String expectedComment) {
+ String sql =
+ String.format(
+ "SELECT TABLE_COMMENT FROM information_schema.TABLES"
+ + " WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME =
'%s'",
+ database, table);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery(sql)) {
+ Assertions.assertTrue(rs.next(), "Table " + table + " not found in
information_schema");
+ Assertions.assertEquals(
+ expectedComment,
+ rs.getString("TABLE_COMMENT"),
+ "Source table comment should have been updated by the
ALTER TABLE COMMENT DDL");
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private List<List<Object>> query(String sql) {
try (Connection connection = getJdbcConnection()) {
ResultSet resultSet =
connection.createStatement().executeQuery(sql);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/comment_changes.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/comment_changes.sql
new file mode 100644
index 0000000000..9835397c81
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/comment_changes.sql
@@ -0,0 +1,63 @@
+--
+-- 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.
+--
+
+--
----------------------------------------------------------------------------------------------------------------
+-- DATABASE: shop
+-- Table state at this point:
+-- id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY
+-- name LONGTEXT NULL
+-- description VARCHAR(512)
+-- weight FLOAT
+-- add_column INT NOT NULL DEFAULT 1
+--
----------------------------------------------------------------------------------------------------------------
+CREATE DATABASE IF NOT EXISTS `shop`;
+use shop;
+
+-- case5: comment changes; the CDC pipeline must survive comment-related DDL
+-- and continue replicating data normally. The JDBC sink intentionally
+-- ignores comment events, so only data continuity is verified
end-to-end.
+
+-- Step 1: Table comment change (comment-only DDL)
+ALTER TABLE products COMMENT = 'Product catalog table';
+
+delete from products where id < 182;
+insert into products
+values (182,"motorcycle","Sport motorcycle",200.0,1),
+ (183,"bicycle","Mountain bicycle",15.5,2),
+ (184,"skateboard","Street skateboard",3.2,3),
+ (185,"rollerblade","Inline rollerblade",4.5,4),
+ (186,"football","Standard football",0.45,5),
+ (187,"basketball","Standard basketball",0.62,6),
+ (188,"tennis racket","Carbon fiber racket",0.28,7),
+ (189,"badminton racket","Lightweight racket",0.12,8),
+ (190,"ping-pong paddle","Table tennis paddle",0.18,9);
+
+-- Step 2: Column comment embedded in MODIFY COLUMN (comment is part of a
structural change)
+-- This tests that column comments flow through when bundled with
type/nullability changes
+ALTER TABLE products MODIFY COLUMN description VARCHAR(512) COMMENT 'Product
description text';
+
+insert into products
+values (191,"golf club","Steel driver club",0.95,1),
+ (192,"baseball bat","Wooden bat",1.2,2),
+ (193,"cricket bat","Willow cricket bat",1.1,3);
+
+-- Step 3: Update table comment again (verifies comment updates are idempotent)
+ALTER TABLE products COMMENT = 'Updated product catalog with sports equipment';
+
+insert into products
+values (194,"hockey stick","Composite hockey stick",0.6,1),
+ (195,"rugby ball","Official size rugby ball",0.42,2);
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
index 0ce26072b3..44c362a93d 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
@@ -557,11 +557,16 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
return supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN);
case SCHEMA_CHANGE_CHANGE_COLUMN:
return supportedTypes.contains(SchemaChangeType.RENAME_COLUMN);
+ case SCHEMA_CHANGE_ALTER_TABLE_COMMENT:
+ return
supportedTypes.contains(SchemaChangeType.ALTER_TABLE_COMMENT);
+ case SCHEMA_CHANGE_ALTER_COLUMN_COMMENT:
+ return
supportedTypes.contains(SchemaChangeType.ALTER_COLUMN_COMMENT);
case SCHEMA_CHANGE_UPDATE_COLUMNS:
return supportedTypes.contains(SchemaChangeType.ADD_COLUMN)
||
supportedTypes.contains(SchemaChangeType.DROP_COLUMN)
||
supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN)
- ||
supportedTypes.contains(SchemaChangeType.RENAME_COLUMN);
+ ||
supportedTypes.contains(SchemaChangeType.RENAME_COLUMN)
+ ||
supportedTypes.contains(SchemaChangeType.ALTER_COLUMN_COMMENT);
default:
log.error("Unknown schema change event type: {}",
event.getEventType());
throw SchemaValidationException.unsupportedChangeType(
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
index f187355696..552eb727c7 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
@@ -24,6 +24,8 @@ import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.schema.SchemaChangeType;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableCommentEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
import
org.apache.seatunnel.api.table.schema.exception.SchemaCoordinationException;
import
org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionException;
import org.apache.seatunnel.api.table.type.BasicType;
@@ -176,6 +178,28 @@ public class SchemaOperatorTest {
assertTrue(pendingQueue.peek().isSchemaChange);
}
+ @Test
+ void testCommentEventIsSupportedSchemaChange() throws Exception {
+ OperatorTestContext context =
+ createOperator(
+
Collections.singletonList(SchemaChangeType.ALTER_TABLE_COMMENT), false);
+ AlterTableCommentEvent event =
+ AlterTableCommentEvent.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ "old comment",
+ "new comment");
+ SeaTunnelRow row = createDataRow("row-after-comment-change");
+
+ context.operator.processElement(new
StreamRecord<>(createSchemaRow(event), 400L));
+ context.operator.processElement(new StreamRecord<>(row, 401L));
+ context.operator.notifyCheckpointComplete(40L);
+ context.operator.notifyCheckpointComplete(41L);
+
+ assertEquals(2, context.output.records.size());
+ assertSchemaBroadcast(context.output.records.get(0), event);
+ assertEquals(row, context.output.records.get(1).getValue());
+ }
+
/**
* Verifies that {@link SchemaOperator#handleFallbackTimerOnTaskThread()}
correctly respects the
* checkpoint-completion safety fence even when called from a
stall-detection timer.
@@ -252,9 +276,22 @@ public class SchemaOperatorTest {
private static OperatorTestContext createOperator(
OperatorStateStoreStub stateStore, boolean restored) throws
Exception {
+ return createOperator(
+ stateStore,
Collections.singletonList(SchemaChangeType.ADD_COLUMN), restored);
+ }
+
+ private static OperatorTestContext createOperator(
+ List<SchemaChangeType> supportedTypes, boolean restored) throws
Exception {
+ return createOperator(new OperatorStateStoreStub(), supportedTypes,
restored);
+ }
+
+ private static OperatorTestContext createOperator(
+ OperatorStateStoreStub stateStore,
+ List<SchemaChangeType> supportedTypes,
+ boolean restored)
+ throws Exception {
SupportSchemaEvolution source =
Mockito.mock(SupportSchemaEvolution.class);
- Mockito.when(source.supports())
-
.thenReturn(Collections.singletonList(SchemaChangeType.ADD_COLUMN));
+ Mockito.when(source.supports()).thenReturn(supportedTypes);
SchemaOperator operator =
new SchemaOperator(
@@ -299,7 +336,7 @@ public class SchemaOperatorTest {
PhysicalColumn.of("added_col", BasicType.STRING_TYPE, 64L,
true, null, null));
}
- private static SeaTunnelRow createSchemaRow(AlterTableAddColumnEvent
event) {
+ private static SeaTunnelRow createSchemaRow(SchemaChangeEvent event) {
SeaTunnelRow row = new SeaTunnelRow(0);
row.setTableId("__SCHEMA_CHANGE_EVENT__");
Map<String, Object> options = new LinkedHashMap<>();
@@ -316,9 +353,9 @@ public class SchemaOperatorTest {
}
private static void assertSchemaBroadcast(
- StreamRecord<SeaTunnelRow> record, AlterTableAddColumnEvent event)
{
+ StreamRecord<SeaTunnelRow> record, SchemaChangeEvent event) {
Object broadcastEvent =
record.getValue().getOptions().get("schema_change_broadcast");
- assertInstanceOf(AlterTableAddColumnEvent.class, broadcastEvent);
+ assertInstanceOf(event.getClass(), broadcastEvent);
assertEquals(event, broadcastEvent);
}
