Method executeInTransactionMode

backend/plugin/db/starrocks/starrocks.go:194–251

(ctx context.Context, statement string, opts db.ExecuteOptions)

}

func (d *Driver) executeInTransactionMode(ctx context.Context, statement string, opts db.ExecuteOptions) (int64, error) {
	conn, err := d.db.Conn(ctx)
	if err != nil {
		return 0, err
	}
	defer conn.Close()

	connectionID, err := getConnectionID(ctx, conn)
	if err != nil {
		return 0, err
	}
	slog.Debug("connectionID", slog.String("connectionID", connectionID))

	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, err.Error())
		return 0, errors.Wrapf(err, "failed to begin execute transaction")
	}
	opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, "")

	committed := false
	defer func() {
		if !committed {
			err := tx.Rollback()
			var rerr string
			if err != nil && !errors.Is(err, sql.ErrTxDone) {
				rerr = err.Error()
				slog.Debug("failed to rollback transaction", log.BBError(err))
			}
			opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_ROLLBACK, rerr)
		}
	}()

	sqlResult, err := tx.ExecContext(ctx, statement)
	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))
			}
		}
		return 0, err
	}

	rowsAffected, err := sqlResult.RowsAffected()
	if err != nil {
		// Since we cannot differentiate DDL and DML yet, we have to ignore the error.
		slog.Debug("rowsAffected returns error", log.BBError(err))
	}

	if err := tx.Commit(); err != nil {
		opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_COMMIT, err.Error())
		return 0, errors.Wrapf(err, "failed to commit execute transaction")
	}
	opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_COMMIT, "")
	committed = true
	return rowsAffected, nil
}