@Test
public void testDeleteSecretMaliciousResults() {
    QueryRequest request = constructQueryRequest(SECRET_NAME);

    // Result contains entries that do no match the filter in the request (i.e. items for secret2). So it should be
    // considered malicious.
    QueryResult maliciousResult = constructQueryResult(true);
    when(mockDynamoDBClient.query(request)).thenReturn(maliciousResult);

    // Call the delete secret method.
    boolean consideredMalicious = false;
    try {
        dynamoDB.delete(new SecretIdentifier(SECRET_NAME));
    } catch (PotentiallyMaliciousDataException e) {
        consideredMalicious = true;
    }
    assertTrue(consideredMalicious);

    // Verify nothing was actually deleted because the malicious check failed.
    verify(mockDynamoDBClient, times(1)).query(request);
    verify(mockDynamoDBClient, never()).deleteItem(tableName, constructKey(SECRET_NAME, 1));
    verify(mockDynamoDBClient, never()).deleteItem(tableName, constructKey(SECRET_NAME, 2));
    verify(mockDynamoDBClient, never()).deleteItem(tableName, constructKey(SECRET2_NAME, 1));
}
 

public RetryResult<QueryResult> queryTable(
    String tableName, DynamoDBQueryFilter dynamoDBQueryFilter, Map<String, AttributeValue>
    exclusiveStartKey, long limit, Reporter reporter) {
  final QueryRequest queryRequest = new QueryRequest()
      .withTableName(tableName)
      .withExclusiveStartKey(exclusiveStartKey)
      .withKeyConditions(dynamoDBQueryFilter.getKeyConditions())
      .withLimit(Ints.checkedCast(limit))
      .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

  RetryResult<QueryResult> retryResult = getRetryDriver().runWithRetry(
      new Callable<QueryResult>() {
        @Override
        public QueryResult call() {
          log.debug("Executing DynamoDB query: " + queryRequest);
          return dynamoDB.query(queryRequest);
        }
      }, reporter, PrintCounter.DynamoDBReadThrottle);
  return retryResult;
}
 

private static void findRepliesForAThread(String forumName, String threadSubject) {

        String replyId = forumName + "#" + threadSubject;

        Condition partitionKeyCondition = new Condition().withComparisonOperator(ComparisonOperator.EQ)
            .withAttributeValueList(new AttributeValue().withS(replyId));

        Map<String, Condition> keyConditions = new HashMap<String, Condition>();
        keyConditions.put("Id", partitionKeyCondition);

        QueryRequest queryRequest = new QueryRequest().withTableName(tableName).withKeyConditions(keyConditions);

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }
    }
 

private static void findRepliesInLast15DaysWithConfig(String forumName, String threadSubject) {

        long twoWeeksAgoMilli = (new Date()).getTime() - (15L * 24L * 60L * 60L * 1000L);
        Date twoWeeksAgo = new Date();
        twoWeeksAgo.setTime(twoWeeksAgoMilli);
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        String twoWeeksAgoStr = df.format(twoWeeksAgo);

        Condition sortKeyCondition = new Condition().withComparisonOperator(ComparisonOperator.GT.toString())
            .withAttributeValueList(new AttributeValue().withS(twoWeeksAgoStr));

        Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);
        keyConditions.put("ReplyDateTime", sortKeyCondition);

        QueryRequest queryRequest = new QueryRequest().withTableName(tableName).withKeyConditions(keyConditions)
            .withProjectionExpression("Message, ReplyDateTime, PostedBy");

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }

    }
 

private static void findRepliesPostedWithinTimePeriod(String forumName, String threadSubject) {

        long startDateMilli = (new Date()).getTime() - (15L * 24L * 60L * 60L * 1000L);
        long endDateMilli = (new Date()).getTime() - (5L * 24L * 60L * 60L * 1000L);
        java.text.SimpleDateFormat df = new java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        String startDate = df.format(startDateMilli);
        String endDate = df.format(endDateMilli);

        Condition sortKeyCondition = new Condition().withComparisonOperator(ComparisonOperator.BETWEEN.toString())
            .withAttributeValueList(new AttributeValue().withS(startDate), new AttributeValue().withS(endDate));

        Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);
        keyConditions.put("ReplyDateTime", sortKeyCondition);

        QueryRequest queryRequest = new QueryRequest().withTableName(tableName).withKeyConditions(keyConditions)
            .withProjectionExpression("Message, ReplyDateTime, PostedBy");

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }
    }
 

