maria: classify compressed columns errors + add validation for create mirror

flow/alerting/classifier.go

@@ -152,6 +152,9 @@ var (
 	ErrorNotifyBinlogPartialJsonUnsupported = ErrorClass{
 		Class: "NOTIFY_BINLOG_PARTIAL_JSON_UNSUPPORTED", action: NotifyUser,
 	}
+	ErrorNotifyMySQLCompressedColumnUnsupported = ErrorClass{
+		Class: "NOTIFY_MYSQL_COMPRESSED_COLUMN_UNSUPPORTED", action: NotifyUser,
+	}
 	ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser,
 	}
@@ -1136,6 +1139,16 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if compressedColumnError, ok := errors.AsType[*exceptions.MySQLUnsupportedCompressedColumnError](err); ok {
+		return ErrorNotifyMySQLCompressedColumnUnsupported, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_COMPRESSED_COLUMN",
+			AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+				ErrorAttributeKeyTable: fmt.Sprintf("%s.%s", compressedColumnError.SchemaName, compressedColumnError.TableName),
+			},
+		}
+	}
+
 	if unsupportedDDLError, ok := errors.AsType[*exceptions.MySQLUnsupportedDDLError](err); ok {
 		return ErrorNotifyBinlogRowMetadataInvalid, ErrorInfo{
 			Source: ErrorSourceMySQL,

flow/alerting/classifier_test.go

@@ -1285,3 +1285,18 @@ func TestMySQLUnsupportedBinlogRowValueOptionsErrorShouldBeNotifyBinlogPartialJs
 		},
 	}, errInfo)
 }
+
+func TestMySQLUnsupportedCompressedColumnErrorShouldBeNotifyUser(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewMySQLUnsupportedCompressedColumnError("mydb", "mytable", []string{"v"})
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err))
+	assert.Equal(t, ErrorNotifyMySQLCompressedColumnUnsupported, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_COMPRESSED_COLUMN",
+		AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+			ErrorAttributeKeyTable: "mydb.mytable",
+		},
+	}, errInfo)
+}

flow/connectors/mysql/cdc.go

@@ -732,7 +732,13 @@ func (c *MySqlConnector) PullRecords(
 				}
 			}
 		case *replication.TableMapEvent:
-			tableIdToName[ev.TableID] = string(ev.Schema) + "." + string(ev.Table)
+			schemaTable := string(ev.Schema) + "." + string(ev.Table)
+			tableIdToName[ev.TableID] = schemaTable
+			if req.TableNameSchemaMapping[req.TableNameMapping[schemaTable].Name] != nil {
+				if err := checkTableMapForCompressedColumns(ev); err != nil {
+					return err
+				}
+			}
 		case *replication.RowsEvent:
 			sourceTableName := string(ev.Table.Schema) + "." + string(ev.Table.Table) // TODO this is fragile
 			destinationTableName := req.TableNameMapping[sourceTableName].Name
@@ -1018,6 +1024,31 @@ func (c *MySqlConnector) PullRecords(
 	return nil
 }
 
