import com.google.datastore.v1.Key; //导入依赖的package包/类
@ProcessElement
public void processElement(ProcessContext c) {
  Entity.Builder entityBuilder = Entity.newBuilder();
  Key key = makeKey(makeKey(kind, ancestorKey).build(), kind, c.element().getKey()).build();

  entityBuilder.setKey(key);
  List<Value> candidates = new ArrayList<>();
  Map<String, Value> properties = new HashMap<>();
  for (CompletionCandidate tag : c.element().getValue()) {
    Entity.Builder tagEntity = Entity.newBuilder();
    properties.put("tag", makeValue(tag.value).build());
    properties.put("count", makeValue(tag.count).build());
    candidates.add(makeValue(tagEntity).build());
  }
  properties.put("candidates", makeValue(candidates).build());
  entityBuilder.putAllProperties(properties);
  c.output(entityBuilder.build());
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
  String entityJson = c.element();

  if (getJSTransform().hasTransform()) {
    entityJson = getJSTransform().invoke(entityJson);
  }

  Entity.Builder builder = Entity.newBuilder();
  getJsonParser().merge(entityJson, builder);
  Entity entity = builder.build();

  // Remove old project id reference from key
  Key k = entity.getKey();
  builder.setKey(Key.newBuilder()
      .addAllPath(k.getPathList())
      .setPartitionId(PartitionId.newBuilder()
          .setNamespaceId(k.getPartitionId().getNamespaceId()))
      .build());
  c.output(builder.build());
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Override
public void processElement(ProcessContext c) {
	String[] columns = c.element().split(",");

	Builder builder = Key.newBuilder();
	PathElement pathElement = builder.addPathBuilder().setKind("Product").setName(columns[0]).build();
	Key key = builder.setPath(0, pathElement).build();

	Entity.Builder entityBuilder = Entity.newBuilder();
	entityBuilder.setKey(key);
	entityBuilder.getMutableProperties().put("ProductCode", makeValue(columns[0]).build());
	entityBuilder.getMutableProperties().put("CategoryCode", makeValue(columns[1]).build());
	entityBuilder.getMutableProperties().put("ProductName", makeValue(columns[2]).build());
	entityBuilder.getMutableProperties().put("Price", makeValue(Integer.valueOf(columns[3])).build());
	
	String imageURL = "";
	if (columns.length > 4) {
		imageURL = columns[4];
	}
	entityBuilder.getMutableProperties().put("ImageURL", makeValue(imageURL).build());
	Entity entity = entityBuilder.build();
	c.output(entity);
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Build an entity for the given ancestorKey, kind, namespace and value.
 * @param largePropertySize if greater than 0, add an unindexed property of the given size.
 */
static Entity makeEntity(Long value, Key ancestorKey, String kind, @Nullable String namespace,
    int largePropertySize) {
  Entity.Builder entityBuilder = Entity.newBuilder();
  Key.Builder keyBuilder = makeKey(ancestorKey, kind, UUID.randomUUID().toString());
  // NOTE: Namespace is not inherited between keys created with DatastoreHelper.makeKey, so
  // we must set the namespace on keyBuilder. TODO: Once partitionId inheritance is added,
  // we can simplify this code.
  if (namespace != null) {
    keyBuilder.getPartitionIdBuilder().setNamespaceId(namespace);
  }

  entityBuilder.setKey(keyBuilder.build());
  entityBuilder.putProperties("value", makeValue(value).build());
  if (largePropertySize > 0) {
    entityBuilder.putProperties("unindexed_value", makeValue(new String(
        new char[largePropertySize]).replace("\0", "A")).setExcludeFromIndexes(true).build());
  }
  return entityBuilder.build();
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Override
public int compare(Key thisKey, Key otherKey) {
  if (!thisKey.getPartitionId().equals(otherKey.getPartitionId())) {
    throw new IllegalArgumentException("Cannot compare keys with different partition ids.");
  }

  Iterator<PathElement> thisPath = thisKey.getPathList().iterator();
  Iterator<PathElement> otherPath = otherKey.getPathList().iterator();
  while (thisPath.hasNext()) {
    if (!otherPath.hasNext()) {
      return 1;
    }
    int result = comparePathElement(thisPath.next(), otherPath.next());
    if (result != 0) {
      return result;
    }
  }

  return otherPath.hasNext() ? -1 : 0;
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Override
public List<Query> getSplits(
    Query query, PartitionId partition, int numSplits, Datastore datastore)
    throws DatastoreException, IllegalArgumentException {

  List<Query> splits = new ArrayList<Query>(numSplits);
  if (numSplits == 1) {
    splits.add(query);
    return splits;
  }
  validateQuery(query);
  validateSplitSize(numSplits);

  List<Key> scatterKeys = getScatterKeys(numSplits, query, partition, datastore);
  Key lastKey = null;
  for (Key nextKey : getSplitKey(scatterKeys, numSplits)) {
    splits.add(createSplit(lastKey, nextKey, query));
    lastKey = nextKey;
  }
  splits.add(createSplit(lastKey, null, query));
  return splits;
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Create a new {@link Query} given the query and range.
 *
 * @param lastKey the previous key. If null then assumed to be the beginning.
 * @param nextKey the next key. If null then assumed to be the end.
 * @param query the desired query.
 */
private Query createSplit(Key lastKey, Key nextKey, Query query) {
  if (lastKey == null && nextKey == null) {
    return query;
  }
  List<Filter> keyFilters = new ArrayList<Filter>();
  if (query.hasFilter()) {
    keyFilters.add(query.getFilter());
  }
  if (lastKey != null) {
    Filter lowerBound = DatastoreHelper.makeFilter(DatastoreHelper.KEY_PROPERTY_NAME,
        PropertyFilter.Operator.GREATER_THAN_OR_EQUAL,
        DatastoreHelper.makeValue(lastKey)).build();
    keyFilters.add(lowerBound);
  }
  if (nextKey != null) {
    Filter upperBound = DatastoreHelper.makeFilter(DatastoreHelper.KEY_PROPERTY_NAME,
        PropertyFilter.Operator.LESS_THAN,
        DatastoreHelper.makeValue(nextKey)).build();
    keyFilters.add(upperBound);
  }
  return Query.newBuilder(query).setFilter(makeAndFilter(keyFilters)).build();
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Gets a list of split keys given a desired number of splits.
 *
 * <p>This list will contain multiple split keys for each split. Only a single split key
 * will be chosen as the split point, however providing multiple keys allows for more uniform
 * sharding.
 *
 * @param numSplits the number of desired splits.
 * @param query the user query.
 * @param partition the partition to run the query in.
 * @param datastore the datastore containing the data.
 * @throws DatastoreException if there was an error when executing the datastore query.
 */
private List<Key> getScatterKeys(
    int numSplits, Query query, PartitionId partition, Datastore datastore)
    throws DatastoreException {
  Query.Builder scatterPointQuery = createScatterQuery(query, numSplits);

  List<Key> keySplits = new ArrayList<Key>();

  QueryResultBatch batch;
  do {
    RunQueryRequest scatterRequest =
        RunQueryRequest.newBuilder()
            .setPartitionId(partition)
            .setQuery(scatterPointQuery)
            .build();
    batch = datastore.runQuery(scatterRequest).getBatch();
    for (EntityResult result : batch.getEntityResultsList()) {
      keySplits.add(result.getEntity().getKey());
    }
    scatterPointQuery.setStartCursor(batch.getEndCursor());
    scatterPointQuery.getLimitBuilder().setValue(
        scatterPointQuery.getLimit().getValue() - batch.getEntityResultsCount());
  } while (batch.getMoreResults() == MoreResultsType.NOT_FINISHED);
  Collections.sort(keySplits, DatastoreHelper.getKeyComparator());
  return keySplits;
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Test
public void testMakeKey_PartitionId() {
  PartitionId partitionId = PartitionId.newBuilder()
      .setNamespaceId("namespace-id")
      .build();
  Key parent = PARENT.toBuilder()
      .setPartitionId(partitionId)
      .build();
  assertEquals(
      Key.newBuilder()
          .setPartitionId(partitionId)
          .addPath(PARENT.getPath(0))
          .addPath(Key.PathElement.newBuilder().setKind("Child"))
          .build(),
      makeKey(parent, "Child").build());
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Nullable
private String keyToString(Key k) {
  StringBuilder sb = new StringBuilder();

  List<String> paths = new ArrayList<>();
  for (PathElement p : k.getPathList()) {
    if (!Strings.isNullOrEmpty(p.getName())) {
      paths.add(String.format("%s, \'%s\'", p.getKind(), p.getName().replace("'", "\'")));
    } else {
      paths.add(String.format("%s, %s", p.getKind(), Long.toString(p.getId())));
    }
  }

  return "key(" + String.join(", ", paths) + ")";
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
public Entity makeEntity(KV<Integer, Iterable<TableRow>> content) {

			Key key = Key.newBuilder()
					.addPath(PathElement.newBuilder().setKind(this.kind).setId(content.getKey())).build();

			String keyword = "";
			List<Value> list = new ArrayList<>();
			for (TableRow row : content.getValue()) {
				String utterance = row.get("utterance").toString();
				if (utterance == null || utterance.length() < 1) {
					continue;
				}
				String word = row.get("keyword").toString();
				if (keyword.equals(row.get("keyword")) == false) {
					keyword = word;
				}
				if (list.size() > 1000) {
					LOG.info("Truncated the text." + "keyword_id = " + content.getKey() + ", keyword = " + word);
					break;
				}
				list.add(Value.newBuilder().setStringValue(utterance).build());
			}

			Entity.Builder entityBuilder = Entity.newBuilder();
			entityBuilder.setKey(key);

			Map<String, Value> propertyMap = new HashMap<String, Value>();
			propertyMap.put("KeywordID", Value.newBuilder().setIntegerValue(content.getKey()).build());
			propertyMap.put("Keyword", Value.newBuilder().setStringValue(keyword).build());
			ArrayValue array = ArrayValue.newBuilder().addAllValues(list).build();
			propertyMap.put("Candidates", Value.newBuilder().setArrayValue(array).build());

			entityBuilder.putAllProperties(propertyMap);

			return entityBuilder.build();
		}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Returns true if a Cloud Datastore key is complete. A key is complete if its last element
 * has either an id or a name.
 */
static boolean isValidKey(Key key) {
  List<PathElement> elementList = key.getPathList();
  if (elementList.isEmpty()) {
    return false;
  }
  PathElement lastElement = elementList.get(elementList.size() - 1);
  return (lastElement.getId() != 0 || !lastElement.getName().isEmpty());
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Override
public Mutation apply(Key key) {
  // Verify that the entity to delete has a complete key.
  checkArgument(isValidKey(key),
      "Keys to be deleted from the Cloud Datastore must be complete:\n%s", key);

  return makeDelete(key).build();
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
private static void writeEntitiesToDatastore(V1TestOptions options, String project,
    String ancestor, long numEntities) throws Exception {
  Datastore datastore = getDatastore(options, project);
  // Write test entities to datastore
  V1TestWriter writer = new V1TestWriter(datastore, new UpsertMutationBuilder());
  Key ancestorKey = makeAncestorKey(options.getNamespace(), options.getKind(), ancestor);

  for (long i = 0; i < numEntities; i++) {
    Entity entity = makeEntity(i, ancestorKey, options.getKind(), options.getNamespace(), 0);
    writer.write(entity);
  }
  writer.close();
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Test
public void testDeleteKeyPrimitiveDisplayData() {
  DisplayDataEvaluator evaluator = DisplayDataEvaluator.create();
  PTransform<PCollection<Key>, ?> write =
      DatastoreIO.v1().deleteKey().withProjectId("myProject");

  Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(write);
  assertThat("DatastoreIO write should include the project in its primitive display data",
      displayData, hasItem(hasDisplayItem("projectId")));
  assertThat("DatastoreIO write should include the deleteKeyFn in its primitive display data",
      displayData, hasItem(hasDisplayItem("deleteKeyFn")));
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Test the detection of complete and incomplete keys.
 */
@Test
public void testHasNameOrId() {
  Key key;
  // Complete with name, no ancestor
  key = makeKey("bird", "finch").build();
  assertTrue(isValidKey(key));

  // Complete with id, no ancestor
  key = makeKey("bird", 123).build();
  assertTrue(isValidKey(key));

  // Incomplete, no ancestor
  key = makeKey("bird").build();
  assertFalse(isValidKey(key));

  // Complete with name and ancestor
  key = makeKey("bird", "owl").build();
  key = makeKey(key, "bird", "horned").build();
  assertTrue(isValidKey(key));

  // Complete with id and ancestor
  key = makeKey("bird", "owl").build();
  key = makeKey(key, "bird", 123).build();
  assertTrue(isValidKey(key));

  // Incomplete with ancestor
  key = makeKey("bird", "owl").build();
  key = makeKey(key, "bird").build();
  assertFalse(isValidKey(key));

  key = makeKey().build();
  assertFalse(isValidKey(key));
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Test that entities with incomplete keys cannot be updated.
 */
@Test
public void testAddEntitiesWithIncompleteKeys() throws Exception {
  Key key = makeKey("bird").build();
  Entity entity = Entity.newBuilder().setKey(key).build();
  UpsertFn upsertFn = new UpsertFn();

  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("Entities to be written to the Cloud Datastore must have complete keys");

  upsertFn.apply(entity);
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
@Test
/**
 * Test that entities with valid keys are transformed to upsert mutations.
 */
public void testAddEntities() throws Exception {
  Key key = makeKey("bird", "finch").build();
  Entity entity = Entity.newBuilder().setKey(key).build();
  UpsertFn upsertFn = new UpsertFn();

  Mutation exceptedMutation = makeUpsert(entity).build();
  assertEquals(upsertFn.apply(entity), exceptedMutation);
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Test that entities with incomplete keys cannot be deleted.
 */
@Test
public void testDeleteEntitiesWithIncompleteKeys() throws Exception {
  Key key = makeKey("bird").build();
  Entity entity = Entity.newBuilder().setKey(key).build();
  DeleteEntityFn deleteEntityFn = new DeleteEntityFn();

  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("Entities to be deleted from the Cloud Datastore must have complete keys");

  deleteEntityFn.apply(entity);
}
 

import com.google.datastore.v1.Key; //导入依赖的package包/类
/**
 * Test that entities with valid keys are transformed to delete mutations.
 */
@Test
public void testDeleteEntities() throws Exception {
  Key key = makeKey("bird", "finch").build();
  Entity entity = Entity.newBuilder().setKey(key).build();
  DeleteEntityFn deleteEntityFn = new DeleteEntityFn();

  Mutation exceptedMutation = makeDelete(entity.getKey()).build();
  assertEquals(deleteEntityFn.apply(entity), exceptedMutation);
}