private static void findRepliesUsingAFilterExpression(String forumName, String threadSubject) {

        Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);

        Map<String, AttributeValue> expressionAttributeValues = new HashMap<String, AttributeValue>();
        expressionAttributeValues.put(":val", new AttributeValue().withS("User B"));

        QueryRequest queryRequest = new QueryRequest().withTableName(tableName).withKeyConditions(keyConditions)
            .withFilterExpression("PostedBy = :val").withExpressionAttributeValues(expressionAttributeValues)
            .withProjectionExpression("Message, ReplyDateTime, PostedBy");

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }
    }
 

public QueryResult query(final QueryRequest request, final int permitsToConsume) throws BackendException {
    setUserAgent(request);
    QueryResult result;
    timedReadThrottle(QUERY, request.getTableName(), permitsToConsume);
    final Timer.Context apiTimerContext = getTimerContext(QUERY, request.getTableName());
    try {
        result = client.query(request);
    } catch (Exception e) {
        throw processDynamoDbApiException(e, QUERY, request.getTableName());
    } finally {
        apiTimerContext.stop();
    }
    meterConsumedCapacity(QUERY, result.getConsumedCapacity());
    measureItemCount(QUERY, request.getTableName(), result.getCount());
    return result;
}
 

@Override
public QueryResultWrapper next() throws BackendException {
    final Query backoff = new ExponentialBackoff.Query(request, delegate, permitsToConsume);
    final QueryResult result = backoff.runWithBackoff();
    final ConsumedCapacity consumedCapacity = result.getConsumedCapacity();
    if (null != consumedCapacity) {
        permitsToConsume = Math.max((int) (consumedCapacity.getCapacityUnits() - 1.0), 1);
        totalCapacityUnits += consumedCapacity.getCapacityUnits();
    }

    if (result.getLastEvaluatedKey() != null && !result.getLastEvaluatedKey().isEmpty()) {
        request.setExclusiveStartKey(result.getLastEvaluatedKey());
    } else {
        markComplete();
    }
    // a update returned count
    returnedCount += result.getCount();

    // b update scanned count
    scannedCount += result.getScannedCount();
    // c add scanned finalItemList
    finalItemList.addAll(result.getItems());
    return new QueryResultWrapper(titanKey, result);
}
 

@Override
public boolean hasNext() {
    if (closed) {
        return false;
    }
    if (currentIterator.hasNext()) {
        return true;
    }
    // Loop until the query finds a new result.
    // This is necessary because even if the query worker has a next page it might have no results.
    while (queryWorker.hasNext() && !currentIterator.hasNext()) {
        try {
            final QueryResultWrapper resultWrapper = queryWorker.next();
            final QueryResult queryResult = resultWrapper.getDynamoDBResult();

            currentIterator = buildRecordIteratorFromQueryResult(queryResult);
        } catch (BackendException e) {
            throw new RuntimeException(e);
        }
    }
    return currentIterator.hasNext();
}
 

private static void findRepliesForAThread(String forumName, String threadSubject) {

        String replyId = forumName + "#" + threadSubject;
        
        Condition hashKeyCondition = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ)
            .withAttributeValueList(new AttributeValue().withS(replyId));
        
        Map<String, Condition> keyConditions = new HashMap<String, Condition>();
        keyConditions.put("Id", hashKeyCondition);
        
        QueryRequest queryRequest = new QueryRequest()
            .withTableName(tableName)
            .withKeyConditions(keyConditions);

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }
    }
 

private static void findRepliesForAThreadSpecifyOptionalLimit(String forumName, String threadSubject) {
    
    Map<String, AttributeValue> lastEvaluatedKey = null;
    do {
        QueryRequest queryRequest = new QueryRequest()
                .withTableName(tableName)
                .withKeyConditions(makeReplyKeyConditions(forumName, threadSubject))
                .withLimit(1)
                .withExclusiveStartKey(lastEvaluatedKey);

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }
        lastEvaluatedKey = result.getLastEvaluatedKey();
    } while (lastEvaluatedKey != null);        
}
 

