Make it work

Author: fanyang01

adapter/adapter.go

@@ -11,6 +11,9 @@ type ConnectionHolder interface {
 	GetConn(ctx context.Context) (*stdsql.Conn, error)
 	GetTxn(ctx context.Context, options *stdsql.TxOptions) (*stdsql.Tx, error)
 	GetCatalogConn(ctx context.Context) (*stdsql.Conn, error)
+	GetCatalogTxn(ctx context.Context, options *stdsql.TxOptions) (*stdsql.Tx, error)
+	TryGetTxn() *stdsql.Tx
+	CloseTxn()
 }
 
 func GetConn(ctx *sql.Context) (*stdsql.Conn, error) {
@@ -21,6 +24,18 @@ func GetTxn(ctx *sql.Context, options *stdsql.TxOptions) (*stdsql.Tx, error) {
 	return ctx.Session.(ConnectionHolder).GetTxn(ctx, options)
 }
 
+func GetCatalogTxn(ctx *sql.Context, options *stdsql.TxOptions) (*stdsql.Tx, error) {
+	return ctx.Session.(ConnectionHolder).GetCatalogTxn(ctx, options)
+}
+
+func TryGetTxn(ctx *sql.Context) *stdsql.Tx {
+	return ctx.Session.(ConnectionHolder).TryGetTxn()
+}
+
+func CloseTxn(ctx *sql.Context) {
+	ctx.Session.(ConnectionHolder).CloseTxn()
+}
+
 func Query(ctx *sql.Context, query string, args ...any) (*stdsql.Rows, error) {
 	conn, err := GetConn(ctx)
 	if err != nil {
@@ -75,6 +90,14 @@ func ExecCatalog(ctx *sql.Context, query string, args ...any) (stdsql.Result, er
 	return conn.ExecContext(ctx, query, args...)
 }
 
+func ExecCatalogInTxn(ctx *sql.Context, query string, args ...any) (stdsql.Result, error) {
+	tx, err := ctx.Session.(ConnectionHolder).GetCatalogTxn(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	return tx.ExecContext(ctx, query, args...)
+}
+
 func ExecInTxn(ctx *sql.Context, query string, args ...any) (stdsql.Result, error) {
 	tx, err := GetTxn(ctx, nil)
 	if err != nil {

backend/connpool.go

@@ -119,6 +119,14 @@ func (p *ConnectionPool) GetTxn(ctx context.Context, id uint32, schemaName strin
 	return tx, nil
 }
 
+func (p *ConnectionPool) TryGetTxn(id uint32) *stdsql.Tx {
+	entry, ok := p.txns.Load(id)
+	if !ok {
+		return nil
+	}
+	return entry.(*stdsql.Tx)
+}
+
 func (p *ConnectionPool) CloseTxn(id uint32) {
 	p.txns.Delete(id)
 }

backend/session.go

@@ -90,7 +90,7 @@ func (sess Session) StartTransaction(ctx *sql.Context, tCharacteristic sql.Trans
 
 	var tx *stdsql.Tx
 	if startUnderlyingTx {
-		sess.GetLogger().Infoln("StartDuckTransaction")
+		sess.GetLogger().Trace("StartDuckTransaction")
 		tx, err = sess.GetTxn(ctx, &stdsql.TxOptions{ReadOnly: tCharacteristic == sql.ReadOnly})
 		if err != nil {
 			return nil, err
@@ -101,10 +101,10 @@ func (sess Session) StartTransaction(ctx *sql.Context, tCharacteristic sql.Trans
 
 // CommitTransaction implements sql.TransactionSession.
 func (sess Session) CommitTransaction(ctx *sql.Context, tx sql.Transaction) error {
-	sess.GetLogger().Infoln("CommitTransaction")
+	sess.GetLogger().Trace("CommitTransaction")
 	transaction := tx.(*Transaction)
 	if transaction.tx != nil {
-		sess.GetLogger().Infoln("CommitDuckTransaction")
+		sess.GetLogger().Trace("CommitDuckTransaction")
 		defer sess.CloseTxn()
 		if err := transaction.tx.Commit(); err != nil {
 			return err
@@ -115,10 +115,10 @@ func (sess Session) CommitTransaction(ctx *sql.Context, tx sql.Transaction) erro
 
 // Rollback implements sql.TransactionSession.
 func (sess Session) Rollback(ctx *sql.Context, tx sql.Transaction) error {
-	sess.GetLogger().Infoln("Rollback")
+	sess.GetLogger().Trace("Rollback")
 	transaction := tx.(*Transaction)
 	if transaction.tx != nil {
-		sess.GetLogger().Infoln("RollbackDuckTransaction")
+		sess.GetLogger().Trace("RollbackDuckTransaction")
 		defer sess.CloseTxn()
 		if err := transaction.tx.Rollback(); err != nil {
 			return err
@@ -193,10 +193,22 @@ func (sess Session) GetCatalogConn(ctx context.Context) (*stdsql.Conn, error) {
 	return sess.pool.GetConn(ctx, sess.ID())
 }
 
+// GetTxn implements adapter.ConnectionHolder.
 func (sess Session) GetTxn(ctx context.Context, options *stdsql.TxOptions) (*stdsql.Tx, error) {
 	return sess.pool.GetTxn(ctx, sess.ID(), sess.GetCurrentDatabase(), options)
 }
 
+// GetCatalogTxn implements adapter.ConnectionHolder.
+func (sess Session) GetCatalogTxn(ctx context.Context, options *stdsql.TxOptions) (*stdsql.Tx, error) {
+	return sess.pool.GetTxn(ctx, sess.ID(), "", options)
+}
+
+// TryGetTxn implements adapter.ConnectionHolder.
+func (sess Session) TryGetTxn() *stdsql.Tx {
+	return sess.pool.TryGetTxn(sess.ID())
+}
+
+// CloseTxn implements adapter.ConnectionHolder.
 func (sess Session) CloseTxn() {
 	sess.pool.CloseTxn(sess.ID())
 }

binlogreplication/binlog_position_store.go

@@ -32,7 +32,6 @@ import (
 )
 
 const binlogPositionDirectory = ".replica"
-const binlogPositionFilename = "binlog-position"
 const mysqlFlavor = "MySQL56"
 const defaultChannelName = ""
 
@@ -61,10 +60,7 @@ func (store *binlogPositionStore) Load(ctx *sql.Context, engine *gms.Engine) (po
 	}
 
 	// Strip off the "MySQL56/" prefix
-	prefix := "MySQL56/"
-	if strings.HasPrefix(positionString, prefix) {
-		positionString = positionString[len(prefix):]
-	}
+	positionString = strings.TrimPrefix(positionString, "MySQL56/")
 
 	return replication.ParsePosition(mysqlFlavor, positionString)
 }
@@ -81,7 +77,11 @@ func (store *binlogPositionStore) Save(ctx *sql.Context, engine *gms.Engine, pos
 	store.mu.Lock()
 	defer store.mu.Unlock()
 
-	if _, err := adapter.ExecInTxn(ctx, catalog.InternalTables.BinlogPosition.UpsertStmt(), defaultChannelName, position.String()); err != nil {
+	if _, err := adapter.ExecCatalogInTxn(
+		ctx,
+		catalog.InternalTables.BinlogPosition.UpsertStmt(),
+		defaultChannelName, position.String(),
+	); err != nil {
 		return fmt.Errorf("unable to save binlog position: %w", err)
 	}
 	return nil
@@ -94,7 +94,7 @@ func (store *binlogPositionStore) Delete(ctx *sql.Context, engine *gms.Engine) e
 	store.mu.Lock()
 	defer store.mu.Unlock()
 
-	_, err := adapter.ExecInTxn(ctx, catalog.InternalTables.BinlogPosition.DeleteStmt(), defaultChannelName)
+	_, err := adapter.ExecCatalogInTxn(ctx, catalog.InternalTables.BinlogPosition.DeleteStmt(), defaultChannelName)
 	return err
 }