observe mysql/postgres column type changes (#4488)

add metric to count column type change events for MySQL and postgres

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	_ "github.com/pingcap/tidb/pkg/types/parser_driver"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/text/encoding"
 	"google.golang.org/protobuf/proto"
@@ -590,7 +591,8 @@ func (c *MySqlConnector) PullRecords(
 			for _, stmt := range stmts {
 				if alterTableStmt, ok := stmt.(*ast.AlterTableStmt); ok {
 					if err := c.processAlterTableQuery(
-						ctx, catalogPool, req, alterTableStmt, string(ev.Schema), binlogRowMetadataSupported, req.InternalVersion); err != nil {
+						ctx, catalogPool, otelManager, req, alterTableStmt,
+						string(ev.Schema), binlogRowMetadataSupported, req.InternalVersion); err != nil {
 						return fmt.Errorf("failed to process ALTER TABLE query: %w", err)
 					}
 				}
@@ -669,7 +671,7 @@ func (c *MySqlConnector) PullRecords(
 				if ev.Table.ColumnName != nil {
 					var err error
 					fields, err = c.processTableMapEventSchema(
-						ctx, catalogPool, req, ev.Table,
+						ctx, catalogPool, otelManager, req, ev.Table,
 						sourceTableName, destinationTableName, schema, exclusion,
 					)
 					if err != nil {
@@ -828,6 +830,7 @@ func (c *MySqlConnector) PullRecords(
 }
 
 func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool shared.CatalogPool,
+	otelManager *otel_metrics.OtelManager,
 	req *model.PullRecordsRequest[model.RecordItems], stmt *ast.AlterTableStmt, stmtSchema string,
 	binlogRowMetadataSupported bool, mirrorVersion uint32,
 ) error {
@@ -855,6 +858,13 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 		NullableEnabled: currentSchema != nil && currentSchema.NullableEnabled,
 	}
 
+	existingColTypes := make(map[string]string)
+	if currentSchema != nil {
+		for _, col := range currentSchema.Columns {
+			existingColTypes[col.Name] = col.Type
+		}
+	}
+
 	hasPositionShiftingDdlChanges := false
 
 	for _, spec := range stmt.Specs {
@@ -869,18 +879,28 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 					continue
 				}
 
+				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr(), binlogRowMetadataSupported, mirrorVersion)
+				if err != nil {
+					return err
+				}
+
+				if oldType, exists := existingColTypes[col.Name.OrigColName()]; exists && oldType != string(qkind) {
+					c.logger.Warn("column type change detected via ALTER TABLE, not propagating",
+						slog.String("table", sourceTableName),
+						slog.String("column", col.Name.OrigColName()),
+						slog.String("from", oldType),
+						slog.String("to", string(qkind)))
+					c.recordColumnTypeChange(ctx, otelManager, types.QValueKind(oldType), qkind,
+						otel_metrics.SourceEventTypeDDL)
+				}
+
 				if spec.Position != nil && spec.Position.Tp != ast.ColumnPositionNone {
 					hasPositionShiftingDdlChanges = true
 					c.logger.Warn("column added with position specifier (FIRST/AFTER)",
 						slog.String("columnName", col.Name.String()),
 						slog.String("tableName", sourceTableName))
 				}
 
-				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr(), binlogRowMetadataSupported, mirrorVersion)
-				if err != nil {
-					return err
-				}
-
 				nullable := true
 				for _, option := range col.Options {
 					if option.Tp == ast.ColumnOptionNotNull {
@@ -956,12 +976,25 @@ func parseIncidentEvent(data []byte) (uint16, string) {
 	return incident, string(data[3:end])
 }
 
+func (c *MySqlConnector) recordColumnTypeChange(
+	ctx context.Context, otelManager *otel_metrics.OtelManager, from types.QValueKind, to types.QValueKind,
+	eventType string,
+) {
+	otelManager.Metrics.ColumnTypeChangesCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
+		attribute.String(otel_metrics.SourcePeerType, "mysql"),
+		attribute.String(otel_metrics.TypeChangeFromKey, string(from)),
+		attribute.String(otel_metrics.TypeChangeToKey, string(to)),
+		attribute.String(otel_metrics.SourceEventTypeKey, eventType),
+	)))
+}
+
 // processTableMapEventSchema compares the TABLE_MAP_EVENT schema against the cached schema
 // and returns a TableSchemaDelta if new columns are detected (e.g., after gh-ost migration).
 // It also returns a slice mapping binlog column index to FieldDescription for efficient row processing.
 func (c *MySqlConnector) processTableMapEventSchema(
 	ctx context.Context,
 	catalogPool shared.CatalogPool,
+	otelManager *otel_metrics.OtelManager,
 	req *model.PullRecordsRequest[model.RecordItems],
 	tableMap *replication.TableMapEvent,
 	sourceTableName string,
@@ -989,14 +1022,26 @@ func (c *MySqlConnector) processTableMapEventSchema(
 			continue
 		}
 
+		var charset uint16
+		if collation, ok := collationMap[idx]; ok {
+			charset = uint16(collation)
+		}
+
 		if fd, exists := existingCols[colName]; exists {
 			newFds[idx] = fd
+
+			if qkind, err := qkindFromMysqlType(
+				tableMap.ColumnType[idx], unsignedMap[idx], charset, req.InternalVersion,
+			); err == nil && qkind != types.QValueKind(fd.Type) {
+				c.logger.Warn("column type change detected from TABLE_MAP_EVENT, not propagating",
+					slog.String("table", sourceTableName),
+					slog.String("column", colName),
+					slog.String("from", fd.Type),
+					slog.String("to", string(qkind)))
+				c.recordColumnTypeChange(ctx, otelManager, types.QValueKind(fd.Type), qkind, otel_metrics.SourceEventTypeEventMetadata)
+			}
 		} else {
 			// New column detected - get type from TABLE_MAP_EVENT
-			var charset uint16
-			if collation, ok := collationMap[idx]; ok {
-				charset = uint16(collation)
-			}
 			mytype := tableMap.ColumnType[idx]
 			qkind, err := qkindFromMysqlType(mytype, unsignedMap[idx], charset, req.InternalVersion)
 			if err != nil {

flow/connectors/postgres/cdc.go

@@ -1305,6 +1305,12 @@ func processRelationMessage[Items model.Items](
 		} else if prevRelMap[column.Name] != currRelMap[column.Name] {
 			p.logger.Warn(fmt.Sprintf("Detected column %s with type changed from %s to %s in table %s, but not propagating",
 				column.Name, prevRelMap[column.Name], currRelMap[column.Name], schemaDelta.SrcTableName))
+			p.otelManager.Metrics.ColumnTypeChangesCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
+				attribute.String(otel_metrics.SourcePeerType, "postgres"),
+				attribute.String(otel_metrics.TypeChangeFromKey, prevRelMap[column.Name]),
+				attribute.String(otel_metrics.TypeChangeToKey, currRelMap[column.Name]),
+				attribute.String(otel_metrics.SourceEventTypeKey, otel_metrics.SourceEventTypeEventMetadata),
+			)))
 		}
 	}
 	for _, column := range prevSchema.Columns {

flow/otel_metrics/attributes.go

@@ -46,6 +46,9 @@ const (
 	EndBatchIDKey              = "endBatchID"
 	ResumeTokenKey             = "resumeToken"
 	GtidKey                    = "gtid"
+	TypeChangeFromKey          = "from"
+	TypeChangeToKey            = "to"
+	SourceEventTypeKey         = "sourceEventType"
 )
 
 const (
@@ -63,3 +66,8 @@ const (
 	InstanceStatusUnknown     = "unknown"
 	InstanceStatusReady       = "ready"
 )
+
+const (
+	SourceEventTypeDDL           = "ddl"
+	SourceEventTypeEventMetadata = "eventMetadata"
+)

flow/otel_metrics/otel_manager.go

@@ -81,6 +81,7 @@ const (
 	CodeNotificationCounterName          = "code_notification"
 	ServerWalEndLagGaugeName             = "wal_end_lag"
 	UsedMySQLCharsetsName                = "used_mysql_charsets"
+	ColumnTypeChangesName                = "column_type_changes"
 )
 
 type Metrics struct {
@@ -136,6 +137,7 @@ type Metrics struct {
 	UnchangedToastValuesCounter      metric.Int64Counter
 	ServerWalEndLagGauge             metric.Int64Gauge
 	UsedMySQLCharsetsCounter         metric.Int64Counter
+	ColumnTypeChangesCounter         metric.Int64Counter
 }
 
 type SlotMetricGauges struct {
@@ -582,6 +584,14 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error {
 		return err
 	}
 
+	if om.Metrics.ColumnTypeChangesCounter, err = om.GetOrInitInt64Counter(BuildMetricName(ColumnTypeChangesName),
+		metric.WithDescription(
+			"Counter of column type changes detected on the CDC path, with `source` label holding the source peer type, "+
+				"`from`/`to` labels holding the source/target type and `sourceEventType` holding the source of event(ddl, eventMetadata)"),
+	); err != nil {
+		return err
+	}
+
 	if CodeNotificationCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CodeNotificationCounterName),
 		metric.WithDescription("One-off notifications with unique `message` attribute, triggers generic non-paging alert"),
 	); err != nil {