Java 类com.amazonaws.services.dynamodbv2.model.ScanRequest 实例源码

/**
 * Gets all.
 *
 * @return the all
 */
public Collection<Ticket> getAll() {
    final Collection<Ticket> tickets = new ArrayList<>();
    final Collection<TicketDefinition> metadata = this.ticketCatalog.findAll();
    metadata.forEach(r -> {
        final ScanRequest scan = new ScanRequest(r.getProperties().getStorageName());
        LOGGER.debug("Scanning table with request [{}]", scan);
        final ScanResult result = this.amazonDynamoDBClient.scan(scan);
        LOGGER.debug("Scanned table with result [{}]", scan);

        tickets.addAll(result.getItems()
                .stream()
                .map(DynamoDbTicketRegistryFacilitator::deserializeTicket)
                .collect(Collectors.toList()));
    });
    return tickets;
}

@Override
public PropertySource<?> locate(final Environment environment) {
    final AmazonDynamoDBClient amazonDynamoDBClient = getAmazonDynamoDbClient(environment);
    createSettingsTable(amazonDynamoDBClient, false);

    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}]", scan);
    final ScanResult result = amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);

    final Properties props = new Properties();
    result.getItems()
            .stream()
            .map(DynamoDbCloudConfigBootstrapConfiguration::retrieveSetting)
            .forEach(p -> props.put(p.getKey(), p.getValue()));
    return new PropertiesPropertySource(getClass().getSimpleName(), props);
}

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 ScanResult scan(final ScanRequest request, final int permitsToConsume) throws BackendException {
    setUserAgent(request);
    ScanResult result;
    timedReadThrottle(SCAN, request.getTableName(), permitsToConsume);

    final Timer.Context apiTimerContext = getTimerContext(SCAN, request.getTableName());
    try {
        result = client.scan(request);
    } catch (Exception e) {
        throw processDynamoDbApiException(e, SCAN, request.getTableName());
    } finally {
        apiTimerContext.stop();
    }
    meterConsumedCapacity(SCAN, result.getConsumedCapacity());
    measureItemCount(SCAN, request.getTableName(), result.getCount());
    return result;
}

@Override
public KeyIterator getKeys(final SliceQuery query, final StoreTransaction txh) throws BackendException {
    log.debug("Entering getKeys table:{} query:{} txh:{}", getTableName(), encodeForLog(query), txh);

    final ScanRequest scanRequest = super.createScanRequest();

    final Scanner scanner;
    if (client.isEnableParallelScan()) {
        scanner = client.getDelegate().getParallelScanCompletionService(scanRequest);
    } else {
        scanner = new SequentialScanner(client.getDelegate(), scanRequest);
    }
    // Because SINGLE records cannot be split across scan results, we can use the same interpreter for both
    // sequential and parallel scans.
    final KeyIterator result = new ScanBackedKeyIterator(scanner, new SingleRowScanInterpreter(query));

    log.debug("Exiting getKeys table:{} query:{} txh:{} returning:{}", getTableName(), encodeForLog(query), txh, result);
    return result;
}

@Override
public KeyIterator getKeys(final SliceQuery query, final StoreTransaction txh) throws BackendException {
    log.debug("Entering getKeys table:{} query:{} txh:{}", getTableName(), encodeForLog(query), txh);
    final Expression filterExpression = new FilterExpressionBuilder().rangeKey()
                                                                     .range(query)
                                                                     .build();

    final ScanRequest scanRequest = super.createScanRequest()
             .withFilterExpression(filterExpression.getConditionExpression())
             .withExpressionAttributeValues(filterExpression.getAttributeValues());

    final Scanner scanner;
    final ScanContextInterpreter interpreter;
    if (client.isEnableParallelScan()) {
        scanner = client.getDelegate().getParallelScanCompletionService(scanRequest);
        interpreter = new MultiRowParallelScanInterpreter(this, query);
    } else {
        scanner = new SequentialScanner(client.getDelegate(), scanRequest);
        interpreter = new MultiRowSequentialScanInterpreter(this, query);
    }

    final KeyIterator result = new ScanBackedKeyIterator(scanner, interpreter);
    log.debug("Exiting getKeys table:{} query:{} txh:{} returning:{}", getTableName(), encodeForLog(query), txh, result);
    return result;
}

