Method executeInTransactionMode

backend/plugin/db/tidb/tidb.go:240–312

executeInTransactionMode executes statements within a single transaction

(ctx context.Context, conn *sql.Conn, commands []base.Statement, opts db.ExecuteOptions, connectionID string)

// executeInTransactionMode executes statements within a single transaction
func (d *Driver) executeInTransactionMode(ctx context.Context, conn *sql.Conn, commands []base.Statement, opts db.ExecuteOptions, connectionID string) (int64, error) {
	var totalRowsAffected int64

	if err := conn.Raw(func(driverConn any) error {
		//nolint
		exer := driverConn.(driver.ExecerContext)
		//nolint
		txer := driverConn.(driver.ConnBeginTx)
		tx, err := txer.BeginTx(ctx, driver.TxOptions{})
		if err != nil {
			opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, err.Error())
			return err
		} else {
			opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, "")
		}

		committed := false
		defer func() {
			err := tx.Rollback()
			if committed {
				return
			}
			var rerr string
			if err != nil {
				rerr = err.Error()
			}
			opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_ROLLBACK, rerr)
		}()

		for _, command := range commands {
			opts.LogCommandExecute(command.Range, command.Text)

			sqlWithBytebaseAppComment := util.MySQLPrependBytebaseAppComment(command.Text)
			sqlResult, err := exer.ExecContext(ctx, sqlWithBytebaseAppComment, nil)
			if err != nil {
				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
					slog.Info("cancel connection", slog.String("connectionID", connectionID))
					if err := d.StopConnectionByID(connectionID); err != nil {
						slog.Error("failed to cancel connection", slog.String("connectionID", connectionID), log.BBError(err))
					}
				}

				opts.LogCommandResponse(0, nil, err.Error())

				return err
			}

			allRowsAffected := sqlResult.(mysql.Result).AllRowsAffected()
			var rowsAffected int64
			var allRowsAffectedInt64 []int64
			for _, a := range allRowsAffected {
				rowsAffected += a
				allRowsAffectedInt64 = append(allRowsAffectedInt64, a)
			}
			totalRowsAffected += rowsAffected

			opts.LogCommandResponse(rowsAffected, allRowsAffectedInt64, "")
		}