private static void findRepliesInLast15DaysWithConfig(String forumName, String threadSubject) {

        long twoWeeksAgoMilli = (new Date()).getTime() - (15L*24L*60L*60L*1000L);
        Date twoWeeksAgo = new Date();
        twoWeeksAgo.setTime(twoWeeksAgoMilli);
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        String twoWeeksAgoStr = df.format(twoWeeksAgo);

        Condition rangeKeyCondition = new Condition()
            .withComparisonOperator(ComparisonOperator.GT.toString())
            .withAttributeValueList(new AttributeValue().withS(twoWeeksAgoStr));
        
        Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);
        keyConditions.put("ReplyDateTime", rangeKeyCondition);
        
        QueryRequest queryRequest = new QueryRequest().withTableName(tableName)
            .withKeyConditions(keyConditions)
            .withProjectionExpression("Message, ReplyDateTime, PostedBy");

        QueryResult result = client.query(queryRequest);
        for (Map<String, AttributeValue> item : result.getItems()) {
            printItem(item);
        }

    }
 

private static void findRepliesUsingAFilterExpression(String forumName, String threadSubject) {
         
   Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);
   
   Map<String, AttributeValue> expressionAttributeValues = new HashMap<String, AttributeValue>();
   expressionAttributeValues.put(":val", new AttributeValue().withS("User B")); 
   
   QueryRequest queryRequest = new QueryRequest()
       .withTableName(tableName)
       .withKeyConditions(keyConditions)
       .withFilterExpression("PostedBy = :val")
       .withExpressionAttributeValues(expressionAttributeValues)
       .withProjectionExpression("Message, ReplyDateTime, PostedBy");

    QueryResult result = client.query(queryRequest);
    for (Map<String, AttributeValue> item : result.getItems()) {
        printItem(item);
    }        
}
 

public void run() {
	QueryRequest queryRequest = DynamoDBUtil.copyQueryRequest(geoQueryRequest.getQueryRequest());
	long hashKey = S2Manager.generateHashKey(range.getRangeMin(), config.getHashKeyLength());

	List<QueryResult> queryResults = dynamoDBManager.queryGeohash(queryRequest, hashKey, range);

	for (QueryResult queryResult : queryResults) {
		if (isInterrupted()) {
			return;
		}

		// getQueryResults() returns a synchronized list.
		geoQueryResult.getQueryResults().add(queryResult);

		List<Map<String, AttributeValue>> filteredQueryResult = filter(queryResult.getItems(), geoQueryRequest);

		// getItem() returns a synchronized list.
		geoQueryResult.getItem().addAll(filteredQueryResult);
	}
}
 

@Override
public final Response query(final Request request) {

    final Map<String, AttributeValue> expressionAttributeValues = new HashMap<>();
    expressionAttributeValues.put(":" + request.getPartitionKey(),
            new AttributeValue(request.getPartitionKeyValue()));

    final String keyConditionExpression = request.getPartitionKey() + "=:" + request.getPartitionKey();

    QueryRequest queryRequest = new QueryRequest()
            .withTableName(request.getTableName())
            .withKeyConditionExpression(keyConditionExpression)
            .withExpressionAttributeValues(expressionAttributeValues);

    StringBuilder filterExpressionBuilder;
    if (request.getFilterData() != null) {
        filterExpressionBuilder = new StringBuilder();
        processFilterData(request, filterExpressionBuilder, expressionAttributeValues);
        //Add to QueryRequest.
        queryRequest.withFilterExpression(filterExpressionBuilder.toString());
    }

    final QueryResult queryResult = dynamoDBClient.query(queryRequest);

    final StringBuilder response = new StringBuilder();
    response.append("PK of items read with query (V2): ");
    for (Map<String, AttributeValue> item : queryResult.getItems()) {
        response.append(prepareKeyStr(item, request));
    }


    return new Response(response.toString(), null);
}
 

TestItem(TableSchema<?> schema,
         QueryResponse v2Response,

         Class<?> v1BeanClass,
         DynamoDBQueryExpression v1QueryExpression,
         QueryResult v1Response) {
    this.schema = schema;
    this.v2Response = v2Response;

    this.v1BeanClass = v1BeanClass;
    this.v1QueryExpression = v1QueryExpression;
    this.v1Response = v1Response;
}
 