/**
 * This method gets a segmentedScanResult and submits the next scan request for that segment, if there is one.
 * @return the next available ScanResult
 * @throws ExecutionException if one of the segment pages threw while executing
 * @throws InterruptedException if one of the segment pages was interrupted while executing.
 */
private ScanContext grab() throws ExecutionException, InterruptedException {
    final Future<ScanContext> ret = exec.take();

    final ScanRequest originalRequest = ret.get().getScanRequest();
    final int segment = originalRequest.getSegment();

    final ScanSegmentWorker sw = workers[segment];

    if (sw.hasNext()) {
        currentFutures[segment] = exec.submit(sw);
    } else {
        finishSegment(segment);
        currentFutures[segment] = null;
    }

    return ret.get(); //This might block if nothing is available.
}

/**
 * Returns the list of usernames stored in the identity table.
 * 
 * @return list of existing usernames in DynamoDB table
 */
public List<String> listUsers() {
    List<String> users = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(USER_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        users.add(s);
    }

    return users;
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

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

/**
 * This function copies a scan request for the number of segments and then
 * adds those workers to the executor service to begin scanning.
 * 
 * @param totalSections
 * @param section
 * 
 * @return <ParallelScanExecutor> the parallel scan executor to grab results
 *         when a segment is finished.
 */
public ParallelScanExecutor getParallelScanCompletionService(
        ScanRequest initialRequest, int numSegments, Executor executor,
        int section, int totalSections) {
    final int segments = Math.max(1, numSegments);
    final ParallelScanExecutor completion = new ParallelScanExecutor(
            executor, segments);

    int sectionSize = segments / totalSections;
    int start = sectionSize * section;
    int end = start + sectionSize;
    if (section + 1 == totalSections) {
        end = segments;
    }

    for (int segment = start; segment < end; segment++) {
        ScanRequest scanSegment = copyScanRequest(initialRequest)
                .withTotalSegments(segments).withSegment(segment);
        completion.addWorker(new ScanSegmentWorker(this.client,
                this.rateLimiter, scanSegment), segment);
    }

    return completion;
}

@Override
public List<LandUnit> getLandUnits(Long userId) {
    List<LandUnit> retval = new ArrayList<LandUnit>();
    try {
        /*
         * Scan items for movies with user id attribute.
         */
        Map<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withN(userId.toString()));
        scanFilter.put(LandUnit.USER_ID_ATTR_NAME, condition);
        ScanRequest scanRequest =
                new ScanRequest(LANDUNIT_DYNAMO_DB_TABLE_NAME).withScanFilter(scanFilter);
        ScanResult scanResult = dynamoDB.scan(scanRequest);
        LOG.debug("DDB Scan Result: " + scanResult);
        retval = mapItemsToLandUnit(scanResult.getItems());
    } catch (Exception e) {
        LOG.error("Unable to retrieve land units from DDB " + e.getMessage());
    }
    return retval;
}

/**
 * Generate a list of attribute names found in the Aggregator's dynamo
 * table. Assumes that all Items in the Aggregator table are of the same
 * structure.
 * 
 * @param dynamoClient
 *            Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable
 *            The Dynamo Table for the Aggregator
 * @return A list of attribute names from the Dynamo table
 * @throws Exception
 */
public static List<String> getDictionaryEntry(
        final AmazonDynamoDB dynamoClient, final String dynamoTable)
        throws Exception {
    // get a list of all columns in the table, with keys first
    List<String> columns = new ArrayList<>();
    List<KeySchemaElement> keys = dynamoClient.describeTable(dynamoTable)
            .getTable().getKeySchema();
    for (KeySchemaElement key : keys) {
        columns.add(key.getAttributeName());
    }
    ScanResult scan = dynamoClient.scan(new ScanRequest()
            .withTableName(dynamoTable).withSelect(Select.ALL_ATTRIBUTES)
            .withLimit(1));
    List<Map<String, AttributeValue>> scannedItems = scan.getItems();
    for (Map<String, AttributeValue> map : scannedItems) {
        for (String s : map.keySet()) {
            if (!columns.contains(s))
                columns.add(s);
        }
    }

    return columns;
}

