minor code simplification

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -39,14 +39,9 @@ func (c *MySqlConnector) GetTableSchema(
 	system protos.TypeSystem,
 	tableMappings []*protos.TableMapping,
 ) (map[string]*protos.TableSchema, error) {
-	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
-	}
-
 	res := make(map[string]*protos.TableSchema, len(tableMappings))
 	for _, tm := range tableMappings {
-		tableSchema, err := c.getTableSchemaForTable(ctx, env, tm, system, binlogRowMetadataSupported, version)
+		tableSchema, err := c.getTableSchemaForTable(ctx, env, tm, system, version)
 		if err != nil {
 			c.logger.Info("error fetching schema", slog.String("table", tm.SourceTableIdentifier), slog.Any("error", err))
 			return nil, err
@@ -63,8 +58,7 @@ func (c *MySqlConnector) getTableSchemaForTable(
 	env map[string]string,
 	tm *protos.TableMapping,
 	system protos.TypeSystem,
-	binlogRowMetadataSupported bool,
-	version uint32,
+	mirrorVersion uint32,
 ) (*protos.TableSchema, error) {
 	qualifiedTable, err := common.ParseTableIdentifier(tm.SourceTableIdentifier)
 	if err != nil {
@@ -93,8 +87,13 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		name       string
 		seqInIndex int64
 	}
-	var primaryEntries []pkEntry
 
+	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
+	}
+
+	var primaryEntries []pkEntry
 	for idx := range rs.RowNumber() {
 		columnName, err := rs.GetString(idx, 0)
 		if err != nil {
@@ -120,7 +119,7 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		if err != nil {
 			return nil, err
 		}
-		qkind, err := QkindFromMysqlColumnType(dataType, binlogRowMetadataSupported, version)
+		qkind, err := QkindFromMysqlColumnType(dataType, binlogRowMetadataSupported, mirrorVersion)
 		if err != nil {
 			return nil, err
 		}
@@ -707,7 +706,7 @@ func (c *MySqlConnector) PullRecords(
 
 func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool shared.CatalogPool,
 	req *model.PullRecordsRequest[model.RecordItems], stmt *ast.AlterTableStmt, stmtSchema string,
-	binlogRowMetadataSupported bool, version uint32,
+	binlogRowMetadataSupported bool, mirrorVersion uint32,
 ) error {
 	// if ALTER TABLE doesn't have database/schema name, use one attached to event
 	var sourceSchemaName string
@@ -754,7 +753,7 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 						slog.String("tableName", sourceTableName))
 				}
 
-				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr(), binlogRowMetadataSupported, version)
+				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr(), binlogRowMetadataSupported, mirrorVersion)
 				if err != nil {
 					return err
 				}

flow/connectors/mysql/qrep.go

@@ -211,13 +211,9 @@ func (c *MySqlConnector) PullQRepRecords(
 	partition *protos.QRepPartition,
 	stream *model.QRecordStream,
 ) (int64, int64, error) {
-	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
-	if err != nil {
-		return 0, 0, fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
-	}
 	tableSchema, err := c.getTableSchemaForTable(ctx, config.Env,
 		&protos.TableMapping{SourceTableIdentifier: config.WatermarkTable}, protos.TypeSystem_Q,
-		binlogRowMetadataSupported, config.Version)
+		config.Version)
 	if err != nil {
 		return 0, 0, fmt.Errorf("failed to get schema for watermark table %s: %w", config.WatermarkTable, err)
 	}