import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
private static Page<Item, QueryOutcome> queryGSI(String appid, Pager p) {
	Pager pager = (p != null) ? p : new Pager();
	Index index = getSharedIndex();
	QuerySpec spec = new QuerySpec().
			withMaxPageSize(pager.getLimit()).
			withMaxResultSize(pager.getLimit()).
			withKeyConditionExpression(Config._APPID + " = :aid").
			withValueMap(new ValueMap().withString(":aid", appid));

	if (!StringUtils.isBlank(pager.getLastKey())) {
		spec = spec.withExclusiveStartKey(new KeyAttribute(Config._APPID, appid),	// HASH/PARTITION KEY
				new KeyAttribute(Config._ID, pager.getLastKey()), // RANGE/SORT KEY
				new KeyAttribute(Config._KEY, getKeyForAppid(pager.getLastKey(), appid))); // TABLE PRIMARY KEY
	}
	return index != null ? index.query(spec).firstPage() : null;
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override public Void create(Group group) {
  Item item = preparePutItem(group);

  PutItemSpec putItemSpec = new PutItemSpec()
      .withItem(item)
      .withConditionExpression("attribute_not_exists(#ns_key)")
      .withNameMap(new NameMap().with("#ns_key", HASH_KEY));

  Table table = dynamoDB.getTable(groupTableName);
  final Supplier<PutItemOutcome> putItemOutcomeSupplier = () -> {
    try {
      return table.putItem(putItemSpec);
    } catch (ConditionalCheckFailedException e) {
      throwConflictAlreadyExists(group);
      return null;
    }
  };
  return putItem(group, putItemOutcomeSupplier);
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
private QueryOutcome query(Object hk, QuerySpec querySpec, PageIterator pageIterator) {
    if ( null == _convertMarker ) {
        throw new IllegalStateException("Index must first be initialized with ConvertMarker");
    }
    if ( pageIterator.getPageSize() <= 0 ) {
        return new QueryOutcome(new QueryResult());
    }
    ItemCollection<QueryOutcome> itemCollection =
        maybeBackoff(true, () ->
                     _query.query(withMarker(querySpec.withHashKey(_hkName, hk), pageIterator, hk)));

    if ( null != itemCollection ) {
        Iterator<Page<Item, QueryOutcome>> iterator = itemCollection.pages().iterator();
        if ( iterator.hasNext() ) {
            QueryOutcome outcome = maybeBackoff(true, () -> iterator.next().getLowLevelResult());
            QueryResult result = outcome.getQueryResult();
            if ( null != pageIterator.getMarker() && null != result.getItems() && result.getItems().size() > 0 ) {
                pageIterator.setPrevMarker(toMarker(result.getItems().get(0), true));
            } else {
                pageIterator.setPrevMarker(null);
            }
            Map<String,AttributeValue> lastKey = result.getLastEvaluatedKey();
            if ( null != lastKey && ! lastKey.isEmpty() ) {
                pageIterator.setMarker(toMarker(lastKey, true));
            } else {
                pageIterator.setMarker(null);
            }
            return outcome;
        }
    }
    pageIterator.setPrevMarker(null);
    pageIterator.setMarker(null);
    return new QueryOutcome(new QueryResult());
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override
public void transform(Item scoreItem, DynamoDB dynamodb) {
    String playerName = scoreItem.getString(PLAYER_NAME);
    int score = scoreItem.getInt(SCORE);
    int gameLength = scoreItem.getInt(GAME_LENGTH);
    
    /*
     * The XSpec API allows you to use DynamoDB's expression language
     * to execute expressions on the service-side.
     *  
     * https://java.awsblog.com/post/TxBG87QOQZRZJF/-DynamoDB-XSpec-API  
     */
    Table viewTable = dynamodb.getTable(PLAYER_STATS_TABLE_NAME);
    UpdateItemExpressionSpec incrementTotalOrder = new ExpressionSpecBuilder()
            .addUpdate(N(TOTAL_SCORE).add(score))
            .addUpdate(N(TOTAL_GAMEPLAY).add(gameLength))
            .addUpdate(N(TOTAL_GAMES).add(1))
            .buildForUpdate();
    viewTable.updateItem(PLAYER_NAME, playerName, incrementTotalOrder);
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
private static void findRepliesUsingAFilterExpression(String forumName, String threadSubject) {

        Table table = dynamoDB.getTable(tableName);
        
        String replyId = forumName + "#" + threadSubject;

        QuerySpec spec = new QuerySpec()
            .withProjectionExpression("Message, ReplyDateTime, PostedBy")
            .withKeyConditionExpression("Id = :v_id")
            .withFilterExpression("PostedBy = :v_postedby")
            .withValueMap(new ValueMap()
                .withString(":v_id", replyId)
                .withString(":v_postedby", "User B"));
        
        ItemCollection<QueryOutcome> items = table.query(spec);

        System.out.println("\nfindRepliesUsingAFilterExpression results:");
        Iterator<Item> iterator = items.iterator();
        while (iterator.hasNext()) {
            System.out.println(iterator.next().toJSONPretty());
        }    
     }
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override public Optional<Group> loadByKey(String key) {
  Table table = dynamoDB.getTable(this.groupTableName);

  QuerySpec querySpec = new QuerySpec()
      .withKeyConditionExpression(HASH_KEY + " = :k_app_key")
      .withValueMap(new ValueMap()
          .withString(":k_app_key", key)
      )
      .withMaxResultSize(1)
      .withConsistentRead(true);

  DynamoDbCommand<ItemCollection<QueryOutcome>> cmd = new DynamoDbCommand<>("loadByKey",
      () -> queryTable(table, querySpec),
      () -> {
        throw new RuntimeException("loadByKey");
      },
      dynamodbNamespaceGraphQueryHystrix,
      metrics);

  final ItemCollection<QueryOutcome> items = cmd.execute();
  final IteratorSupport<Item, QueryOutcome> iterator = items.iterator();
  if (iterator.hasNext()) {
    return Optional.of(GroupSupport.toGroup(iterator.next().getString("json")));
  }

  return Optional.empty();
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override public Optional<Feature> loadFeatureByKey(String group, String key) {
  logger.info("{}", kvp("op", "loadFeatureByKey", HASH_KEY, group, RANGE_KEY, key));

  Table table = dynamoDB.getTable(featureTableName);

  DynamoDbCommand<Item> cmd = new DynamoDbCommand<>("loadFeatureByKey",
      () -> getItem(group, key, table),
      () -> {
        throw new RuntimeException("loadFeatureById");
      },
      hystrixReadConfiguration,
      metrics);

  Item item = cmd.execute();
  if (item == null) {
    return Optional.empty();
  }
  return Optional.of(FeatureSupport.toFeature(item.getString("json")));
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override public List<Feature> loadFeatures(String group) {
  logger.info("{}", kvp("op", "loadFeatures", "group", group));
  List<Feature> features = Lists.newArrayList();

  Table table = dynamoDB.getTable(featureTableName);

  QuerySpec querySpec = new QuerySpec()
      .withKeyConditionExpression(HASH_KEY + " = :k_" + HASH_KEY)
      .withValueMap(new ValueMap().withString(":k_" + HASH_KEY, group))
      .withConsistentRead(true);

  DynamoDbCommand<ItemCollection<QueryOutcome>> cmd = new DynamoDbCommand<>("loadFeatures",
      () -> queryTable(table, querySpec),
      () -> {
        throw new RuntimeException("loadFeatureById");
      },
      hystrixReadConfiguration,
      metrics);

  ItemCollection<QueryOutcome> items = cmd.execute();

  for (Page<Item, QueryOutcome> page : items.pages()) {
    page.forEach(item -> features.add(FeatureSupport.toFeature(item.getString("json"))));
  }

  return features;
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
private boolean tryAddMissingPartition(String dyanmoDBTaableName,DynamoDB dynamoDBClient, Partition partition){

        Table ddbTable= dynamoDBClient.getTable(dyanmoDBTaableName);

        Item item=new Item()
                .withPrimaryKey("PartitionSpec",partition.spec())
                .withString("PartitionPath",partition.path())
                .withString("PartitionName", partition.name());

        PutItemSpec itemSpec=new PutItemSpec()
                .withItem(item)
                .withConditionExpression("attribute_not_exists(#ps)")
                .withNameMap(new NameMap()
                        .with("#ps","PartitionSpec"));

        try{
            ddbTable.putItem(itemSpec);
            System.out.println("Item was added to the table.PartitionSpec="+partition.spec()+"; Path="+partition.path());
            return true;
        }
        catch(ConditionalCheckFailedException e){
            System.out.println(e.toString());
            System.out.println("Item already exists. PartitionSpec="+partition.spec()+"; Path="+partition.path());
            return false;
        }
    }
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override
public Map<String, AttributeValue> getNodes(WebsiteModel websiteModel) {

    try
    {
        ObjectMapper mapper = new ObjectMapper();
        String string = mapper.writeValueAsString(websiteModel);
        Item item = new Item().withJSON(Utils.params.nodes, string);
        return InternalUtils.toAttributeValues(item);
    }
    catch (JsonProcessingException e)
    {
        LOG.error(e.getMessage());
    }

    return new HashMap<>();
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
protected void readStateInternal(final Object state, final Class<?> stateClass, final Item item, final ObjectMapper mapper)
{
    try
    {
        if (!state.getClass().equals(stateClass))
        {
            throw new IllegalArgumentException(String.format("State class (%s) did not match expected class (%s), Storage Extension should override generatePutItem method",
                    state.getClass().getName(),
                    stateClass.getName()));
        }

        mapper.readerForUpdating(state).readValue(item.getJSON(DynamoDBUtils.FIELD_NAME_DATA));
    }
    catch (IOException e)
    {
        throw new UncheckedException(e);
    }
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override
public StorageTestState readState(final String identity)
{
    final Table table = dynamoDBConnection.getDynamoDB().getTable(getTableName());
    final Item item = table.getItem("_id", generateItemId(identity));

    if (item != null)
    {
        try
        {
            final StorageTestState testState = new HelloState();
            dynamoDBConnection.getMapper().readerForUpdating(testState).readValue(item.getJSON("_state"));
            return testState;
        }
        catch (Exception e)
        {
            throw new UncheckedException(e);
        }
    }
    return null;
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Override
public int next() {
  Stopwatch watch = Stopwatch.createStarted();
  topLevelState.reset();

  int count = 0;
  Page<Item, ?> page;
  while (resultIter.hasNext()) {
    page = resultIter.next();
    for (Item item : page) {
      int rowCount = count++;
      for (Map.Entry<String, Object> attribute : item.attributes()) {
        String name = attribute.getKey();
        Object value = attribute.getValue();
        SchemaPath column = getSchemaPath(name);
        topLevelState.setColumn(column);
        handleTopField(rowCount, name, value, topLevelState);
      }
    }
  }

  topLevelState.setRowCount(count);
  LOG.debug("Took {} ms to get {} records", watch.elapsed(TimeUnit.MILLISECONDS), count);
  return count;
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
protected void verify(List<Map<String, Object>> rows, Item... items) {
  assertEquals("Wrong number of expected rows!" +
               "\nExpected: " + toString(items) +
               "\nGot rows: " + rows,
    items.length, rows.size());
  for (int i = 0; i < items.length; i++) {
    Map<String, Object> row = rows.get(i);
    Item item = items[i];
    assertEquals("Wrong number of fields in row! Got row: " + row + "\nExpected: " + item,
      row.size(), item.asMap().size());
    for (Map.Entry<String, Object> field : row.entrySet()) {
      String name = field.getKey();
      Object o = field.getValue();
      if (o instanceof Text) {
        o = o.toString();
      }
      if (item.get(name) instanceof Number) {
        equalsNumber(item, name, row);
      } else if (o instanceof byte[]) {
        assertArrayEquals("Array mismatch for: " + name, (byte[]) item.get(name), (byte[]) o);
      } else {
        assertEquals("Mismatch for: " + name, item.get(name), o);
      }
    }
  }
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Test
public void testPrimaryAndSortKeySpecification() throws Exception {
  String pk = "pk", sort = "sort";
  Table table = createHashAndSortTable(pk, sort);
  Item item = new Item();
  item.with(pk, "p1");
  item.with(sort, "s1");
  item.with(COL1, "1");
  table.putItem(item);

  Item item2 = new Item();
  item2.with(pk, "p1");
  item2.with(sort, "s0");
  item2.with(COL1, "2");
  table.putItem(item2);
  // should create a get
  String query = selectStarWithPK("p1", "t", table) + " AND sort = 's1'";
  verify(runAndReadResults(query), item);
  validatePlanWithGets(query,
    pkEquals("p1").and(create("equal", "sort", "s1")));
  // should create a query
  query = selectStarWithPK("p1", "t", table) + " AND sort >= 's1'";
  verify(runAndReadResults(query), item);
  validatePlanWithQueries(query, of(pkEquals("p1").and(
    DynamoPlanValidationUtils.gte("sort", "s1")), null));
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Test
public void testWhereColumnEqualsNull() throws Exception {
  Item item = item();
  item.with(COL1, null);
  Table table = createTableWithItems(item);

  String query =
    selectStarWithPK("pk", "t", table) + " AND t." + COL1 + " = cast(null as varchar)";
  verify(runAndReadResults(query), item);
  ImmutablePair<DynamoFilterSpec, DynamoFilterSpec> spec =
    of(pkEquals("pk"), DynamoPlanValidationUtils


      .equals(COL1, null));
  validatePlanWithQueries(query, spec);
  // we return nulls are varchar b/c we can cast anything from varchar. Make sure that a
  // boolean null cast also works
  query = selectStarWithPK("pk", "t", table) + " AND t." + COL1 + " = cast(null as boolean)";
  verify(runAndReadResults(query), item);
  validatePlanWithQueries(query, spec);
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
/**
 * Similar to above, but we check for the non-existance of a column
 */
@Test
public void testWhereNoColumnValueIsNull() throws Exception {
  Item item = item();
  Table table = createTableWithItems(item);
  String query =
    selectStarWithPK("pk", "t", table) + " AND t." + COL1 + " = cast(null as varchar)";
  assertEquals("Should not have found a row when checking for = null and column not set!",
    0, runAndReadResults(query).size());
  ImmutablePair<DynamoFilterSpec, DynamoFilterSpec> spec =
    of(pkEquals("pk"), DynamoPlanValidationUtils
      .equals(COL1, null));
  validatePlanWithQueries(query, spec);
  // see above for why trying a different type
  query = selectStarWithPK("pk", "t", table) + " AND t." + COL1 + " = cast(null as BOOLEAN)";
  assertEquals("Should not have found a row when checking for = null and column not set!",
    0, runAndReadResults(query).size());
  validatePlanWithQueries(query, spec);
  query = selectStarWithPK("pk", "t", table) + " AND t." + COL1 + " IS NULL";
  verify(runAndReadResults(query), item);
  validatePlanWithQueries(query, of(spec.getLeft(), create("isNull", COL1)));
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Test
public void testSimpleScan() throws Exception {
  Item item = item();
  item.with(COL1, 1);
  Table table = createTableWithItems(item);
  String select = "SELECT *" + from(table) + "t WHERE t." + COL1 + " = 1";
  verify(runAndReadResults(select), item);
  DynamoFilterSpec spec = DynamoPlanValidationUtils.equals(COL1, 1);
  validatePlanWithScan(select, spec);

  Item item2 = new Item();
  item2.with(PK, "pk2");
  item2.with(COL1, 2);
  table.putItem(item2);
  verify(runAndReadResults(select), item);
  // plan doesn't change as the table gets larger
  validatePlanWithScan(select, spec);
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Test
public void testGetAndQuery() throws Exception {
  Item item = item();
  item.with(COL1, "1");
  Item i2 = new Item();
  i2.with(PK, "pk2");
  i2.with(COL1, 2);
  Table table = createTableWithItems(item, i2);
  String query = "SELECT *" + from(table) + "t WHERE " +
                 "t." + PK + " = 'pk' OR " +
                 "t." + PK + " = 'pk2' AND t." + COL1 + " >= 2" +
                 "ORDER BY t." + PK + " ASC";
  verify(runAndReadResults(query), item, i2);
  validatePlan(query, null, null,
    newArrayList(new DynamoQueryFilterSpec(pkEquals("pk2"),
        DynamoPlanValidationUtils
          .gte(COL1, 2)),
      new DynamoGetFilterSpec(pkEquals("pk"))));
}
 

import com.amazonaws.services.dynamodbv2.document.Item; //导入依赖的package包/类
@Test
public void testQueryOrQuery() throws Exception {
  Item item = item();
  item.with(COL1, 1);
  Item i2 = new Item();
  i2.with(PK, "pk2");
  i2.with(COL1, 2);
  Table table = createTableWithItems(item, i2);
  String query = "SELECT *" + from(table) + "t WHERE " +
                 "t." + PK + " = 'pk' AND t." + COL1 + " = 1" +
                 " OR " +
                 "t." + PK + " = 'pk2' AND t." + COL1 + " >= 2" +
                 "ORDER BY t." + PK + " ASC";
  verify(runAndReadResults(query), item, i2);
  validatePlanWithQueries(query,
    of(pkEquals("pk2"), DynamoPlanValidationUtils.gte(COL1, 2)),
    of(pkEquals("pk"), DynamoPlanValidationUtils.equals(COL1, 1)));
}