public RetryResult<ScanResult> scanTable(
    String tableName, DynamoDBQueryFilter dynamoDBQueryFilter, Integer segment, Integer
    totalSegments, Map<String, AttributeValue> exclusiveStartKey, long limit, Reporter reporter) {
  final ScanRequest scanRequest = new ScanRequest(tableName)
      .withExclusiveStartKey(exclusiveStartKey)
      .withLimit(Ints.checkedCast(limit))
      .withSegment(segment)
      .withTotalSegments(totalSegments)
      .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

  if (dynamoDBQueryFilter != null) {
    Map<String, Condition> scanFilter = dynamoDBQueryFilter.getScanFilter();
    if (!scanFilter.isEmpty()) {
      scanRequest.setScanFilter(scanFilter);
    }
  }

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

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;
}
 

/**
 * Begins to pipe the log results by parallel scanning the table and the
 * consumer writing the results.
 */
public void pipe(final AbstractLogConsumer consumer)
        throws ExecutionException, InterruptedException {
    final DynamoDBTableScan scanner = new DynamoDBTableScan(rateLimit,
            client);

    final ScanRequest request = new ScanRequest().withTableName(tableName)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
            .withLimit(BootstrapConstants.SCAN_LIMIT)
            .withConsistentRead(consistentScan);

    final ParallelScanExecutor scanService = scanner
            .getParallelScanCompletionService(request, numSegments,
                    threadPool, section, totalSections);

    while (!scanService.finished()) {
        SegmentedScanResult result = scanService.grab();
        consumer.writeResult(result);
    }

    shutdown(true);
    consumer.shutdown(true);
}
 

/**
 * Splits up a ScanResult into a list of BatchWriteItemRequests of size 25
 * items or less each.
 */
public static List<BatchWriteItemRequest> splitResultIntoBatches(
        ScanResult result, String tableName) {
    List<BatchWriteItemRequest> batches = new LinkedList<BatchWriteItemRequest>();
    Iterator<Map<String, AttributeValue>> it = result.getItems().iterator();

    BatchWriteItemRequest req = new BatchWriteItemRequest()
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
    List<WriteRequest> writeRequests = new LinkedList<WriteRequest>();
    int i = 0;
    while (it.hasNext()) {
        PutRequest put = new PutRequest(it.next());
        writeRequests.add(new WriteRequest(put));

        i++;
        if (i == BootstrapConstants.MAX_BATCH_SIZE_WRITE_ITEM) {
            req.addRequestItemsEntry(tableName, writeRequests);
            batches.add(req);
            req = new BatchWriteItemRequest()
                    .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
            writeRequests = new LinkedList<WriteRequest>();
            i = 0;
        }
    }
    if (i > 0) {
        req.addRequestItemsEntry(tableName, writeRequests);
        batches.add(req);
    }
    return batches;
}
 

