feat(datastore): add support for request tags (#13732)

This PR introduced sending request tags for different datastore
requests. The tags can be passed using 2 mechanisms:
1. Instance Level tags.
2. Request Level tags.
- RequestOptions is the proto that holds all the instance and request
level tags and is a part of all request protos.
- DatastoreOptions allows the ability to set the instance level tags
using setRequestTags. Any operation executed via this Datastore client
automatically includes these global request tags in the outbound RPC
request payload.
- RequestOptionsHelper an internal utility that combines instance-level
tags from DatastoreOptions with any request-level RequestOptions
protobuf message before building the wire request.
- DatastoreExecutionOptions is a holder object for all different options
like ExplainOptions, RequestOptions and ReadOption.
- Overloaded versions for taking in the DatastoreExecutionOptions for
each of the following operations:
    - put
    - get
    - fetch
    - delete
    - newTransaction
    - commit
    - rollback
    - allocateId
    - reserveIds
    - runQuery
    - runAggregationQuery
- Each of these overloaded versions copy the request tags into the
request protos.

Author: rahulmane-goog

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/Datastore.java

@@ -44,6 +44,24 @@ public interface Datastore extends Service<DatastoreOptions>, DatastoreReaderWri
    */
   Transaction newTransaction();
 
+  /**
+   * Returns a new Datastore transaction with specified {@link DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  Transaction newTransaction(DatastoreExecutionOptions executionOptions);
+
+  /**
+   * Returns a new Datastore transaction with specified {@link TransactionOptions} and {@link
+   * DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  Transaction newTransaction(
+      TransactionOptions options, DatastoreExecutionOptions executionOptions);
+
   /**
    * A callback for running with a transactional {@link
    * com.google.cloud.datastore.DatastoreReaderWriter}. The associated transaction will be committed
@@ -179,6 +197,15 @@ interface TransactionCallable<T> {
    */
   List<Key> allocateId(IncompleteKey... keys);
 
+  /**
+   * Returns a list of keys using the allocated ids with specified {@link
+   * DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  List<Key> allocateId(List<IncompleteKey> keys, DatastoreExecutionOptions executionOptions);
+
   /**
    * Reserve one or more keys, preventing them from being automatically allocated by Datastore.
    *
@@ -196,6 +223,14 @@ interface TransactionCallable<T> {
    */
   List<Key> reserveIds(Key... keys);
 
+  /**
+   * Reserve one or more keys with specified {@link DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  List<Key> reserveIds(List<Key> keys, DatastoreExecutionOptions executionOptions);
+
   /**
    * {@inheritDoc}
    *
@@ -226,6 +261,16 @@ interface TransactionCallable<T> {
   @Override
   Entity add(FullEntity<?> entity);
 
+  /**
+   * Datastore add operation: inserts the provided entity with specified {@link
+   * DatastoreExecutionOptions}. This method will automatically allocate an id if necessary.
+   *
+   * @throws DatastoreException upon failure
+   * @throws IllegalArgumentException if the given entity is missing a key
+   */
+  @BetaApi
+  Entity add(FullEntity<?> entity, DatastoreExecutionOptions executionOptions);
+
   /**
    * {@inheritDoc}
    *
@@ -264,6 +309,17 @@ interface TransactionCallable<T> {
   @Override
   List<Entity> add(FullEntity<?>... entities);
 
+  /**
+   * Datastore add operation: inserts the provided entities with specified {@link
+   * DatastoreExecutionOptions}. This method will automatically allocate id for any entity with an
+   * incomplete key.
+   *
+   * @throws DatastoreException upon failure
+   * @throws IllegalArgumentException if any of the given entities is missing a key
+   */
+  @BetaApi
+  List<Entity> add(List<FullEntity<?>> entities, DatastoreExecutionOptions executionOptions);
+
   /**
    * {@inheritDoc}
    *
@@ -290,6 +346,15 @@ interface TransactionCallable<T> {
   @Override
   void update(Entity... entities);
 
+  /**
+   * A Datastore update operation with specified {@link DatastoreExecutionOptions}. The operation
+   * will fail if an entity with the same key does not already exist.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  void update(List<Entity> entities, DatastoreExecutionOptions executionOptions);
+
   /**
    * {@inheritDoc}
    *
@@ -309,31 +374,16 @@ interface TransactionCallable<T> {
   @Override
   Entity put(FullEntity<?> entity);
 
+  @Override
+  List<Entity> put(FullEntity<?>... entities);
+
   /**
-   * {@inheritDoc}
-   *
-   * <p>Example of putting multiple entities.
-   *
-   * <pre>{@code
-   * String keyName1 = "my_key_name1";
-   * String keyName2 = "my_key_name2";
-   * Key key1 = datastore.newKeyFactory().setKind("MyKind").newKey(keyName1);
-   * Entity.Builder entityBuilder1 = Entity.newBuilder(key1);
-   * entityBuilder1.set("propertyName", "value1");
-   * Entity entity1 = entityBuilder1.build();
-   *
-   * Key key2 = datastore.newKeyFactory().setKind("MyKind").newKey(keyName2);
-   * Entity.Builder entityBuilder2 = Entity.newBuilder(key2);
-   * entityBuilder2.set("propertyName", "value2");
-   * Entity entity2 = entityBuilder2.build();
-   *
-   * datastore.put(entity1, entity2);
-   * }</pre>
+   * {@inheritDoc} with specified {@link DatastoreExecutionOptions}.
    *
    * @throws DatastoreException upon failure
    */
-  @Override
-  List<Entity> put(FullEntity<?>... entities);
+  @BetaApi
+  List<Entity> put(List<FullEntity<?>> entities, DatastoreExecutionOptions executionOptions);
 
   /**
    * {@inheritDoc}
@@ -353,6 +403,14 @@ interface TransactionCallable<T> {
   @Override
   void delete(Key... keys);
 
+  /**
+   * A datastore delete operation with specified {@link DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  void delete(List<Key> keys, DatastoreExecutionOptions executionOptions);
+
   /**
    * Returns a new KeyFactory for this service
    *
@@ -381,6 +439,15 @@ interface TransactionCallable<T> {
    */
   Entity get(Key key, ReadOption... options);
 
+  /**
+   * Returns an {@link Entity} for the given {@link Key} with specified {@link
+   * DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  Entity get(Key key, DatastoreExecutionOptions executionOptions);
+
   /**
    * Returns an {@link Entity} for each given {@link Key} that exists in the Datastore. The order of
    * the result is unspecified. Results are loaded lazily, so it is possible to get a {@code
@@ -409,6 +476,15 @@ interface TransactionCallable<T> {
    */
   Iterator<Entity> get(Iterable<Key> keys, ReadOption... options);
 
+  /**
+   * Returns an {@link Entity} for each given {@link Key} with specified {@link
+   * DatastoreExecutionOptions}.
+   *
+   * @throws DatastoreException upon failure
+   */
+  @BetaApi
+  Iterator<Entity> get(Iterable<Key> keys, DatastoreExecutionOptions executionOptions);
+
   /**
    * Returns a list with a value for each given key (ordered by input). {@code null} values are
    * returned for nonexistent keys. When possible prefer using {@link #get(Key...)} to avoid eagerly
@@ -430,6 +506,13 @@ interface TransactionCallable<T> {
    */
   List<Entity> fetch(Iterable<Key> keys, ReadOption... options);
 
+  /**
+   * Returns a list with a value for each given key with specified {@link
+   * DatastoreExecutionOptions}.
+   */
+  @BetaApi
+  List<Entity> fetch(Iterable<Key> keys, DatastoreExecutionOptions executionOptions);
+
   /**
    * Submits a {@link Query} and returns its result. {@link ReadOption}s can be specified if
    * desired.
@@ -492,6 +575,13 @@ interface TransactionCallable<T> {
   @BetaApi
   <T> QueryResults<T> run(Query<T> query, ExplainOptions explainOptions, ReadOption... options);
 
+  /**
+   * Submits a {@link Query} with specified {@link DatastoreExecutionOptions} and returns its
+   * result.
+   */
+  @BetaApi
+  <T> QueryResults<T> run(Query<T> query, DatastoreExecutionOptions executionOptions);
+
   /**
    * Submits a {@link AggregationQuery} and returns {@link AggregationResults}. {@link ReadOption}s
    * can be specified if desired.
@@ -564,6 +654,14 @@ interface TransactionCallable<T> {
   AggregationResults runAggregation(
       AggregationQuery query, ExplainOptions explainOptions, ReadOption... options);
 
+  /**
+   * Submits an {@link AggregationQuery} with specified {@link DatastoreExecutionOptions} and
+   * returns {@link AggregationResults}.
+   */
+  @BetaApi
+  AggregationResults runAggregation(
+      AggregationQuery query, DatastoreExecutionOptions executionOptions);
+
   /**
    * Closes the gRPC channels associated with this instance and frees up their resources. This
    * method blocks until all channels are closed. Once this method is called, this Datastore client

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreExecutionOptions.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.datastore;
+
+import com.google.api.core.BetaApi;
+import com.google.cloud.datastore.models.ExplainOptions;
+import com.google.cloud.datastore.models.RequestOptions;
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import java.util.Collections;
+import java.util.List;
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Class representing options for query execution in Google Cloud Datastore. Combines {@link
+ * ExplainOptions}, {@link RequestOptions}, and {@link ReadOption}s.
+ */
+@BetaApi
+@NullMarked
+public class DatastoreExecutionOptions {
+
+  private final @Nullable ExplainOptions explainOptions;
+  private final RequestOptions requestOptions;
+  private final List<ReadOption> readOptions;
+
+  private DatastoreExecutionOptions(Builder builder) {
+    this.explainOptions = builder.explainOptions;
+    this.requestOptions = builder.requestOptions;
+    this.readOptions = ImmutableList.copyOf(builder.readOptions);
+  }
+
+  public @Nullable ExplainOptions getExplainOptions() {
+    return explainOptions;
+  }
+
+  public RequestOptions getRequestOptions() {
+    return requestOptions;
+  }
+
+  public List<ReadOption> getReadOptions() {
+    return readOptions;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (!(o instanceof DatastoreExecutionOptions)) return false;
+    DatastoreExecutionOptions that = (DatastoreExecutionOptions) o;
+    return Objects.equal(explainOptions, that.explainOptions)
+        && Objects.equal(requestOptions, that.requestOptions)
+        && Objects.equal(readOptions, that.readOptions);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(explainOptions, requestOptions, readOptions);
+  }
+
+  public Builder toBuilder() {
+    return new Builder(this);
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /** Returns a default {@code DatastoreExecutionOptions} instance. */
+  public static DatastoreExecutionOptions getDefaultInstance() {
+    return newBuilder().build();
+  }
+
+  /** Builder for {@link DatastoreExecutionOptions}. */
+  public static class Builder {
+    private @Nullable ExplainOptions explainOptions;
+    private RequestOptions requestOptions = RequestOptions.getDefaultInstance();
+    private List<ReadOption> readOptions = Collections.emptyList();
+
+    private Builder() {}
+
+    private Builder(DatastoreExecutionOptions options) {
+      this.explainOptions = options.explainOptions;
+      this.requestOptions = options.requestOptions;
+      this.readOptions = options.readOptions;
+    }
+
+    public Builder setExplainOptions(@Nullable ExplainOptions explainOptions) {
+      this.explainOptions = explainOptions;
+      return this;
+    }
+
+    public Builder setRequestOptions(RequestOptions requestOptions) {
+      Preconditions.checkNotNull(requestOptions, "requestOptions cannot be null");
+      this.requestOptions = requestOptions;
+      return this;
+    }
+
+    public Builder setReadOptions(List<ReadOption> readOptions) {
+      Preconditions.checkNotNull(readOptions, "readOptions cannot be null");
+      this.readOptions = readOptions;
+      return this;
+    }
+
+    public DatastoreExecutionOptions build() {
+      return new DatastoreExecutionOptions(this);
+    }
+  }
+}

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreHelper.java

@@ -74,7 +74,7 @@ static List<Entity> fetch(Datastore reader, Key[] keys, ReadOption... options) {
     return compileEntities(keys, reader.get(Arrays.asList(keys), options));
   }
 
-  private static List<Entity> compileEntities(Key[] keys, Iterator<Entity> entities) {
+  static List<Entity> compileEntities(Key[] keys, Iterator<Entity> entities) {
     Map<Key, Entity> map = Maps.newHashMapWithExpectedSize(keys.length);
     while (entities.hasNext()) {
       Entity entity = entities.next();