Java 类com.amazonaws.services.dynamodb.model.AttributeValue 实例源码

@Override
public QueryResult call() throws Exception {
    QueryRequest query = new QueryRequest();
    query.setTableName(tableName);
    query.withHashKeyValue(new AttributeValue(normalize(path)));
    query.setConsistentRead(true);

    if(startKey != null) {
        query.setExclusiveStartKey(startKey);
    }

    if(log.isDebugEnabled()) {
        log.debug("Querying DynamoDB for path: " + path.toUri());
    }

    return db.query(query);
}

@Override
public UpdateItemResult call() throws Exception {
    UpdateItemRequest update = new UpdateItemRequest();
    update.setTableName(tableName);
    update.setKey(new Key(new AttributeValue(normalize(path.getParent())), new AttributeValue(path.getName())));

    Map<String, AttributeValueUpdate> items = new HashMap<String, AttributeValueUpdate>();
    items.put(DELETE_MARKER, new AttributeValueUpdate().withValue(new AttributeValue().withS(Boolean.TRUE.toString())));
    items.put(EPOCH_VALUE, new AttributeValueUpdate().withValue(new AttributeValue().withN(System.currentTimeMillis()+"")));

    update.setAttributeUpdates(items);

    if(log.isDebugEnabled()) {
        log.debug("Marking DynamoDB path deleted: " + path.toUri());
    }

    return db.updateItem(update);
}

/**
 * Builds query expression depending on query type (range or scan) 
 */
public void buildExpression(){
  AttributeValue hashAttrValue = buildKeyHashAttribute();
  if (hashAttrValue == null)
    throw new IllegalStateException("There is not a key schema defined.");
  if (DynamoDBQuery.getType().equals(RANGE_QUERY)){
    Condition newCondition = buildRangeCondition();
    buildQueryExpression(newCondition, hashAttrValue);
  }
  if (DynamoDBQuery.getType().equals(SCAN_QUERY))
    buildScanExpression(hashAttrValue);
}

/**
 * Builds scan query expression using a hash attribute value where to start
 * @param pHashAttrValueHash attribute value where to start scanning
 */
public void buildScanExpression(AttributeValue pHashAttrValue){
  DynamoDBScanExpression newScanExpression = new DynamoDBScanExpression();
  // TODO right now we only support scanning using the key, but we should support other types of scans
  newScanExpression.addFilterCondition(getKeySchema().getHashKeyElement().getAttributeName(), buildKeyScanCondition());
  dynamoDBExpression = newScanExpression;
}

/**
 * Builds range query expression
 * @param pNewConditionCondition for querying
 * @param pHashAttrValueHash attribute value where to start
 */
public void buildQueryExpression(Condition pNewCondition, AttributeValue pHashAttrValue) {
  DynamoDBQueryExpression newQueryExpression = new DynamoDBQueryExpression(pHashAttrValue); 
  newQueryExpression.setConsistentRead(getConsistencyReadLevel());
  newQueryExpression.setRangeKeyCondition(pNewCondition);
  dynamoDBExpression = newQueryExpression;
}

/**
 * Builds hash key attribute from generic query received
 * @returnAttributeValue build from query
 */
private AttributeValue buildKeyHashAttribute(){
  String pAttrType = getKeySchema().getHashKeyElement().getAttributeType();
  if(pAttrType.equals("S"))
    return new AttributeValue().withS(getHashKey(query.getKey()).toString());
  else if(pAttrType.equals("N"))
    return new AttributeValue().withN(getHashKey(query.getKey()).toString());
  return null;
}

/**
 * Builds query expression depending on query type (range or scan) 
 */
public void buildExpression(){
  AttributeValue hashAttrValue = buildKeyHashAttribute();
  if (hashAttrValue == null)
    throw new IllegalStateException("There is not a key schema defined.");
  if (DynamoDBQuery.getType().equals(RANGE_QUERY)){
    Condition newCondition = buildRangeCondition();
    buildQueryExpression(newCondition, hashAttrValue);
  }
  if (DynamoDBQuery.getType().equals(SCAN_QUERY))
    buildScanExpression(hashAttrValue);
}

/**
 * Builds scan query expression using a hash attribute value where to start
 * @param pHashAttrValueHash attribute value where to start scanning
 */
public void buildScanExpression(AttributeValue pHashAttrValue){
  DynamoDBScanExpression newScanExpression = new DynamoDBScanExpression();
  // TODO right now we only support scanning using the key, but we should support other types of scans
  newScanExpression.addFilterCondition(getKeySchema().getHashKeyElement().getAttributeName(), buildKeyScanCondition());
  dynamoDBExpression = newScanExpression;
}

/**
 * Builds range query expression
 * @param pNewConditionCondition for querying
 * @param pHashAttrValueHash attribute value where to start
 */