+// MariaDB COMPRESSED columns binlog as these wire types
+// which go-mysql cannot decode.
+const (
+	mysqlColumnTypeBlobCompressed    = 140
+	mysqlColumnTypeVarcharCompressed = 141
+)
+
+func checkTableMapForCompressedColumns(ev *replication.TableMapEvent) error {
+	var compressed []string
+	for i, t := range ev.ColumnType {
+		if t != mysqlColumnTypeVarcharCompressed && t != mysqlColumnTypeBlobCompressed {
+			continue
+		}
+		name := fmt.Sprintf("column #%d", i)
+		if i < len(ev.ColumnName) && len(ev.ColumnName[i]) > 0 {
+			name = string(ev.ColumnName[i])
+		}
+		compressed = append(compressed, name)
+	}
+	if len(compressed) > 0 {
+		return exceptions.NewMySQLUnsupportedCompressedColumnError(string(ev.Schema), string(ev.Table), compressed)
+	}
+	return nil
+}
+
 func fieldDescriptionFromMysqlColumn(
 	col *ast.ColumnDef, binlogRowMetadataSupported bool, mirrorVersion uint32,
 ) (*protos.FieldDescription, error) {

flow/connectors/mysql/cdc_test.go

@@ -27,9 +27,43 @@ import (
 	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
+func TestCheckTableMapForCompressedColumns(t *testing.T) {
+	// id (normal), v (varchar compressed, 140), b (blob compressed, 141)
+	ev := &replication.TableMapEvent{
+		Schema:     []byte("db"),
+		Table:      []byte("t"),
+		ColumnType: []byte{0x03 /* MYSQL_TYPE_LONG */, mysqlColumnTypeVarcharCompressed, mysqlColumnTypeBlobCompressed},
+		ColumnName: [][]byte{[]byte("id"), []byte("v"), []byte("b")},
+	}
+
+	t.Run("reports all compressed columns regardless of exclusion", func(t *testing.T) {
+		err := checkTableMapForCompressedColumns(ev)
+		require.ErrorContains(t, err, "COMPRESSED")
+		var ccErr *exceptions.MySQLUnsupportedCompressedColumnError
+		require.ErrorAs(t, err, &ccErr)
+		require.Equal(t, []string{"v", "b"}, ccErr.Columns)
+	})
+
+	t.Run("no compressed columns yields no error", func(t *testing.T) {
+		normal := &replication.TableMapEvent{ColumnType: []byte{0x03}}
+		require.NoError(t, checkTableMapForCompressedColumns(normal))
+	})
+
+	t.Run("without column names the column is reported positionally", func(t *testing.T) {
+		noMeta := &replication.TableMapEvent{
+			Schema:     []byte("db"),
+			Table:      []byte("t"),
+			ColumnType: []byte{mysqlColumnTypeVarcharCompressed},
+		}
+		err := checkTableMapForCompressedColumns(noMeta)
+		require.ErrorContains(t, err, "column #0")
+	})
+}
+
 func startBinlogStream(t *testing.T, ctx context.Context, c *MySqlConnector) *replication.BinlogStreamer {
 	t.Helper()
 

flow/connectors/mysql/qvalue_convert_test.go

@@ -31,6 +31,19 @@ func TestQkindFromMysqlColumnTypeMariaDB(t *testing.T) {
 	}
 }
 
+func TestQkindFromMysqlColumnTypeCompressed(t *testing.T) {
+	for _, ct := range []string{
+		"varchar(100) /*M!100301 COMPRESSED*/",
+		"text /*M!100301 COMPRESSED*/",
+		"blob /*M!100301 COMPRESSED*/",
+	} {
+		t.Run(ct, func(t *testing.T) {
+			_, err := QkindFromMysqlColumnType(ct, true, shared.InternalVersion_Latest)
+			require.ErrorContains(t, err, "COMPRESSED")
+		})
+	}
+}
+
 func TestQkindFromMysqlType_Bit(t *testing.T) {
 	for _, tc := range []struct {
 		name    string

flow/connectors/mysql/type_conversion.go

@@ -5,11 +5,15 @@ import (
 	"strings"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 func QkindFromMysqlColumnType(ct string, binlogRowMetadataSupported bool, version uint32) (types.QValueKind, error) {
+	if mysql_validation.IsCompressedColumnType(ct) {
+		return types.QValueKindInvalid, fmt.Errorf("MariaDB COMPRESSED columns are not supported: %s", ct)
+	}
 	// https://mariadb.com/docs/server/reference/data-types/date-and-time-data-types/timestamp#tab-current-1
 	ct, _ = strings.CutSuffix(ct, " /* mariadb-5.3 */")
 	ct, _ = strings.CutSuffix(ct, " zerofill")

flow/connectors/mysql/validate.go

@@ -124,6 +124,15 @@ func (c *MySqlConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F
 		return fmt.Errorf("check log replica updates error: %w", err)
 	}
 
+	// maria compressed columns aren't supported yet
+	if c.config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
+		for _, table := range sourceTables {
+			if err := mysql_validation.CheckCompressedColumns(conn, table.Namespace, table.Table); err != nil {
+				return err
+			}
+		}
+	}
+
 	requireRowMetadata := false
 	for _, tm := range cfg.TableMappings {
 		if len(tm.Exclude) > 0 {

flow/e2e/clickhouse_mysql_test.go

@@ -122,6 +122,151 @@ func (s ClickHouseSuite) Test_MySQL_MyISAM_NonTransactional() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_MySQL_MariaDB_CompressedColumn_Rejected() {
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("only applies to mysql")
+	}
+	if mySource.Config.Flavor != protos.MySqlFlavor_MYSQL_MARIA {
+		s.t.Skip("column compression is a MariaDB-only feature")
+	}
+
+	srcTableName := "test_compressed"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`CREATE TABLE %s (
+		id int primary key,
+		val text COMPRESSED,
+		plain text
+	)`, srcFullName)))
+
+	mapping := []*protos.TableMapping{{
+		SourceTableIdentifier:      srcFullName,
+		DestinationTableIdentifier: srcTableName,
+	}}
+
+	// The compressed column is rejected with an actionable error.
+	err := mySource.ValidateMirrorSource(s.t.Context(), &protos.FlowConnectionConfigsCore{
+		TableMappings:     mapping,
+		DoInitialSnapshot: true,
+	})
+	require.ErrorContains(s.t, err, "COMPRESSED")
+
+	// Excluding the column does not help: go-mysql cannot decode it from the binlog, so CDC would
+	// fail anyway. Validation rejects it regardless of exclusion.
+	mapping[0].Exclude = []string{"val"}
+	err = mySource.ValidateMirrorSource(s.t.Context(), &protos.FlowConnectionConfigsCore{
+		TableMappings:     mapping,
+		DoInitialSnapshot: true,
+	})
+	require.ErrorContains(s.t, err, "COMPRESSED")
+}
+
+func (s ClickHouseSuite) Test_MariaDB_CompressedColumn_AddedMidCDC() {
+	if s.cluster {
+		s.t.Skip("source-side compressed column coverage does not need to run against ClickHouse cluster")
+	}
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("only applies to mysql")
+	}
+	if mySource.Config.Flavor != protos.MySqlFlavor_MYSQL_MARIA {
+		s.t.Skip("column compression is a MariaDB-only feature")
+	}
+
+	// A compressed column produces a valid MariaDB binlog event that go-mysql cannot decode.
+	// Because every CDC reader consumes the server-wide binlog before table filtering, the event
+	// can terminate unrelated mirrors using the shared CI server. Use a dedicated throwaway
+	// MariaDB to isolate the unsupported event.
+	src, suffix := SetupMySQLTestContainerSource(s.t, "cmpcol", MySQLTestContainerConfig{
+		Image:                "mariadb:12.3",
+		Flavor:               protos.MySqlFlavor_MYSQL_MARIA,
+		ReplicationMechanism: mySource.Config.ReplicationMechanism,
+	})
+
+	srcTableName := "compressed_mid_cdc"
+	srcFullName := fmt.Sprintf("e2e_test_%s.%s", suffix, srcTableName)
+	dstTableName := "compressed_mid_cdc"
+
+	// The table starts without a compressed column, so mirror creation validation passes.
+	require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf(
+		`CREATE TABLE %s (id INT PRIMARY KEY, name TEXT)`, srcFullName)))
+	require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (id, name) VALUES (1, 'snapshot')`, srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      "test_mariadb_compressed_mid_cdc_" + suffix,
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	// The suite source is the shared CI MariaDB; point the mirror at our isolated source instead.
+	flowConnConfig.SourceName = src.GeneratePeer(s.t).Name
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	catalogPool, err := internal.GetCatalogConnectionPoolFromEnv(s.t.Context())
+	require.NoError(s.t, err)
+	s.t.Cleanup(func() {
+		if !s.t.Failed() {
+			return
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		rows, err := catalogPool.Query(ctx, `
+			SELECT error_type, error_message
+			FROM peerdb_stats.flow_errors
+			WHERE flow_name = $1
+			ORDER BY id`, flowConnConfig.FlowJobName)
+		if err != nil {
+			s.t.Log("failed to query flow logs after test failure:", err)
+			return
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var errorType, message string
+			if err := rows.Scan(&errorType, &message); err != nil {
+				s.t.Log("failed to scan flow log after test failure:", err)
+				return
+			}
+			s.t.Logf("flow log after test failure: type=%s message=%s", errorType, message)
+		}
+		if err := rows.Err(); err != nil {
+			s.t.Log("failed while reading flow logs after test failure:", err)
+		}
+	})
+
+	EnvWaitForCount(env, s, "waiting on initial snapshot", dstTableName, "id,name", 1)
+	// Seeing the snapshot row at the destination does not guarantee that the CDC activity has
+	// started. Prove that CDC is consuming this source before introducing the unsupported type.
+	require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (id, name) VALUES (2, 'cdc-ready')`, srcFullName)))
+	EnvWaitForCount(env, s, "waiting for CDC to start", dstTableName, "id,name", 2)
+
+	// Add a COMPRESSED column mid-stream, then write a row so a row event carrying the undecodable
+	// wire type reaches the pipe and trips the TABLE_MAP detection.
+	require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf(
+		`ALTER TABLE %s ADD COLUMN val TEXT COMPRESSED`, srcFullName)))
+	require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (id, name, val) VALUES (3, 'cdc', 'compressed value')`, srcFullName)))
+
+	EnvWaitFor(s.t, env, 3*time.Minute, "waiting for compressed column error", func() bool {
+		count, err := GetLogCount(s.t.Context(), catalogPool, flowConnConfig.FlowJobName, "error", "cannot be replicated via CDC")
+		if err != nil {
+			s.t.Log("Error querying flow_errors:", err)
+			return false
+		}
+		return count > 0
+	})
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_MySQL_Time() {
 	if _, ok := s.source.(*MySqlSource); !ok {
 		s.t.Skip("only applies to mysql")

flow/pkg/mysql/validation.go

@@ -299,6 +299,42 @@ func IsVitess(conn *client.Conn) (bool, error) {
 	return true, nil // is a Vitess server
 }
 
+func IsCompressedColumnType(columnType string) bool {
+	return strings.Contains(strings.ToUpper(columnType), "COMPRESSED")
+}
+
+func CheckCompressedColumns(conn *client.Conn, schema string, table string) error {
+	rs, err := conn.Execute(fmt.Sprintf(
+		"SELECT column_name, column_type FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'",
+		mysql.Escape(schema), mysql.Escape(table)))
+	if err != nil {
+		return fmt.Errorf("failed to fetch column types for %s.%s: %w", schema, table, err)
+	}
+
+	var compressed []string
+	for idx := range rs.RowNumber() {
+		columnType, err := rs.GetString(idx, 1)
+		if err != nil {
+			return err
+		}
+		if IsCompressedColumnType(columnType) {
+			columnName, err := rs.GetString(idx, 0)
+			if err != nil {
+				return err
+			}
+			compressed = append(compressed, columnName)
+		}
+	}
+
+	if len(compressed) > 0 {
+		return fmt.Errorf(
+			"table %s.%s has MariaDB COMPRESSED column(s) [%s], which cannot be replicated via CDC; "+
+				"convert them to a non-compressed type or remove the table from the mirror",
+			schema, table, strings.Join(compressed, ", "))
+	}
+	return nil
+}
+
 func HasReplicaWithServerId(conn *client.Conn, serverID uint32) (bool, error) {
 	// This check assumes that the number of replicas is not beyond the order of magnitude of thousands.
 	// This is a reasonable assumption given that services like RDS limit them to 15 and that