/**
 * Generate a list of attribute names found in the Aggregator's dynamo
 * table. Assumes that all Items in the Aggregator table are of the same
 * structure.
 * 
 * @param dynamoClient Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable The Dynamo Table for the Aggregator
 * @return A list of attribute names from the Dynamo table
 * @throws Exception
 */
protected List<String> getDictionaryEntry() throws Exception {
    // get a list of all columns in the table, with keys first
    List<String> columns = new ArrayList<>();
    List<KeySchemaElement> keys = dynamoClient.describeTable(this.tableName).getTable().getKeySchema();
    for (KeySchemaElement key : keys) {
        columns.add(key.getAttributeName());
    }
    ScanResult scan = dynamoClient.scan(new ScanRequest().withTableName(this.tableName).withSelect(
            Select.ALL_ATTRIBUTES).withLimit(1));
    List<Map<String, AttributeValue>> scannedItems = scan.getItems();
    for (Map<String, AttributeValue> map : scannedItems) {
        for (String s : map.keySet()) {
            if (!columns.contains(s))
                columns.add(s);
        }
    }

    return columns;
}

private void dumpTables() {
    for (String table : client.listTables().getTableNames()) {
        ScanResult scanResult;
        Map<String, AttributeValue> lastKey = null;
        do {
            scanResult = client.scan(new ScanRequest().withTableName(table).withExclusiveStartKey(lastKey));
            lastKey = scanResult.getLastEvaluatedKey();
            for (Map<String, AttributeValue> map : scanResult.getItems()) {
                for (Map.Entry<String, AttributeValue> item : map.entrySet()) {
                    System.out.print("item.put(\"");
                    System.out.print(item.getKey());
                    System.out.print("\", b642Av(\"");
                    System.out.print(Base64.encodeAsString(AttributeValueMarshaller.marshall(item.getValue()).array()));
                    System.out.println("\"));");
                }
                System.out.print("ddb.putItem(new PutItemRequest(\"");
                System.out.print(table);
                System.out.println("\", item));");
                System.out.println("item.clear();");
                System.out.println();
            }
        } while (lastKey != null);

    }
}

@Override
public List<LandUnit> getLandUnits(Long userId) {
    List<LandUnit> retval = new ArrayList<LandUnit>();
    try {
        /*
         * Scan items for movies with user id attribute.
         */
        Map<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withN(userId.toString()));
        scanFilter.put(LandUnit.USER_ID_ATTR_NAME, condition);
        ScanRequest scanRequest =
                new ScanRequest(LANDUNIT_DYNAMO_DB_TABLE_NAME).withScanFilter(scanFilter);
        ScanResult scanResult = dynamoDB.scan(scanRequest);
        LOG.debug("DDB Scan Result: " + scanResult);
        retval = mapItemsToLandUnit(scanResult.getItems());
    } catch (Exception e) {
        LOG.error("Unable to retrieve land units from DDB " + e.getMessage());
    }
    return retval;
}

