def run(self, timeout=9000):
        """Run this Query. This will first query the bucket using indexes and
        get the intersection of these keys. Later this keys are passed to
        MapReduce phase to fetch and sort the data.
        """

        index_key_sets = []
        index_query_threads = []
        queue = Queue.Queue()
        for (field, op, value) in self._filters:
            # spin off threads to do index queries
            thd = threading.Thread(target=self._filter_to_index_query,
                                   args=(field, op, value, queue))
            index_query_threads.append(thd)
            thd.start()

        # wait for the threads to finish the job
        [thd.join() for thd in index_query_threads]

        # get the results back from the queue
        while not queue.empty():
            key_set = queue.get()
            index_key_sets.append(key_set)

        # find the intersection of the key sets.
        if index_key_sets:
            mr_inputs = reduce(lambda x, y: x.intersection(y), index_key_sets)
        else:
            mr_inputs = set()

        if self._filters and not mr_inputs:
            yield []

        if not mr_inputs:
            self._mr_query = self._client.add(self._bucket)
        for key in mr_inputs:
            self._mr_query.add(self._bucket, key)

        self._mr_query.map(JS_MAP_FUNCTION)

        if self._order:
            self._mr_query.reduce(JS_REDUCE_ORDER_FUNC,
                                  {'arg': {'by': self._order[0],
                                           'order': self._order[1].lower()
                                           }
                                   })

        if self._limit:
            start = self._offset
            end = self._offset + self._limit
            if (end > len(mr_inputs)) and self._filters:
                end = 0
            self._mr_query.reduce('Riak.reduceSlice', {'arg': [start, end]})

        for result in self._mr_query.run(timeout):
            yield result
dataset/ETHPy150Open mining/mining/mining/db/backends/mriak/multiquery.py/RiakMultiIndexQuery.run/original