amend to use replica controller

Author: elianddb

enginetest/memory_engine_test.go

@@ -607,10 +607,6 @@ func TestLoadDataFailing(t *testing.T) {
 	enginetest.TestLoadDataFailing(t, enginetest.NewDefaultMemoryHarness())
 }
 
-func TestBinlog(t *testing.T) {
-	enginetest.TestBinlog(t, enginetest.NewDefaultMemoryHarness())
-}
-
 func TestSelectIntoFile(t *testing.T) {
 	enginetest.TestSelectIntoFile(t, enginetest.NewDefaultMemoryHarness())
 }

sql/binlogreplication/binlog_replication.go

@@ -27,7 +27,8 @@ import (
 
 // 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.
+// statements (e.g. START REPLICA, SHOW REPLICA STATUS) are being handled. This interface also extends BinlogConsumer
+// to allow the same infrastructure to handle BINLOG statement execution.
 type BinlogReplicaController interface {
 	// StartReplica tells the binlog replica controller to start up replication processes for the current replication
 	// configuration. An error is returned if replication was unable to be started. Note the error response only signals
@@ -62,6 +63,12 @@ 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.
@@ -148,7 +155,7 @@ type ReplicaStatus struct {
 }
 
 // BinlogReplicaCatalog extends the Catalog interface and provides methods for accessing a BinlogReplicaController
-// for a Catalog.
+// for a Catalog. The controller can also be used as a BinlogConsumer for BINLOG statement execution.
 type BinlogReplicaCatalog interface {
 	// HasBinlogReplicaController returns true if a non-nil BinlogReplicaController is available for this BinlogReplicaCatalog.
 	HasBinlogReplicaController() bool

sql/plan/binlog.go

@@ -16,6 +16,7 @@ package plan
 
 import (
 	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/binlogreplication"
 	"github.com/dolthub/go-mysql-server/sql/types"
 )
 
@@ -27,47 +28,33 @@ 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.
 //
-// This implementation supports row-based replication (RBR) events, which is the modern standard for MySQL/MariaDB
-// replication (default since MySQL 5.7+ and MariaDB 10.2+).
-//
-// The base64 string is split by newlines and decoded into a buffer of raw binlog events. Each event begins with a
-// 19-byte header containing the event type at byte 4 and event length at bytes 9-12. The buffer is processed
-// sequentially, dispatching each event to a handler based on its type.
-//
-// FORMAT_DESCRIPTION_EVENT stores binlog format metadata in session state. This metadata includes header sizes for each
-// event type and the checksum algorithm (OFF, CRC32, or UNDEF). Subsequent events require this metadata to parse their
-// headers and determine whether checksums are present.
-//
-// TABLE_MAP_EVENT creates a mapping from a table ID to the database name, table name, and column metadata. Row events
-// reference tables by ID rather than name for encoding efficiency. The mapping is stored in a global cache for use by
-// subsequent row events.
-//
-// WRITE_ROWS_EVENT, UPDATE_ROWS_EVENT, and DELETE_ROWS_EVENT contain binary-encoded row data. Before parsing row data,
-// any CRC32 checksum appended to the event must be stripped. Checksums verify data integrity during network
-// transmission and disk storage but are not part of the event payload structure.
-//
-// Transaction boundary and metadata events are silently ignored since each BINLOG statement is auto-committed and full
-// replication semantics are not required for mysqldump replay.
-//
-// QUERY_EVENT (statement-based replication) is not currently supported. Statement-based replication is deprecated in
-// favor of row-based replication to correctly support non-deterministic functions.
+// 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.
 //
 // See https://dev.mysql.com/doc/refman/8.4/en/binlog.html for the BINLOG statement specification.
 type Binlog struct {
-	Base64Str string
-	Catalog   sql.Catalog
+	Base64Str         string
+	ReplicaController binlogreplication.BinlogReplicaController
 }
 
 var _ sql.Node = (*Binlog)(nil)
+var _ BinlogReplicaControllerCommand = (*Binlog)(nil)
 
 // NewBinlog creates a new Binlog node.
-func NewBinlog(base64Str string, catalog sql.Catalog) *Binlog {
+func NewBinlog(base64Str string) *Binlog {
 	return &Binlog{
 		Base64Str: base64Str,
-		Catalog:   catalog,
 	}
 }
 
+// WithBinlogReplicaController implements the BinlogReplicaControllerCommand interface.
+func (b *Binlog) WithBinlogReplicaController(controller binlogreplication.BinlogReplicaController) sql.Node {
+	nc := *b
+	nc.ReplicaController = controller
+	return &nc
+}
+
 func (b *Binlog) String() string {
 	return "BINLOG"
 }

sql/plan/replication_commands.go

@@ -62,7 +62,7 @@ type BinlogPrimaryControllerCommand interface {
 // TODO: When PRIVILEGE_CHECKS_USER option is specified, validate that the assigned user account has the
 // REPLICATION_APPLIER privilege. This validation should happen before the option is passed to the integrator's
 // BinlogReplicaController.SetReplicationSourceOptions().
-// See https://github.com/mysql/mysql-server/blob/8.0/sql/rpl_replica.cc (change_master_cmd function)
+// See https://github.com/mysql/mysql-server/blob/8.0/sql/rpl_replica.cc change_master_cmd
 type ChangeReplicationSource struct {
 	ReplicaController binlogreplication.BinlogReplicaController
 	Options           []binlogreplication.ReplicationOption

sql/planbuilder/builder.go

@@ -406,7 +406,11 @@ func (b *Builder) buildSubquery(inScope *scope, stmt ast.Statement, subQuery str
 			b.handleErr(err)
 		}
 		outScope = inScope.push()
-		outScope.node = plan.NewBinlog(n.Base64Str, b.cat)
+		binlogNode := plan.NewBinlog(n.Base64Str)
+		if binCat, ok := b.cat.(binlogreplication.BinlogReplicaCatalog); ok && binCat.HasBinlogReplicaController() {
+			binlogNode = binlogNode.WithBinlogReplicaController(binCat.GetBinlogReplicaController()).(*plan.Binlog)
+		}
+		outScope.node = binlogNode
 	}
 	return
 }