private static void findProductsForPriceLessThanZero() {

    Map<String, AttributeValue> expressionAttributeValues = new HashMap<String, AttributeValue>();
    expressionAttributeValues.put(":pr", new AttributeValue().withN("100"));

    ScanRequest scanRequest = new ScanRequest()
        .withTableName(tableName)
        .withFilterExpression("Price < :pr")
        .withExpressionAttributeValues(expressionAttributeValues)
        .withProjectionExpression("Id, Title, ProductCategory, Price");

    ScanResult result = client.scan(scanRequest);

    System.out.println("Scan of " + tableName + " for items with a price less than 100.");
    for (Map<String, AttributeValue> item : result.getItems()) {
        System.out.println("");
        printItem(item);
    }
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

/**
 * Returns the list of usernames stored in the identity table.
 * 
 * @return list of existing usernames in DynamoDB table
 */
public List<String> listUsers() {
    List<String> users = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(USER_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        users.add(s);
    }

    return users;
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

public ScanResult getSince(long timestamp, int limit) throws Exception {
    try {
        Condition scanFilterCondition = new Condition()
                .withComparisonOperator(ComparisonOperator.GT.toString())
                .withAttributeValueList(
                        new AttributeValue().withN(Long.toString(timestamp)));
        Map<String, Condition> conditions = new HashMap<String, Condition>();
        conditions.put(COL_CREATEDAT, scanFilterCondition);

        ScanRequest scanRequest = new ScanRequest()
                .withTableName(TABLE_NAME)
                .withScanFilter(conditions)
                .withLimit(limit)
                .withAttributesToGet(
                        Arrays.asList(COL_ID, COL_CREATEDAT, COL_LAT,
                                COL_LONG, COL_SCREENNAME, COL_TEXT));

        return dynamoDB.scan(scanRequest);
    } catch (Exception e) {
        handleException(e);
    }

    return null;
}

private long getDynamoItemCount() {

        ScanResult scanResult = dynamoClient.scan(TEST_DYNAMO_TABLE_NAME, Lists.newArrayList("Uuid"));
        long itemCount = scanResult.getCount();

        while(scanResult.getLastEvaluatedKey() != null && !scanResult.getLastEvaluatedKey().isEmpty()) {
            scanResult = dynamoClient.scan(new ScanRequest().withTableName(TEST_DYNAMO_TABLE_NAME).withAttributesToGet("Uuid").withExclusiveStartKey(scanResult.getLastEvaluatedKey()));
            itemCount += scanResult.getCount();
        }

        return itemCount;
    }

/**
 * Delete all.
 *
 * @return the int
 */
public int deleteAll() {
    final AtomicInteger count = new AtomicInteger();
    final Collection<TicketDefinition> metadata = this.ticketCatalog.findAll();
    metadata.forEach(r -> {
        final ScanRequest scan = new ScanRequest(r.getProperties().getStorageName());
        LOGGER.debug("Submitting scan request [{}] to table [{}]", scan, r.getProperties().getStorageName());
        count.addAndGet(this.amazonDynamoDBClient.scan(scan).getCount());
    });
    createTicketTables(true);
    return count.get();
}

/**
 * Count long.
 *
 * @return the long
 */
public long count() {
    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}] to count items", scan);
    final ScanResult result = this.amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);
    return result.getCount();
}

/**
 * Gets all.
 *
 * @return the all
 */
public List<RegisteredService> getAll() {
    final List<RegisteredService> services = new ArrayList<>();
    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}]", scan);
    final ScanResult result = this.amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);

    services.addAll(result.getItems()
            .stream()
            .map(this::deserializeServiceFromBinaryBlob)
            .sorted((o1, o2) -> Integer.valueOf(o1.getEvaluationOrder()).compareTo(o2.getEvaluationOrder()))
            .collect(Collectors.toList()));
    return services;
}

private Stream<Entry> scan(SecretEventStream.Filter<Entry> filter, Converters converters) {
    ScanRequest scanRequest = new ScanRequest();
    scanRequest.withConsistentRead(true);
    scanRequest.withTableName(tableName);

    FilterGenerator filterGenerator = new FilterGenerator();
    FilterGenerator.Filter generated = filterGenerator.process(filter.parsedAttributeCondition.get(), converters);

    if(!generated.expressionAttributeNames.isEmpty()) {
        scanRequest.withExpressionAttributeNames(generated.expressionAttributeNames);
    }

    if (!generated.expressionAttributeValues.isEmpty()) {
        scanRequest.withExpressionAttributeValues(generated.expressionAttributeValues);
    }

    scanRequest.withFilterExpression(generated.filterExpression);

    ScanResult result = client.scan(scanRequest);

    List<Map<String, AttributeValue>> results = new ArrayList<>();
    results.addAll(result.getItems());

    while (result.getLastEvaluatedKey() != null) {
        scanRequest = scanRequest.withExclusiveStartKey(result.getLastEvaluatedKey());

        result = client.scan(scanRequest);

        results.addAll(result.getItems());
    }

    Stream<Entry> typedResult = results.stream().map(this::fromMap);

    if (filter.reverse) {
        typedResult = Lists.reverse(typedResult.collect(Collectors.toCollection(LinkedList::new))).stream();
    }

    return typedResult;
}

