perf: avoid unneeded steps when flushing delta

apecloud · Dec 3, 2024 · 07f655a · 07f655a
1 parent c2372a8
commit 07f655a
Show file tree

Hide file tree

Showing 5 changed files with 331 additions and 123 deletions.
diff --git a/binlogreplication/binlog_replica_applier.go b/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.GetConn(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())
 	}

diff --git a/binlogreplication/writer.go b/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)