/**
 * 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;
}
 

protected Map<String, AttributeValue> getItem(String tableName, Map<String, AttributeValue> key) {
    GetItemResult result = dynamodb.getItem(new GetItemRequest()
        .withTableName(tableName)
        .withKey(key)
        .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
        .withConsistentRead(true));
    return result.getItem();
}
 

public static void query(String indexName) {

        System.out.println("\n***********************************************************\n");
        System.out.println("Querying table " + tableName + "...");

        QueryRequest queryRequest = new QueryRequest().withTableName(tableName).withConsistentRead(true)
            .withScanIndexForward(true).withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

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

        keyConditions.put("CustomerId", new Condition().withComparisonOperator(ComparisonOperator.EQ)
            .withAttributeValueList(new AttributeValue().withS("[email protected]")));

        if (indexName == "IsOpenIndex") {
            System.out.println("\nUsing index: '" + indexName + "': Bob's orders that are open.");
            System.out.println("Only a user-specified list of attributes are returned\n");
            queryRequest.setIndexName(indexName);

            keyConditions.put("IsOpen", new Condition().withComparisonOperator(ComparisonOperator.EQ)
                .withAttributeValueList(new AttributeValue().withN("1")));

            // ProjectionExpression
            queryRequest.setProjectionExpression("OrderCreationDate, ProductCategory, ProductName, OrderStatus");

        }
        else if (indexName == "OrderCreationDateIndex") {
            System.out.println("\nUsing index: '" + indexName + "': Bob's orders that were placed after 01/31/2013.");
            System.out.println("Only the projected attributes are returned\n");
            queryRequest.setIndexName(indexName);

            keyConditions.put("OrderCreationDate", new Condition().withComparisonOperator(ComparisonOperator.GT)
                .withAttributeValueList(new AttributeValue().withN("20130131")));

            // Select
            queryRequest.setSelect(Select.ALL_PROJECTED_ATTRIBUTES);

        }
        else {
            System.out.println("\nNo index: All of Bob's orders, by OrderId:\n");
        }

        queryRequest.setKeyConditions(keyConditions);

        QueryResult result = client.query(queryRequest);
        List<Map<String, AttributeValue>> items = result.getItems();
        Iterator<Map<String, AttributeValue>> itemsIter = items.iterator();
        while (itemsIter.hasNext()) {
            Map<String, AttributeValue> currentItem = itemsIter.next();

            Iterator<String> currentItemIter = currentItem.keySet().iterator();
            while (currentItemIter.hasNext()) {
                String attr = (String) currentItemIter.next();
                if (attr == "OrderId" || attr == "IsOpen" || attr == "OrderCreationDate") {
                    System.out.println(attr + "---> " + currentItem.get(attr).getN());
                }
                else {
                    System.out.println(attr + "---> " + currentItem.get(attr).getS());
                }
            }
            System.out.println();
        }
        System.out.println("\nConsumed capacity: " + result.getConsumedCapacity() + "\n");

    }
 

private static void writeMultipleItemsBatchWrite() {
    try {

        // Create a map for the requests in the batch
        Map<String, List<WriteRequest>> requestItems = new HashMap<String, List<WriteRequest>>();

        // Create a PutRequest for a new Forum item
        Map<String, AttributeValue> forumItem = new HashMap<String, AttributeValue>();
        forumItem.put("Name", new AttributeValue().withS("Amazon RDS"));
        forumItem.put("Threads", new AttributeValue().withN("0"));

        List<WriteRequest> forumList = new ArrayList<WriteRequest>();
        forumList.add(new WriteRequest().withPutRequest(new PutRequest().withItem(forumItem)));
        requestItems.put(table1Name, forumList);

        // Create a PutRequest for a new Thread item
        Map<String, AttributeValue> threadItem = new HashMap<String, AttributeValue>();
        threadItem.put("ForumName", new AttributeValue().withS("Amazon RDS"));
        threadItem.put("Subject", new AttributeValue().withS("Amazon RDS Thread 1"));
        threadItem.put("Message", new AttributeValue().withS("ElastiCache Thread 1 message"));
        threadItem.put("KeywordTags", new AttributeValue().withSS(Arrays.asList("cache", "in-memory")));

        List<WriteRequest> threadList = new ArrayList<WriteRequest>();
        threadList.add(new WriteRequest().withPutRequest(new PutRequest().withItem(threadItem)));

        // Create a DeleteRequest for a Thread item
        Map<String, AttributeValue> threadDeleteKey = new HashMap<String, AttributeValue>();
        threadDeleteKey.put("ForumName", new AttributeValue().withS("Amazon S3"));
        threadDeleteKey.put("Subject", new AttributeValue().withS("S3 Thread 100"));

        threadList.add(new WriteRequest().withDeleteRequest(new DeleteRequest().withKey(threadDeleteKey)));
        requestItems.put(table2Name, threadList);

        BatchWriteItemResult result;
        BatchWriteItemRequest batchWriteItemRequest = new BatchWriteItemRequest()
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

        do {
            System.out.println("Making the request.");

            batchWriteItemRequest.withRequestItems(requestItems);
            result = client.batchWriteItem(batchWriteItemRequest);

            // Print consumed capacity units
            for (ConsumedCapacity consumedCapacity : result.getConsumedCapacity()) {
                String tableName = consumedCapacity.getTableName();
                Double consumedCapacityUnits = consumedCapacity.getCapacityUnits();
                System.out.println("Consumed capacity units for table " + tableName + ": " + consumedCapacityUnits);
            }

            // Check for unprocessed keys which could happen if you exceed
            // provisioned throughput
            System.out.println("Unprocessed Put and Delete requests: \n" + result.getUnprocessedItems());
            requestItems = result.getUnprocessedItems();
        } while (result.getUnprocessedItems().size() > 0);

    }
    catch (AmazonServiceException ase) {
        System.err.println("Failed to retrieve items: ");
        ase.printStackTrace(System.err);
    }

}
 

protected UpdateItemRequest createUpdateItemRequest() {
    return new UpdateItemRequest()
            .withTableName(tableName)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
}
 

protected GetItemRequest createGetItemRequest() {
    return new GetItemRequest()
            .withTableName(tableName)
            .withConsistentRead(forceConsistentRead)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
}
 

protected DeleteItemRequest createDeleteItemRequest() {
    return new DeleteItemRequest()
            .withTableName(tableName)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
}
 

protected QueryRequest createQueryRequest() {
    return new QueryRequest()
            .withTableName(tableName)
            .withConsistentRead(forceConsistentRead)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
}
 

protected ScanRequest createScanRequest() {
    return new ScanRequest().withTableName(tableName)
            .withConsistentRead(forceConsistentRead)
            .withLimit(client.scanLimit(tableName))
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
}
 

public static void query(String indexName) {

        System.out.println("\n***********************************************************\n");
        System.out.println("Querying table " + tableName + "...");

        QueryRequest queryRequest = new QueryRequest()
            .withTableName(tableName)
            .withConsistentRead(true).withScanIndexForward(true)
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

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

        keyConditions.put(
            "CustomerId",
            new Condition()
                .withComparisonOperator(ComparisonOperator.EQ)
                .withAttributeValueList(new AttributeValue()
                    .withS("[email protected]")));

        if (indexName == "IsOpenIndex") {
            System.out.println("\nUsing index: '" + indexName
                    + "': Bob's orders that are open.");
            System.out.println("Only a user-specified list of attributes are returned\n");
            queryRequest.setIndexName(indexName);
            
            keyConditions.put(
                "IsOpen",
                new Condition()
                    .withComparisonOperator(ComparisonOperator.EQ)
                    .withAttributeValueList(new AttributeValue().withN("1")));
            
            // ProjectionExpression
            queryRequest.setProjectionExpression("OrderCreationDate, ProductCategory, ProductName, OrderStatus");

        } else if (indexName == "OrderCreationDateIndex") {
            System.out.println("\nUsing index: '" + indexName
                + "': Bob's orders that were placed after 01/31/2013.");
            System.out.println("Only the projected attributes are returned\n");
            queryRequest.setIndexName(indexName);
            
            keyConditions.put("OrderCreationDate", new Condition()
                .withComparisonOperator(ComparisonOperator.GT)
                .withAttributeValueList(new AttributeValue()
                    .withN("20130131")));
            
            // Select
            queryRequest.setSelect(Select.ALL_PROJECTED_ATTRIBUTES);
            
        } else {
            System.out.println("\nNo index: All of Bob's orders, by OrderId:\n");
        }

        queryRequest.setKeyConditions(keyConditions);

        QueryResult result = client.query(queryRequest);
        List<Map<String, AttributeValue>> items = result.getItems();
        Iterator<Map<String, AttributeValue>> itemsIter = items.iterator();
        while (itemsIter.hasNext()) {
            Map<String, AttributeValue> currentItem = itemsIter.next();
            
            Iterator<String> currentItemIter = currentItem.keySet().iterator();
            while (currentItemIter.hasNext()) {
                String attr = (String) currentItemIter.next();
                if (attr == "OrderId" || attr == "IsOpen"
                        || attr == "OrderCreationDate") {
                    System.out.println(attr + "---> "
                            + currentItem.get(attr).getN());
                } else {
                    System.out.println(attr + "---> "
                            + currentItem.get(attr).getS());
                }
            }
            System.out.println();    
        }
        System.out.println("\nConsumed capacity: " + result.getConsumedCapacity() + "\n");

    }
 

private static void writeMultipleItemsBatchWrite() {
    try {                    
        
       // Create a map for the requests in the batch
       Map<String, List<WriteRequest>> requestItems = new HashMap<String, List<WriteRequest>>();
        
       // Create a PutRequest for a new Forum item
       Map<String, AttributeValue> forumItem = new HashMap<String, AttributeValue>();
       forumItem.put("Name", new AttributeValue().withS("Amazon RDS"));
       forumItem.put("Threads", new AttributeValue().withN("0"));
    
       List<WriteRequest> forumList = new ArrayList<WriteRequest>();
       forumList.add(new WriteRequest().withPutRequest(new PutRequest().withItem(forumItem)));
       requestItems.put(table1Name, forumList);
       
       // Create a PutRequest for a new Thread item
       Map<String, AttributeValue> threadItem = new HashMap<String, AttributeValue>();
       threadItem.put("ForumName", new AttributeValue().withS("Amazon RDS"));
       threadItem.put("Subject", new AttributeValue().withS("Amazon RDS Thread 1"));
       threadItem.put("Message", new AttributeValue().withS("ElasticCache Thread 1 message"));
       threadItem.put("KeywordTags", new AttributeValue().withSS(Arrays.asList("cache", "in-memory")));
    
       List<WriteRequest> threadList = new ArrayList<WriteRequest>();
       threadList.add(new WriteRequest().withPutRequest(new PutRequest().withItem(threadItem)));
       
       // Create a DeleteRequest for a Thread item
       Map<String, AttributeValue> threadDeleteKey = new HashMap<String, AttributeValue>(); 
       threadDeleteKey.put("ForumName", new AttributeValue().withS("Amazon S3"));
       threadDeleteKey.put("Subject", new AttributeValue().withS("S3 Thread 100"));
       
       threadList.add(new WriteRequest().withDeleteRequest(new DeleteRequest().withKey(threadDeleteKey)));
       requestItems.put(table2Name, threadList);
       
       BatchWriteItemResult result;
       BatchWriteItemRequest batchWriteItemRequest = new BatchWriteItemRequest()
           .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
       
       do {
           System.out.println("Making the request.");
                           
           batchWriteItemRequest.withRequestItems(requestItems);
           result = client.batchWriteItem(batchWriteItemRequest);
           
           // Print consumed capacity units
           for(ConsumedCapacity consumedCapacity : result.getConsumedCapacity()) {
                String tableName = consumedCapacity.getTableName();
                Double consumedCapacityUnits = consumedCapacity.getCapacityUnits();
                System.out.println("Consumed capacity units for table " + tableName + ": " + consumedCapacityUnits);
           }
           
           // Check for unprocessed keys which could happen if you exceed provisioned throughput
           System.out.println("Unprocessed Put and Delete requests: \n" + result.getUnprocessedItems());
           requestItems = result.getUnprocessedItems();
       } while (result.getUnprocessedItems().size() > 0);
                    
    }  catch (AmazonServiceException ase) {
        System.err.println("Failed to retrieve items: ");
        ase.printStackTrace(System.err);
    }  

}
 

@JsonIgnore
public abstract void setReturnConsumedCapacity(ReturnConsumedCapacity returnConsumedCapacity);