@Test
public void testKeySet() throws Exception {
    ScanRequest request = new ScanRequest().withConsistentRead(true).withTableName(tableName);
    ScanResult result = constructScanResult();
    when(mockDynamoDBClient.scan(request)).thenReturn(result);

    // Call the KeySet method and assert the expected secret identifiers are returned.
    Set<SecretIdentifier> keys = dynamoDB.keySet();
    assertEquals(keys.size(), 2);
    assertTrue(keys.contains(new SecretIdentifier(SECRET_NAME)));
    assertTrue(keys.contains(new SecretIdentifier(SECRET2_NAME)));

    verify(mockDynamoDBClient, times(1)).scan(request);
}

@Test
public void testKeySetEmpty() throws Exception {
    ScanRequest request = new ScanRequest().withConsistentRead(true).withTableName(tableName);
    ScanResult result = new ScanResult().withCount(0).withItems(new ArrayList<>());
    when(mockDynamoDBClient.scan(request)).thenReturn(result);

    // Call the KeySet method and check that is it empty.
    Set<SecretIdentifier> keySet = dynamoDB.keySet();
    assertTrue(keySet.isEmpty());
    verify(mockDynamoDBClient, times(1)).scan(request);
}

public CompletableFuture<Stream<Map<String, Object>>> scan(final ScanRequest scanRequest) {
    return asyncExecutor.execute(new Callable<Stream<Map<String, Object>>>() {
        @Override
        public Stream<Map<String, Object>> call() throws Exception {
            return dbExecutor.scan(scanRequest);
        }
    });
}

public <T> CompletableFuture<Stream<T>> scan(final Class<T> targetClass, final ScanRequest scanRequest) {
    return asyncExecutor.execute(new Callable<Stream<T>>() {
        @Override
        public Stream<T> call() throws Exception {
            return dbExecutor.scan(targetClass, scanRequest);
        }
    });
}

@Override
public GatewayResponse handleRequest(GatewayRequest request, Context context) {
    ScanRequest scanRequest = new ScanRequest().withTableName("Reply");
    String result = DynamoDBUtil.getClient().scan(scanRequest).getItems().toString();

    return new GatewayResponse(200, result, GatewayResponse.HEADERS_JSON);
}

@Override
public String handleRequest(Book request, Context context) {
    ScanRequest scanRequest = new ScanRequest().withTableName("Books");
    ScanResult result = DynamoDBUtil.getClient().scan(scanRequest);
    System.out.println("-- books listing start --");
    for (Map<String, AttributeValue> item : result.getItems()){
        System.out.println(item);
    }
    System.out.println("-- books listing end --");
    return result.getItems().toString();
}

@Override
public void execute() {
    ScanResult result = ddbClient.scan(new ScanRequest()
            .withTableName(determineTableName())
            .withScanFilter(determineScanFilter()));

    Map tmp = new HashMap<>();
    tmp.put(DdbConstants.ITEMS, result.getItems());
    tmp.put(DdbConstants.LAST_EVALUATED_KEY, result.getLastEvaluatedKey());
    tmp.put(DdbConstants.CONSUMED_CAPACITY, result.getConsumedCapacity());
    tmp.put(DdbConstants.COUNT, result.getCount());
    tmp.put(DdbConstants.SCANNED_COUNT, result.getScannedCount());
    addToResults(tmp);
}