public void buildQueryExpression(Condition pNewCondition, AttributeValue pHashAttrValue) {
  DynamoDBQueryExpression newQueryExpression = new DynamoDBQueryExpression(pHashAttrValue); 
  newQueryExpression.setConsistentRead(getConsistencyReadLevel());
  newQueryExpression.setRangeKeyCondition(pNewCondition);
  dynamoDBExpression = newQueryExpression;
}

/**
 * Builds hash key attribute from generic query received
 * @returnAttributeValue build from query
 */
private AttributeValue buildKeyHashAttribute(){
  String pAttrType = getKeySchema().getHashKeyElement().getAttributeType();
  if(pAttrType.equals("S"))
    return new AttributeValue().withS(getHashKey(query.getKey()).toString());
  else if(pAttrType.equals("N"))
    return new AttributeValue().withN(getHashKey(query.getKey()).toString());
  return null;
}

private static Map<String, AttributeValue> createAttributes(
        HashMap<String, ByteIterator> values) {
    Map<String, AttributeValue> attributes = new HashMap<String, AttributeValue>(
            values.size() + 1); //leave space for the PrimaryKey
    for (Entry<String, ByteIterator> val : values.entrySet()) {
        attributes.put(val.getKey(), new AttributeValue(val.getValue()
                .toString()));
    }
    return attributes;
}

private HashMap<String, ByteIterator> extractResult(Map<String, AttributeValue> item) {
    if(null == item)
        return null;
    HashMap<String, ByteIterator> rItems = new HashMap<String, ByteIterator>(item.size());

    for (Entry<String, AttributeValue> attr : item.entrySet()) {
        logger.debug(String.format("Result- key: %s, value: %s", attr.getKey(), attr.getValue()) );
        rItems.put(attr.getKey(), new StringByteIterator(attr.getValue().getS()));
    }
    return rItems;
}

/**
 * Returns a list of files that should exist in the FileSystem with 
 * optional inclusion of deleted entries.
 * 
 * @param paths
 * @param includeDeleted
 * @return
 * @throws Exception 
 */
public List<FileInfo> list(List<Path> paths, boolean includeDeleted) throws Exception {
    List<FileInfo> listing = new ArrayList<FileInfo>();

    for(Path path : paths) {
        Key startKey = null;

        do {
            RetryTask<QueryResult> queryTask = new RetryTask(new QueryTask(path, startKey), retryCount, timeout);
            QueryResult result = queryTask.call();

            for(Map<String, AttributeValue> item : result.getItems()) {
                FileInfo file = new FileInfo(new Path(scheme+":"+item.get(HASH_KEY).getS() +"/"+ item.get(RANGE_KEY).getS()));

                if(item.containsKey(DELETE_MARKER)) {
                    file.setDeleted(Boolean.parseBoolean(item.get(DELETE_MARKER).getS()));

                    //@TODO: cleanup deleteMarker logic after deployed
                    if(!includeDeleted) {
                        continue;
                    }
                }

                if(item.containsKey(DIRECTORY_VALUE)) {
                    file.setDirectory(Boolean.parseBoolean((item.get(DIRECTORY_VALUE).getS())));
                }

                listing.add(file);
            }

            startKey = result.getLastEvaluatedKey();
        } while(startKey != null);
    }

    return listing;
}

@Override
public DeleteItemResult call() throws Exception {
    DeleteItemRequest delete = new DeleteItemRequest();
    delete.setTableName(tableName);
    delete.setKey(new Key(new AttributeValue(normalize(path.getParent())), new AttributeValue(path.getName())));
    delete.setReturnValues(ReturnValue.NONE);

    if(log.isDebugEnabled()) {
        log.debug("Deleting DynamoDB path: " + path.toUri());
    }

    return db.deleteItem(delete);
}

private static Key createPrimaryKey(String key) {
    Key k = new Key().withHashKeyElement(new AttributeValue().withS(key));
    return k;
}

@Override
public Object call() throws Exception {
    long epoch = System.currentTimeMillis();

    AttributeValue avPath = new AttributeValue(normalize(path.getParent()));
    AttributeValue avFile = new AttributeValue(path.getName());
    AttributeValue avEpoch = new AttributeValue().withN(epoch+"");

    PutItemRequest put = new PutItemRequest();
    put.setTableName(tableName);
    Map<String, AttributeValue> items = new HashMap<String, AttributeValue>();

    items.put(HASH_KEY, avPath);
    items.put(RANGE_KEY, avFile);
    items.put(EPOCH_VALUE, avEpoch);

    if(directory) {
        items.put(DIRECTORY_VALUE, new AttributeValue(Boolean.TRUE.toString()));
    }

    put.setItem(items);

    if(log.isDebugEnabled()) {
        log.debug("Adding metastore entry for: " + path.toUri());
    }

    db.putItem(put);

    PutItemRequest tsPut = new PutItemRequest();
    tsPut.setTableName(tableName);
    Map<String, AttributeValue> tsItems = new HashMap<String, AttributeValue>();

    tsItems.put(HASH_KEY, new AttributeValue(TIMESERIES_KEY));
    tsItems.put(RANGE_KEY, new AttributeValue(epoch+"-"+rand.nextInt()));
    tsItems.put(LINK_HASH_KEY, avPath);
    tsItems.put(LINK_RANGE_KEY, avFile);
    tsPut.setItem(tsItems);

    db.putItem(tsPut);

    return null;
}

