use binlogconsumer

Author: elianddb

sql/analyzer/catalog.go

@@ -34,6 +34,8 @@ type Catalog struct {
 	DbProvider    sql.DatabaseProvider
 	AuthHandler   sql.AuthorizationHandler
 
+	// BinlogConsumer holds an optional consumer that processes binlog events (e.g. for BINLOG statements).
+	BinlogConsumer binlogreplication.BinlogConsumer
 	// BinlogReplicaController holds an optional controller that receives forwarded binlog
 	// replication messages (e.g. "start replica").
 	BinlogReplicaController binlogreplication.BinlogReplicaController
@@ -53,6 +55,7 @@ func (c *Catalog) DropDbStats(ctx *sql.Context, db string, flush bool) error {
 }
 
 var _ sql.Catalog = (*Catalog)(nil)
+var _ binlogreplication.BinlogConsumerCatalog = (*Catalog)(nil)
 var _ binlogreplication.BinlogReplicaCatalog = (*Catalog)(nil)
 var _ binlogreplication.BinlogPrimaryCatalog = (*Catalog)(nil)
 
@@ -76,6 +79,14 @@ func NewCatalog(provider sql.DatabaseProvider) *Catalog {
 	return c
 }
 
+func (c *Catalog) HasBinlogConsumer() bool {
+	return c.BinlogConsumer != nil
+}
+
+func (c *Catalog) GetBinlogConsumer() binlogreplication.BinlogConsumer {
+	return c.BinlogConsumer
+}
+
 func (c *Catalog) HasBinlogReplicaController() bool {
 	return c.BinlogReplicaController != nil
 }

sql/binlogreplication/binlog_replication.go

@@ -25,6 +25,17 @@ import (
 	"github.com/dolthub/vitess/go/mysql"
 )
 
+// BinlogConsumer processes binlog events. This interface can be used by any component that needs to consume
+// and apply binlog events, such as BINLOG statement execution, streaming replication, or other binlog processing.
+type BinlogConsumer interface {
+	// ProcessEvent processes a single binlog event.
+	ProcessEvent(ctx *sql.Context, event mysql.BinlogEvent) error
+
+	// HasFormatDescription returns true if a FORMAT_DESCRIPTION_EVENT has been processed.
+	// This is required before processing TABLE_MAP and row events in BINLOG statements.
+	HasFormatDescription() bool
+}
+
 // BinlogReplicaController allows callers to control a binlog replica. Providers built on go-mysql-server may optionally
 // implement this interface and use it when constructing a SQL engine in order to receive callbacks when replication
 // statements (e.g. START REPLICA, SHOW REPLICA STATUS) are being handled.
@@ -62,12 +73,6 @@ type BinlogReplicaController interface {
 	// error indicating that replication needs to be stopped before it can be reset. If any errors were encountered
 	// resetting the replica state, an error is returned, otherwise nil is returned if the reset was successful.
 	ResetReplica(ctx *sql.Context, resetAll bool) error
-
-	// ConsumeBinlogEvent processes a single parsed binlog event.
-	ConsumeBinlogEvent(ctx *sql.Context, event mysql.BinlogEvent) error
-
-	// HasFormatDescription returns true if a FORMAT_DESCRIPTION_EVENT has been processed.
-	HasFormatDescription() bool
 }
 
 // BinlogPrimaryController allows an integrator to extend GMS with support for operating as a binlog primary server.
@@ -153,6 +158,15 @@ type ReplicaStatus struct {
 	SourceSsl             bool
 }
 
+// BinlogConsumerCatalog extends the Catalog interface and provides methods for accessing a BinlogConsumer
+// for BINLOG statement execution and other binlog event processing.
+type BinlogConsumerCatalog interface {
+	// HasBinlogConsumer returns true if a non-nil BinlogConsumer is available for this catalog.
+	HasBinlogConsumer() bool
+	// GetBinlogConsumer returns the BinlogConsumer registered with this catalog.
+	GetBinlogConsumer() BinlogConsumer
+}
+
 // BinlogReplicaCatalog extends the Catalog interface and provides methods for accessing a BinlogReplicaController
 // for a Catalog.
 type BinlogReplicaCatalog interface {

sql/plan/binlog.go

@@ -28,18 +28,18 @@ const DynamicPrivilege_BinlogAdmin = "binlog_admin"
 // mysqldump, mysqlbinlog, and mariadb-binlog read these binary events from log files and output them as base64-encoded
 // BINLOG statements for replay.
 //
-// The BINLOG statement execution is delegated to the BinlogReplicaController. The base64-encoded event data is decoded
-// and passed to the controller's ConsumeBinlogEvents method for processing. This allows integrators like Dolt to handle
-// BINLOG statement execution using their existing binlog replication infrastructure.
+// The BINLOG statement execution is delegated to the BinlogConsumer. The base64-encoded event data is decoded
+// and passed to the consumer's ProcessEvent method for processing. This allows integrators like Dolt to handle
+// BINLOG statement execution using their binlog event processing infrastructure.
 //
 // See https://dev.mysql.com/doc/refman/8.4/en/binlog.html for the BINLOG statement specification.
 type Binlog struct {
-	Base64Str         string
-	ReplicaController binlogreplication.BinlogReplicaController
+	Base64Str string
+	Consumer  binlogreplication.BinlogConsumer
 }
 
 var _ sql.Node = (*Binlog)(nil)
-var _ BinlogReplicaControllerCommand = (*Binlog)(nil)
+var _ BinlogConsumerCommand = (*Binlog)(nil)
 
 // NewBinlog creates a new Binlog node.
 func NewBinlog(base64Str string) *Binlog {
@@ -48,10 +48,10 @@ func NewBinlog(base64Str string) *Binlog {
 	}
 }
 
-// WithBinlogReplicaController implements the BinlogReplicaControllerCommand interface.
-func (b *Binlog) WithBinlogReplicaController(controller binlogreplication.BinlogReplicaController) sql.Node {
+// WithBinlogConsumer implements the BinlogConsumerCommand interface.
+func (b *Binlog) WithBinlogConsumer(consumer binlogreplication.BinlogConsumer) sql.Node {
 	nc := *b
-	nc.ReplicaController = controller
+	nc.Consumer = consumer
 	return &nc
 }
 

sql/plan/replication_commands.go

@@ -36,6 +36,15 @@ const DynamicPrivilege_ReplicationSlaveAdmin = "replication_slave_admin"
 // See https://dev.mysql.com/doc/refman/8.0/en/privileges-provided.html#priv_replication-applier
 const DynamicPrivilege_ReplicationApplier = "replication_applier"
 
+// BinlogConsumerCommand represents a SQL statement that requires a BinlogConsumer
+// (e.g. BINLOG statement).
+type BinlogConsumerCommand interface {
+	sql.Node
+
+	// WithBinlogConsumer returns a new instance of this command, with the binlog consumer configured.
+	WithBinlogConsumer(consumer binlogreplication.BinlogConsumer) sql.Node
+}
+
 // BinlogReplicaControllerCommand represents a SQL statement that requires a BinlogReplicaController
 // (e.g. Start Replica, Show Replica Status).
 type BinlogReplicaControllerCommand interface {

sql/planbuilder/builder.go

@@ -407,8 +407,8 @@ func (b *Builder) buildSubquery(inScope *scope, stmt ast.Statement, subQuery str
 		}
 		outScope = inScope.push()
 		binlogNode := plan.NewBinlog(n.Base64Str)
-		if binCat, ok := b.cat.(binlogreplication.BinlogReplicaCatalog); ok && binCat.HasBinlogReplicaController() {
-			binlogNode = binlogNode.WithBinlogReplicaController(binCat.GetBinlogReplicaController()).(*plan.Binlog)
+		if binCat, ok := b.cat.(binlogreplication.BinlogConsumerCatalog); ok && binCat.HasBinlogConsumer() {
+			binlogNode = binlogNode.WithBinlogConsumer(binCat.GetBinlogConsumer()).(*plan.Binlog)
 		}
 		outScope.node = binlogNode
 	}

sql/rowexec/binlog.go

@@ -39,8 +39,8 @@ import (
 //
 // See https://dev.mysql.com/doc/refman/8.4/en/binlog.html for the BINLOG statement specification.
 func (b *BaseBuilder) buildBinlog(ctx *sql.Context, n *plan.Binlog, row sql.Row) (sql.RowIter, error) {
-	if n.ReplicaController == nil {
-		return nil, fmt.Errorf("BINLOG statement requires BinlogReplicaController")
+	if n.Consumer == nil {
+		return nil, fmt.Errorf("BINLOG statement requires BinlogConsumer")
 	}
 
 	var decoded []byte
@@ -60,16 +60,16 @@ func (b *BaseBuilder) buildBinlog(ctx *sql.Context, n *plan.Binlog, row sql.Row)
 	}
 
 	return &binlogIter{
-		controller: n.ReplicaController,
-		decoded:    decoded,
+		consumer: n.Consumer,
+		decoded:  decoded,
 	}, nil
 }
 
 // binlogIter processes decoded binlog events one at a time, returning an OkResult when all events are processed.
 type binlogIter struct {
-	controller binlogreplication.BinlogReplicaController
-	decoded    []byte
-	offset     int
+	consumer binlogreplication.BinlogConsumer
+	decoded  []byte
+	offset   int
 }
 
 var _ sql.RowIter = (*binlogIter)(nil)
@@ -119,13 +119,13 @@ func (bi *binlogIter) Next(ctx *sql.Context) (sql.Row, error) {
 
 	// Check that TABLE_MAP and row events have a FORMAT_DESCRIPTION first
 	if event.IsTableMap() || event.IsWriteRows() || event.IsUpdateRows() || event.IsDeleteRows() {
-		if !bi.controller.HasFormatDescription() {
+		if !bi.consumer.HasFormatDescription() {
 			return nil, sql.ErrNoFormatDescriptionEventBeforeBinlogStatement.New(event.TypeName())
 		}
 	}
 
-	// Process this event using the controller
-	err := bi.controller.ConsumeBinlogEvent(ctx, event)
+	// Process this event using the consumer
+	err := bi.consumer.ProcessEvent(ctx, event)
 	if err != nil {
 		return nil, err
 	}

sql/rowexec/binlog_test.go

@@ -21,7 +21,6 @@ import (
 	"testing"
 
 	"github.com/dolthub/go-mysql-server/sql"
-	"github.com/dolthub/go-mysql-server/sql/binlogreplication"
 	"github.com/dolthub/go-mysql-server/sql/plan"
 	"github.com/dolthub/go-mysql-server/sql/types"
 	"github.com/dolthub/vitess/go/mysql"
@@ -36,7 +35,7 @@ func TestBuildBinlog_InvalidBase64(t *testing.T) {
 
 	_, err := builder.buildBinlog(ctx, binlogNode, nil)
 	require.Error(t, err)
-	require.Contains(t, err.Error(), "BinlogReplicaController")
+	require.Contains(t, err.Error(), "BinlogConsumer")
 }
 
 func TestBuildBinlog_NoBinlogReplicaController(t *testing.T) {
@@ -52,57 +51,33 @@ func TestBuildBinlog_NoBinlogReplicaController(t *testing.T) {
 
 	_, err := builder.buildBinlog(ctx, binlogNode, nil)
 	require.Error(t, err)
-	require.Contains(t, err.Error(), "BinlogReplicaController")
+	require.Contains(t, err.Error(), "BinlogConsumer")
 }
 
-// mockBinlogReplicaController is a test implementation of BinlogReplicaController
-type mockBinlogReplicaController struct {
+// mockBinlogConsumer is a test implementation of BinlogConsumer
+type mockBinlogConsumer struct {
 	consumedEvents []mysql.BinlogEvent
 	returnError    error
 	hasFormatDesc  bool
 }
 
-func (m *mockBinlogReplicaController) ConsumeBinlogEvent(ctx *sql.Context, event mysql.BinlogEvent) error {
+func (m *mockBinlogConsumer) ProcessEvent(ctx *sql.Context, event mysql.BinlogEvent) error {
 	m.consumedEvents = append(m.consumedEvents, event)
 	if event.IsFormatDescription() {
 		m.hasFormatDesc = true
 	}
 	return m.returnError
 }
 
-func (m *mockBinlogReplicaController) HasFormatDescription() bool {
+func (m *mockBinlogConsumer) HasFormatDescription() bool {
 	return m.hasFormatDesc
 }
 
-func (m *mockBinlogReplicaController) StartReplica(ctx *sql.Context) error {
-	return nil
-}
-
-func (m *mockBinlogReplicaController) StopReplica(ctx *sql.Context) error {
-	return nil
-}
-
-func (m *mockBinlogReplicaController) SetReplicationSourceOptions(ctx *sql.Context, options []binlogreplication.ReplicationOption) error {
-	return nil
-}
-
-func (m *mockBinlogReplicaController) SetReplicationFilterOptions(ctx *sql.Context, options []binlogreplication.ReplicationOption) error {
-	return nil
-}
-
-func (m *mockBinlogReplicaController) GetReplicaStatus(ctx *sql.Context) (*binlogreplication.ReplicaStatus, error) {
-	return nil, nil
-}
-
-func (m *mockBinlogReplicaController) ResetReplica(ctx *sql.Context, resetAll bool) error {
-	return nil
-}
-
 func TestBuildBinlog_WithBinlogReplicaController(t *testing.T) {
 	builder := &BaseBuilder{}
 	ctx := sql.NewEmptyContext()
 
-	mockController := &mockBinlogReplicaController{}
+	mockConsumer := &mockBinlogConsumer{}
 
 	// Create a minimal valid binlog event (FORMAT_DESCRIPTION_EVENT)
 	// Event header: timestamp(4) + type(1) + server_id(4) + event_length(4) + next_position(4) + flags(2)
@@ -112,7 +87,7 @@ func TestBuildBinlog_WithBinlogReplicaController(t *testing.T) {
 
 	encoded := base64.StdEncoding.EncodeToString(eventData)
 
-	binlogNode := plan.NewBinlog(encoded).WithBinlogReplicaController(mockController).(*plan.Binlog)
+	binlogNode := plan.NewBinlog(encoded).WithBinlogConsumer(mockConsumer).(*plan.Binlog)
 
 	iter, err := builder.buildBinlog(ctx, binlogNode, nil)
 	require.NoError(t, err)
@@ -124,7 +99,7 @@ func TestBuildBinlog_WithBinlogReplicaController(t *testing.T) {
 	require.Equal(t, types.OkResult{}, row[0])
 
 	// Verify controller received one event
-	require.Len(t, mockController.consumedEvents, 1)
+	require.Len(t, mockConsumer.consumedEvents, 1)
 
 	// Next call should return EOF
 	_, err = iter.Next(ctx)
@@ -135,7 +110,7 @@ func TestBuildBinlog_MultilineBase64WithController(t *testing.T) {
 	builder := &BaseBuilder{}
 	ctx := sql.NewEmptyContext()
 
-	mockController := &mockBinlogReplicaController{}
+	mockConsumer := &mockBinlogConsumer{}
 
 	// Create two minimal events
 	event1 := make([]byte, 19)
@@ -151,7 +126,7 @@ func TestBuildBinlog_MultilineBase64WithController(t *testing.T) {
 	part2 := base64.StdEncoding.EncodeToString(combined[10:])
 	multiline := part1 + "\n" + part2
 
-	binlogNode := plan.NewBinlog(multiline).WithBinlogReplicaController(mockController).(*plan.Binlog)
+	binlogNode := plan.NewBinlog(multiline).WithBinlogConsumer(mockConsumer).(*plan.Binlog)
 
 	iter, err := builder.buildBinlog(ctx, binlogNode, nil)
 	require.NoError(t, err)
@@ -162,7 +137,7 @@ func TestBuildBinlog_MultilineBase64WithController(t *testing.T) {
 	require.NotNil(t, row)
 	require.Equal(t, types.OkResult{}, row[0])
 
-	require.Len(t, mockController.consumedEvents, 2)
+	require.Len(t, mockConsumer.consumedEvents, 2)
 
 	_, err = iter.Next(ctx)
 	require.Equal(t, io.EOF, err)
@@ -172,7 +147,7 @@ func TestBuildBinlog_ControllerError(t *testing.T) {
 	builder := &BaseBuilder{}
 	ctx := sql.NewEmptyContext()
 
-	mockController := &mockBinlogReplicaController{
+	mockConsumer := &mockBinlogConsumer{
 		returnError: sql.ErrUnsupportedFeature.New("test error"),
 	}
 
@@ -181,7 +156,7 @@ func TestBuildBinlog_ControllerError(t *testing.T) {
 	binary.LittleEndian.PutUint32(eventData[9:13], 19)
 	encoded := base64.StdEncoding.EncodeToString(eventData)
 
-	binlogNode := plan.NewBinlog(encoded).WithBinlogReplicaController(mockController).(*plan.Binlog)
+	binlogNode := plan.NewBinlog(encoded).WithBinlogConsumer(mockConsumer).(*plan.Binlog)
 
 	iter, err := builder.buildBinlog(ctx, binlogNode, nil)
 	require.NoError(t, err)