@SuppressWarnings("unchecked")
@Override
public ScanResult scan(ScanRequest scanRequest) {
    this.scanRequest = scanRequest;
    ConsumedCapacity consumed = new ConsumedCapacity();
    consumed.setCapacityUnits(1.0);
    Map<String, AttributeValue> lastEvaluatedKey = new HashMap<String, AttributeValue>();
    lastEvaluatedKey.put("1", new AttributeValue("LAST_KEY"));
    return new ScanResult()
            .withConsumedCapacity(consumed)
            .withCount(1)
            .withItems(getAttributes())
            .withScannedCount(10)
            .withLastEvaluatedKey(lastEvaluatedKey);
}

ParallelScanner getParallelScanCompletionService(final ScanRequest initialRequest) throws BackendException {
    final int segments = Math.max(1, clientThreadPool.getMaximumPoolSize() / maxConcurrentUsers);
    final ParallelScanner completion = new ParallelScanner(clientThreadPool, segments, this);

    for (int segment = 0; segment < segments; segment++) {
        // dont need to set user agent here because ExponentialBackoff.Scan
        // calls DynamoDbDelegate.scan which sets it
        final ScanRequest scanSegment = copyScanRequest(initialRequest).withTotalSegments(segments).withSegment(segment);
        completion.addWorker(new ScanSegmentWorker(this, scanSegment), segment);
    }

    return completion;
}

public static ScanRequest copyScanRequest(final ScanRequest request) {
    return new ScanRequest().withAttributesToGet(request.getAttributesToGet())
        .withScanFilter(request.getScanFilter())
        .withConditionalOperator(request.getConditionalOperator())
        .withExclusiveStartKey(request.getExclusiveStartKey())
        .withExpressionAttributeNames(request.getExpressionAttributeNames())
        .withExpressionAttributeValues(cloneItem(request.getExpressionAttributeValues()))
        .withFilterExpression(request.getFilterExpression())
        .withIndexName(request.getIndexName()).withLimit(request.getLimit())
        .withProjectionExpression(request.getProjectionExpression())
        .withReturnConsumedCapacity(request.getReturnConsumedCapacity())
        .withScanFilter(request.getScanFilter()).withSelect(request.getSelect())
        .withTableName(request.getTableName()).withTotalSegments(request.getTotalSegments())
        .withSegment(request.getSegment());
}

@Override
public ScanContext call() throws Exception {
    try {
        final ScanRequest originalRequest = DynamoDbDelegate.copyScanRequest(request);
        final ScanResult result = next();
        return new ScanContext(originalRequest, result);
    } catch (BackendRuntimeException e) {
        throw e.getBackendException();
    }
}

public SequentialScanner(final DynamoDbDelegate dynamoDbDelegate, final ScanRequest request) {
    this.dynamoDbDelegate = dynamoDbDelegate;
    Preconditions.checkArgument(request.getExclusiveStartKey() == null || request.getExclusiveStartKey().isEmpty(),
                                "A scan worker should start with a fresh ScanRequest");
    this.request = DynamoDbDelegate.copyScanRequest(request);
    this.lastConsumedCapacity = dynamoDbDelegate.estimateCapacityUnits(DynamoDbDelegate.SCAN, request.getTableName());
    this.currentFuture = dynamoDbDelegate.scanAsync(request, lastConsumedCapacity);
}

ScanSegmentWorker(final AmazonDynamoDBClient client,
        final RateLimiter rateLimiter, ScanRequest request) {
    this.request = request;
    this.client = client;
    this.rateLimiter = rateLimiter;
    this.hasNext = true;
    this.exponentialBackoffTime = BootstrapConstants.INITIAL_RETRY_TIME_MILLISECONDS;
    lastConsumedCapacity = 256;
}

public ScanRequest copyScanRequest(ScanRequest request) {
    return new ScanRequest()
            .withTableName(request.getTableName())
            .withTotalSegments(request.getTotalSegments())
            .withSegment(request.getSegment())
            .withReturnConsumedCapacity(request.getReturnConsumedCapacity())
            .withLimit(request.getLimit())
            .withConsistentRead(request.getConsistentRead());
}