core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccessImpl.java
@Override
    public Observable<Observable<Row>> findAllDataFromBucket(long timestamp, int pageSize, int maxConcurrency) {
        PreparedStatement ts =
                getTempStatement(MetricType.UNDEFINED, TempStatement.SCAN_WITH_TOKEN_RANGES, timestamp);

        // The table does not exists - case such as when starting Hawkular-Metrics for the first time just before
        // compression kicks in.
        if(ts == null || prepMap.floorKey(timestamp) == 0L) {
            return Observable.empty();
        }

        return Observable.from(getTokenRanges())
                .map(tr -> rxSession.executeAndFetch(
                        ts
                                .bind()
                                .setToken(0, tr.getStart())
                                .setToken(1, tr.getEnd())
                                .setFetchSize(pageSize)));
    }
@Override
    public Observable<Observable<Row>> findAllDataFromBucket(long timestamp, int pageSize, int maxConcurrency) {
        PreparedStatement ts =
                getTempStatement(MetricType.UNDEFINED, TempStatement.SCAN_WITH_TOKEN_RANGES, timestamp);

        // The table does not exists - case such as when starting Hawkular-Metrics for the first time just before
        // compression kicks in.
        if(ts == null || prepMap.floorKey(timestamp) == 0L) {
            return Observable.empty();
        }

        return Observable.from(getTokenRanges())
                .map(tr -> rxSession.executeAndFetch(
                        ts
                                .bind()
                                .setToken(0, tr.getStart())
                                .setToken(1, tr.getEnd())
                                .setFetchSize(pageSize)));
    }
@ Override public Observable < Observable < Row > > findAllDataFromBucket ( long timestamp , int pageSize , int maxConcurrency ) { PreparedStatement ts = getTempStatement ( MetricType . UNDEFINED , TempStatement . SCAN_WITH_TOKEN_RANGES , timestamp ) ; // The table does not exists - case such as when starting Hawkular-Metrics for the first time just before // compression kicks in. if ( ts == null || prepMap . floorKey ( timestamp ) == 0L ) { return Observable . empty ( ) ; } return Observable . from ( getTokenRanges ( ) ) . map ( tr -> rxSession . executeAndFetch ( ts . bind ( ) . setToken ( 0 , tr . getStart ( ) ) . setToken ( 1 , tr . getEnd ( ) ) . setFetchSize ( pageSize ) ) ) ; }
Fetch all the data from a temporary table for the compression job. Using TokenRanges avoids fetching first
all the metrics' partition keys and then requesting them.

Performance can be improved by using data locality and fetching with multiple threads.

@param timestamp A timestamp inside the wanted bucket (such as the previous starting row timestamp)
@param pageSize How many rows to fetch each time
@param maxConcurrency To how many streams should token ranges be split to
@return Observable of Observables per partition key
Fetch all the data from a temporary table for the compression job . Using TokenRanges avoids fetching first all the metrics partition keys and then requesting them .