perf: avoid arrow IPC & unneeded steps when flushing delta (#247)

* perf: avoid unneeded steps when flushing delta
* fix: implement workaround for RegisterView bug

Author: fanyang01

adapter/adapter.go

@@ -21,6 +21,10 @@ func GetConn(ctx *sql.Context) (*stdsql.Conn, error) {
 	return ctx.Session.(ConnectionHolder).GetConn(ctx)
 }
 
+func GetCatalogConn(ctx *sql.Context) (*stdsql.Conn, error) {
+	return ctx.Session.(ConnectionHolder).GetCatalogConn(ctx)
+}
+
 func CloseBackendConn(ctx *sql.Context) {
 	ctx.Session.(ConnectionHolder).CloseBackendConn()
 }

binlogreplication/binlog_replica_applier.go

@@ -1245,14 +1245,18 @@ func (a *binlogReplicaApplier) appendRowFormatChanges(
 }
 
 func (a *binlogReplicaApplier) flushDeltaBuffer(ctx *sql.Context, reason delta.FlushReason) error {
+	conn, err := adapter.GetCatalogConn(ctx)
+	if err != nil {
+		return err
+	}
 	tx, err := adapter.GetCatalogTxn(ctx, nil)
 	if err != nil {
 		return err
 	}
 
 	defer a.deltaBufSize.Store(0)
 
-	if err = a.tableWriterProvider.FlushDeltaBuffer(ctx, tx, reason); err != nil {
+	if err = a.tableWriterProvider.FlushDeltaBuffer(ctx, conn, tx, reason); err != nil {
 		ctx.GetLogger().Errorf("Failed to flush changelog: %v", err.Error())
 		MyBinlogReplicaController.setSqlError(sqlerror.ERUnknownError, err.Error())
 	}

binlogreplication/writer.go

@@ -51,7 +51,7 @@ type TableWriterProvider interface {
 	) (DeltaAppender, error)
 
 	// FlushDelta writes the accumulated changes to the database.
-	FlushDeltaBuffer(ctx *sql.Context, tx *stdsql.Tx, reason delta.FlushReason) error
+	FlushDeltaBuffer(ctx *sql.Context, conn *stdsql.Conn, tx *stdsql.Tx, reason delta.FlushReason) error
 
 	// DiscardDeltaBuffer discards the accumulated changes.
 	DiscardDeltaBuffer(ctx *sql.Context)