@Test
public void testDeleteSecret() {
    QueryRequest request = constructQueryRequest(SECRET_NAME);
    QueryResult result = constructQueryResult(false);
    when(mockDynamoDBClient.query(request)).thenReturn(result);

    // Call the delete secret method.
    dynamoDB.delete(new SecretIdentifier(SECRET_NAME));

    // Verify only the entries matching the correct secret were deleted.
    verify(mockDynamoDBClient, times(1)).query(request);
    verify(mockDynamoDBClient, times(1)).deleteItem(tableName, constructKey(SECRET_NAME, 1));
    verify(mockDynamoDBClient, times(1)).deleteItem(tableName, constructKey(SECRET_NAME, 2));
    verify(mockDynamoDBClient, never()).deleteItem(tableName, constructKey(SECRET2_NAME, 1));
}
 

@Test
public void testDeleteSecretNotInTable() {
    QueryRequest request = constructQueryRequest(SECRET_NAME);
    QueryResult result = new QueryResult().withCount(0).withItems(new ArrayList<>());
    when(mockDynamoDBClient.query(request)).thenReturn(result);

    // Try deleting the secret. Method should complete without throwing exception but deleteItem will not be
    // called for any secrets.
    dynamoDB.delete(new SecretIdentifier(SECRET_NAME));
    verify(mockDynamoDBClient, times(1)).query(request);
    verify(mockDynamoDBClient, never()).deleteItem(tableName, constructKey(SECRET_NAME, 1));
}
 

@Override
protected PageResults<Map<String, AttributeValue>> fetchPage(RequestLimit lim) {
  // Read from DynamoDB
  RetryResult<QueryResult> retryResult = context.getClient().queryTable(tableName, context
      .getSplit().getFilterPushdown(), lastEvaluatedKey, lim.items, context.getReporter());

  QueryResult result = retryResult.result;
  int retries = retryResult.retries;

  return new PageResults<>(result.getItems(), result.getLastEvaluatedKey(), result
      .getConsumedCapacity().getCapacityUnits(), retries);
}
 

private static void findRepliesForAThreadSpecifyOptionalLimit(String forumName, String threadSubject) {

        Map<String, AttributeValue> lastEvaluatedKey = null;
        do {
            QueryRequest queryRequest = new QueryRequest().withTableName(tableName)
                .withKeyConditions(makeReplyKeyConditions(forumName, threadSubject)).withLimit(1)
                .withExclusiveStartKey(lastEvaluatedKey);

            QueryResult result = client.query(queryRequest);
            for (Map<String, AttributeValue> item : result.getItems()) {
                printItem(item);
            }
            lastEvaluatedKey = result.getLastEvaluatedKey();
        } while (lastEvaluatedKey != null);
    }
 

@Override
public Map<StaticBuffer, EntryList> getSlice(final List<StaticBuffer> keys, final SliceQuery query, final StoreTransaction txh) throws BackendException {
    log.debug("Entering getSliceMultiSliceQuery table:{} keys:{} query:{} txh:{}",
              getTableName(),
              encodeForLog(keys),
              encodeForLog(query),
              txh);

    final Map<StaticBuffer, EntryList> resultMap = Maps.newHashMapWithExpectedSize(keys.size());

    final List<QueryWorker> queryWorkers = Lists.newLinkedList();
    for (StaticBuffer hashKey : keys) {
        final QueryWorker queryWorker = buildQueryWorker(hashKey, query);
        queryWorkers.add(queryWorker);

        resultMap.put(hashKey, EntryList.EMPTY_LIST);
    }

    final List<QueryResultWrapper> results = client.getDelegate().parallelQuery(queryWorkers);
    for (QueryResultWrapper resultWrapper : results) {
        final StaticBuffer titanKey = resultWrapper.getTitanKey();

        final QueryResult dynamoDBResult = resultWrapper.getDynamoDBResult();
        final EntryList entryList = createEntryListFromItems(dynamoDBResult.getItems(), query);
        resultMap.put(titanKey, entryList);
    }

    log.debug("Exiting getSliceMultiSliceQuery table:{} keys:{} query:{} txh:{} returning:{}",
              getTableName(),
              encodeForLog(keys),
              encodeForLog(query),
              txh,
              resultMap.size());
    return resultMap;
}
 

@Override
protected QueryResultWrapper getMergedPages() {
    final QueryResult mergedDynamoResult = new QueryResult().withItems(getFinalItemList())
                                                      .withCount(returnedCount)
                                                      .withScannedCount(scannedCount)
                                                      .withConsumedCapacity(new ConsumedCapacity()
                                                                                    .withTableName(request.getTableName())
                                                                                    .withCapacityUnits(totalCapacityUnits));
    return new QueryResultWrapper(titanKey, mergedDynamoResult);
}
 

