class MetricsRecordHandler extends RecordHandler {

    private void readMetricSamplesWithConstraint(BlockSpiller blockSpiller, ReadRecordsRequest request, QueryStatusChecker queryStatusChecker)
            throws TimeoutException
    {
        GetMetricDataRequest dataRequest = MetricUtils.makeGetMetricDataRequest(request);
        Map<String, MetricDataQuery> queries = new HashMap<>();
        for (MetricDataQuery query : dataRequest.getMetricDataQueries()) {
            queries.put(query.getId(), query);
        }

        String prevToken;
        ValueSet dimensionNameConstraint = request.getConstraints().getSummary().get(DIMENSION_NAME_FIELD);
        ValueSet dimensionValueConstraint = request.getConstraints().getSummary().get(DIMENSION_NAME_FIELD);
        do {
            prevToken = dataRequest.getNextToken();
            GetMetricDataResult result = invoker.invoke(() -> metrics.getMetricData(dataRequest));
            for (MetricDataResult nextMetric : result.getMetricDataResults()) {
                MetricStat metricStat = queries.get(nextMetric.getId()).getMetricStat();
                List<Date> timestamps = nextMetric.getTimestamps();
                List<Double> values = nextMetric.getValues();
                for (int i = 0; i < nextMetric.getValues().size(); i++) {
                    int sampleNum = i;
                    blockSpiller.writeRows((Block block, int row) -> {
                        /**
                         * Most constraints were already applied at split generation so we only need to apply
                         * a subset.
                         */
                        block.offerValue(METRIC_NAME_FIELD, row, metricStat.getMetric().getMetricName());
                        block.offerValue(NAMESPACE_FIELD, row, metricStat.getMetric().getNamespace());
                        block.offerValue(STATISTIC_FIELD, row, metricStat.getStat());

                        block.offerComplexValue(DIMENSIONS_FIELD,
                                row,
                                (Field field, Object val) -> {
                                    if (field.getName().equals(DIMENSION_NAME_FIELD)) {
                                        return ((Dimension) val).getName();
                                    }
                                    else if (field.getName().equals(DIMENSION_VALUE_FIELD)) {
                                        return ((Dimension) val).getValue();
                                    }

                                    throw new RuntimeException("Unexpected field " + field.getName());
                                },
                                metricStat.getMetric().getDimensions());

                        //This field is 'faked' in that we just use it as a convenient way to filter single dimensions. As such
                        //we always populate it with the value of the filter if the constraint passed and the filter was singleValue
                        String dimName = (dimensionNameConstraint == null || !dimensionNameConstraint.isSingleValue())
                                ? null : dimensionNameConstraint.getSingleValue().toString();
                        block.offerValue(DIMENSION_NAME_FIELD, row, dimName);

                        //This field is 'faked' in that we just use it as a convenient way to filter single dimensions. As such
                        //we always populate it with the value of the filter if the constraint passed and the filter was singleValue
                        String dimVal = (dimensionValueConstraint == null || !dimensionValueConstraint.isSingleValue())
                                ? null : dimensionValueConstraint.getSingleValue().toString();
                        block.offerValue(DIMENSION_VALUE_FIELD, row, dimVal);

                        block.offerValue(PERIOD_FIELD, row, metricStat.getPeriod());

                        boolean matches = true;
                        block.offerValue(VALUE_FIELD, row, values.get(sampleNum));
                        long timestamp = timestamps.get(sampleNum).getTime() / 1000;
                        block.offerValue(TIMESTAMP_FIELD, row, timestamp);

                        return matches ? 1 : 0;
                    });
                }
            }
            dataRequest.setNextToken(result.getNextToken());
        }
        while (dataRequest.getNextToken() != null && !dataRequest.getNextToken().equalsIgnoreCase(prevToken) && queryStatusChecker.isQueryRunning());
    }

    public  MetricsRecordHandler();
    @VisibleForTesting protected  MetricsRecordHandler(AmazonS3 amazonS3, AWSSecretsManager secretsManager, AmazonAthena athena, AmazonCloudWatch metrics);

}