@Override
public Object call() throws Exception {
    running = true;

    long deleteEpoch = System.currentTimeMillis() - age;

    QueryRequest query = new QueryRequest();
    query.setTableName(MetastoreJanitor.tableName);
    query.setHashKeyValue(new AttributeValue().withS(DynamoDBMetastore.TIMESERIES_KEY));
    query.setLimit(queueSize/2);

    QueryResult result;

    int scanCount = 0;
    int deleteCount = 0;

    do {
        //Can't set a hard limit on the queue since paths can be resubmitted by delete task
        //which can cause a deadlock.
        synchronized (deleteQueue) {
            while (deleteQueue.size() >= queueSize) {
                deleteQueue.wait();
            }
        }

        if(!running) {
            break;
        }

        result = db.query(query);

        scanCount += result.getCount();

        long epoch = deleteEpoch;

        for (Map<String, AttributeValue> i : result.getItems()) {
            epoch = Long.parseLong(i.get(DynamoDBMetastore.RANGE_KEY).getS().split("-")[0]);

            if (epoch >= deleteEpoch) {
                log.info("Timeseries scan complete.  Exiting.");
                running = false;
                break;
            }

            deleteCount += 2;

            deleteQueue.put(new Key(i.get(DynamoDBMetastore.HASH_KEY), i.get(DynamoDBMetastore.RANGE_KEY)));
            deleteQueue.put(new Key(i.get(DynamoDBMetastore.LINK_HASH_KEY), i.get(DynamoDBMetastore.LINK_RANGE_KEY)));
        }

        if(scanCount % reportInterval == 0) {
            log.info(format("scanned: %d, added: %d, queue_size: %d, current_date: %s", scanCount, deleteCount, deleteQueue.size(), new Date(epoch)));
        }

        limiter.acquire(result.getConsumedCapacityUnits().intValue());

        query.setExclusiveStartKey(result.getLastEvaluatedKey());
    } while (running && result.getLastEvaluatedKey() != null);

    log.info(format("Scan Complete.%nEntries Scanned: %d%nEntries Deleted: %d", scanCount, deleteCount));

    return Boolean.TRUE;
}

@Override
public Object call() throws Exception {
    running = true;

    long deleteEpoch = System.currentTimeMillis() - age;

    Map<String, Condition> filter = new HashMap<String, Condition>();

    AttributeValue value = new AttributeValue();
    value.setN(deleteEpoch + "");

    Condition c = new Condition();
    c.setComparisonOperator(ComparisonOperator.LT);
    c.setAttributeValueList(Collections.singletonList(value));

    filter.put("epoch", c);

    ScanRequest scan = new ScanRequest(MetastoreJanitor.tableName);
    scan.setScanFilter(filter);
    scan.setLimit( (int) limiter.getRate());

    ScanResult result;

    int scanTotal = 0;
    int matched = 0;

    do {
        //Can't set a hard limit on the queue since paths can be resubmitted by delete task
        synchronized (deleteQueue) {
            while (deleteQueue.size() >= queueSize) {
                deleteQueue.wait();
            }
        }

        if(!running) {
            break;
        }

        result = db.scan(scan);
        scanTotal += result.getScannedCount();
        matched += result.getCount();

        log.info(String.format("Total scanned: %d, matched: %d, added: %d, queue size: %d, consumed capacity: %f, max rate: %f", scanTotal, matched, result.getCount(), deleteQueue.size(), result.getConsumedCapacityUnits(), limiter.getRate()));

        for (Map<String, AttributeValue> i : result.getItems()) {
            if (!i.containsKey("epoch")) {
                continue;
            }
            deleteQueue.put(new Key(i.get(DynamoDBMetastore.HASH_KEY), i.get(DynamoDBMetastore.RANGE_KEY)));
        }

        limiter.acquire(result.getConsumedCapacityUnits().intValue());

        scan.setExclusiveStartKey(result.getLastEvaluatedKey());
    } while (running && result.getLastEvaluatedKey() != null);

    return Boolean.TRUE;
}