private StaticRecordIterator buildRecordIteratorFromQueryResult(final QueryResult queryResult) {
    final List<Entry> entries = Lists.newLinkedList();
    for (Map<String, AttributeValue> item : queryResult.getItems()) {
        // DynamoDB's between includes the end of the range, but Titan's slice queries expect the end key to be exclusive
        final Entry entry = new EntryBuilder(item).slice(rangeKeySliceQuery.getSliceStart(), rangeKeySliceQuery.getSliceEnd())
                                                  .build();
        if (entry != null) {
            entries.add(entry);
        }
    }
    return new StaticRecordIterator(entries);
}
 

@Override
  public List<DColumn> getColumns(String storeName, String rowKey, String startColumn, String endColumn, int count) {
  	Timer t = new Timer();
  	String key = storeName + "_" + rowKey;
  	HashMap<String,Condition> keyConditions = new HashMap<String,Condition>();
  	keyConditions.put("key", new Condition()
	.withComparisonOperator(ComparisonOperator.EQ)
	.withAttributeValueList(new AttributeValue().withS(key)));
  	if(startColumn != null && endColumn != null) {
      	keyConditions.put("column", new Condition()
  			.withComparisonOperator(ComparisonOperator.BETWEEN)
  			.withAttributeValueList(new AttributeValue().withS(startColumn), new AttributeValue(endColumn)));
  	} else if(startColumn != null) {
      	keyConditions.put("column", new Condition()
  			.withComparisonOperator(ComparisonOperator.GE)
  			.withAttributeValueList(new AttributeValue().withS(startColumn)));
  	} else if(endColumn != null) {
      	keyConditions.put("column", new Condition()
  			.withComparisonOperator(ComparisonOperator.LT)
  			.withAttributeValueList(new AttributeValue().withS(endColumn)));
  	}

QueryRequest request = new QueryRequest()
	.withTableName(getTenant().getName())
	.withLimit(Math.min(100,  count))
	.withKeyConditions(keyConditions);
QueryResult result = m_client.query(request);
List<DColumn> list = fromItems(result.getItems());
      m_logger.debug("get columns range for {} in {}", getTenant().getName(), t);
  	return list;
  }
 

@Override
public boolean delete(final MetadataQuery metadata) {
  // the nature of metadata deleter is that primary ID is always
  // well-defined and it is deleting a single entry at a time
  final String tableName = operations.getMetadataTableName(metadataType);
  final QueryRequest queryRequest = new QueryRequest(tableName);

  if (metadata.hasSecondaryId()) {
    queryRequest.withFilterExpression(
        DynamoDBOperations.METADATA_SECONDARY_ID_KEY
            + " = :secVal").addExpressionAttributeValuesEntry(
                ":secVal",
                new AttributeValue().withB(ByteBuffer.wrap(metadata.getSecondaryId())));
  }
  queryRequest.withKeyConditionExpression(
      DynamoDBOperations.METADATA_PRIMARY_ID_KEY
          + " = :priVal").addExpressionAttributeValuesEntry(
              ":priVal",
              new AttributeValue().withB(ByteBuffer.wrap(metadata.getPrimaryId())));

  final QueryResult queryResult = operations.getClient().query(queryRequest);
  for (final Map<String, AttributeValue> entry : queryResult.getItems()) {
    final Map<String, AttributeValue> key = new HashMap<>();
    key.put(
        DynamoDBOperations.METADATA_PRIMARY_ID_KEY,
        entry.get(DynamoDBOperations.METADATA_PRIMARY_ID_KEY));
    key.put(
        DynamoDBOperations.METADATA_TIMESTAMP_KEY,
        entry.get(DynamoDBOperations.METADATA_TIMESTAMP_KEY));
    operations.getClient().deleteItem(tableName, key);
  }

  return true;
}
 

public LazyPaginatedQuery(
    final QueryResult currentResult,
    final QueryRequest request,
    final AmazonDynamoDBAsync dynamoDBClient) {
  this.currentResult = currentResult;
  this.request = request;
  this.dynamoDBClient = dynamoDBClient;
}
 

