class CloudwatchMetadataHandler extends MetadataHandler {

    @Override
    public void getPartitions(BlockWriter blockWriter, GetTableLayoutRequest request, QueryStatusChecker queryStatusChecker)
            throws Exception
    {
        CloudwatchTableName cwTableName = tableResolver.validateTable(request.getTableName());

        DescribeLogStreamsRequest cwRequest = new DescribeLogStreamsRequest(cwTableName.getLogGroupName());
        if (!ALL_LOG_STREAMS_TABLE.equals(cwTableName.getLogStreamName())) {
            cwRequest.setLogStreamNamePrefix(cwTableName.getLogStreamName());
        }

        DescribeLogStreamsResult result;
        do {
            result = invoker.invoke(() -> awsLogs.describeLogStreams(cwRequest));
            for (LogStream next : result.getLogStreams()) {
                //Each log stream that matches any possible partition pruning should be added to the partition list.
                blockWriter.writeRows((Block block, int rowNum) -> {
                    boolean matched = block.setValue(LOG_GROUP_FIELD, rowNum, cwRequest.getLogGroupName());
                    matched &= block.setValue(LOG_STREAM_FIELD, rowNum, next.getLogStreamName());
                    matched &= block.setValue(LOG_STREAM_SIZE_FIELD, rowNum, next.getStoredBytes());
                    return matched ? 1 : 0;
                });
            }
            cwRequest.setNextToken(result.getNextToken());
        }
        while (result.getNextToken() != null && queryStatusChecker.isQueryRunning());
    }

    public  CloudwatchMetadataHandler();
    @VisibleForTesting protected  CloudwatchMetadataHandler(AWSLogs awsLogs,
            EncryptionKeyFactory keyFactory,
            AWSSecretsManager secretsManager,
            AmazonAthena athena,
            String spillBucket,
            String spillPrefix);

}