'Runs a query against Druid and returns a dataframe.'

def run_query(self, groupby, metrics, granularity, from_dttm, to_dttm, filter=None, is_timeseries=True, timeseries_limit=None, timeseries_limit_metric=None, row_limit=None, inner_from_dttm=None, inner_to_dttm=None, orderby=None, extras=None, select=None, columns=None, phase=2, client=None, form_data=None):

    client = (client or self.cluster.get_pydruid_client())
    if (not is_timeseries):
        granularity = 'all'
    inner_from_dttm = (inner_from_dttm or from_dttm)
    inner_to_dttm = (inner_to_dttm or to_dttm)
    from_dttm = from_dttm.replace(tzinfo=DRUID_TZ)
    to_dttm = to_dttm.replace(tzinfo=DRUID_TZ)
    timezone = from_dttm.tzname()
    query_str = ''
    metrics_dict = {m.metric_name: m for m in self.metrics}
    columns_dict = {c.column_name: c for c in self.columns}
    (all_metrics, post_aggs) = self._metrics_and_post_aggs(metrics, metrics_dict)
    aggregations = OrderedDict()
    for m in self.metrics:
        if (m.metric_name in all_metrics):
            aggregations[m.metric_name] = m.json_obj
    rejected_metrics = [m.metric_name for m in self.metrics if (m.is_restricted and (m.metric_name in aggregations.keys()) and (not sm.has_access('metric_access', m.perm)))]
    if rejected_metrics:
        raise MetricPermException(('Access    to    the    metrics    denied:    ' + ',    '.join(rejected_metrics)))
    dimensions = []
    groupby = [gb for gb in groupby if (gb in columns_dict)]
    for column_name in groupby:
        col = columns_dict.get(column_name)
        dim_spec = col.dimension_spec
        if dim_spec:
            dimensions.append(dim_spec)
        else:
            dimensions.append(column_name)
    qry = dict(datasource=self.datasource_name, dimensions=dimensions, aggregations=aggregations, granularity=DruidDatasource.granularity(granularity, timezone=timezone, origin=extras.get('druid_time_origin')), post_aggregations=post_aggs, intervals=((from_dttm.isoformat() + '/') + to_dttm.isoformat()))
    filters = self.get_filters(filter)
    if filters:
        qry['filter'] = filters
    having_filters = self.get_having_filters(extras.get('having_druid'))
    if having_filters:
        qry['having'] = having_filters
    orig_filters = filters
    if ((len(groupby) == 0) and (not having_filters)):
        del qry['dimensions']
        client.timeseries(**qry)
    if ((not having_filters) and (len(groupby) == 1)):
        qry['threshold'] = (timeseries_limit or 1000)
        if (row_limit and (granularity == 'all')):
            qry['threshold'] = row_limit
        qry['dimension'] = list(qry.get('dimensions'))[0]
        del qry['dimensions']
        qry['metric'] = list(qry['aggregations'].keys())[0]
        client.topn(**qry)
    elif ((len(groupby) > 1) or having_filters):
        if (timeseries_limit and is_timeseries):
            order_by = (metrics[0] if metrics else self.metrics[0])
            if timeseries_limit_metric:
                order_by = timeseries_limit_metric
            pre_qry = deepcopy(qry)
            pre_qry['granularity'] = 'all'
            pre_qry['limit_spec'] = {'type': 'default', 'limit': timeseries_limit, 'intervals': ((inner_from_dttm.isoformat() + '/') + inner_to_dttm.isoformat()), 'columns': [{'dimension': order_by, 'direction': 'descending'}]}
            client.groupby(**pre_qry)
            query_str += '//    Two    phase    query\n//    Phase    1\n'
            query_str += json.dumps(client.query_builder.last_query.query_dict, indent=2)
            query_str += '\n'
            if (phase == 1):
                return query_str
            query_str += "//\nPhase    2    (built    based    on    phase    one's    results)\n"
            df = client.export_pandas()
            if ((df is not None) and (not df.empty)):
                dims = qry['dimensions']
                filters = []
                for (unused, row) in df.iterrows():
                    fields = []
                    for dim in dims:
                        f = (Dimension(dim) == row[dim])
                        fields.append(f)
                    if (len(fields) > 1):
                        filt = Filter(type='and', fields=fields)
                        filters.append(filt)
                    elif fields:
                        filters.append(fields[0])
                if filters:
                    ff = Filter(type='or', fields=filters)
                    if (not orig_filters):
                        qry['filter'] = ff
                    else:
                        qry['filter'] = Filter(type='and', fields=[ff, orig_filters])
                qry['limit_spec'] = None
        if row_limit:
            qry['limit_spec'] = {'type': 'default', 'limit': row_limit, 'columns': [{'dimension': (metrics[0] if metrics else self.metrics[0]), 'direction': 'descending'}]}
        client.groupby(**qry)
    query_str += json.dumps(client.query_builder.last_query.query_dict, indent=2)
    return query_str