/**
 * Get the next query data If the last request is equal to null then we have no more query
 * requests to fire
 *
 * <p> If asyncQueryResults is not empty, we have already fetched the next query data that can be
 * read immediately
 *
 * <p> If due to max async query limit, we couldn't fire async requests, we fire the request now
 */
@Override
protected Iterator<? extends Map<String, AttributeValue>> nextIterator(final int arg0) {

  synchronized (monitorLock) {
    if ((lastRequest == null) && asyncQueryResults.isEmpty()) {
      return null;
    }

    QueryResult result = null;
    if ((lastRequest != null) && (asyncRequestsInProgress == 0)) {
      makeAsyncQuery();
    }

    while (asyncQueryResults.isEmpty()) {
      try {
        monitorLock.wait();
      } catch (final InterruptedException e) {
        LOGGER.error("Exception in Async paginated query " + e);
        e.printStackTrace();
      }
    }
    result = asyncQueryResults.remove();

    return result == null ? null : result.getItems().iterator();
  }
}
 

@Override
public <T> PaginatedQueryList<T> query(Class<T> clazz,
		QueryRequest queryRequest) {
	QueryResult queryResult = amazonDynamoDB.query(queryRequest);
	return new PaginatedQueryList<T>(dynamoDBMapper, clazz, amazonDynamoDB, queryRequest, queryResult,
			dynamoDBMapperConfig.getPaginationLoadingStrategy(), dynamoDBMapperConfig);
}
 

private static void findRepliesPostedWithinTimePeriod(String forumName, String threadSubject) {
    
    long startDateMilli = (new Date()).getTime() - (15L*24L*60L*60L*1000L); 
    long endDateMilli = (new Date()).getTime() - (5L*24L*60L*60L*1000L);    
    java.text.SimpleDateFormat df = new java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
    String startDate = df.format(startDateMilli);
    String endDate = df.format(endDateMilli);

    Condition rangeKeyCondition = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
            new AttributeValue().withS(startDate), 
            new AttributeValue().withS(endDate));
    
    Map<String, Condition> keyConditions = makeReplyKeyConditions(forumName, threadSubject);
    keyConditions.put("ReplyDateTime", rangeKeyCondition);
    
    QueryRequest queryRequest = new QueryRequest()
        .withTableName(tableName)
        .withKeyConditions(keyConditions)
        .withProjectionExpression("Message, ReplyDateTime, PostedBy");

    QueryResult result = client.query(queryRequest);
    for (Map<String, AttributeValue> item : result.getItems()) {
        printItem(item);
    }        
}
 

/**
 * Query Amazon DynamoDB
 * 
 * @param hashKey
 *            Hash key for the query request.
 * 
 * @param range
 *            The range of geohashs to query.
 * 
 * @return The query result.
 */
public List<QueryResult> queryGeohash(QueryRequest queryRequest, long hashKey, GeohashRange range) {
	List<QueryResult> queryResults = new ArrayList<QueryResult>();
	Map<String, AttributeValue> lastEvaluatedKey = null;

	do {
		Map<String, Condition> keyConditions = new HashMap<String, Condition>();

		Condition hashKeyCondition = new Condition().withComparisonOperator(ComparisonOperator.EQ)
				.withAttributeValueList(new AttributeValue().withN(String.valueOf(hashKey)));
		keyConditions.put(config.getHashKeyAttributeName(), hashKeyCondition);

		AttributeValue minRange = new AttributeValue().withN(Long.toString(range.getRangeMin()));
		AttributeValue maxRange = new AttributeValue().withN(Long.toString(range.getRangeMax()));

		Condition geohashCondition = new Condition().withComparisonOperator(ComparisonOperator.BETWEEN)
				.withAttributeValueList(minRange, maxRange);
		keyConditions.put(config.getGeohashAttributeName(), geohashCondition);

		queryRequest.withTableName(config.getTableName()).withKeyConditions(keyConditions)
				.withIndexName(config.getGeohashIndexName()).withConsistentRead(true)
				.withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL).withExclusiveStartKey(lastEvaluatedKey);

		QueryResult queryResult = config.getDynamoDBClient().query(queryRequest);
		queryResults.add(queryResult);

		lastEvaluatedKey = queryResult.getLastEvaluatedKey();

	} while (lastEvaluatedKey != null);

	return queryResults;
}