feat: add PartitionedUpdate support to executor

README.md

@@ -49,20 +49,20 @@ If you are using Maven without BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.9.0')
+implementation platform('com.google.cloud:libraries-bom:26.10.0')
 
 implementation 'com.google.cloud:google-cloud-spanner'
 ```
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-spanner:6.36.1'
+implementation 'com.google.cloud:google-cloud-spanner:6.37.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.36.1"
+libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.37.0"
 ```
 
 ## Authentication

google-cloud-spanner-executor/src/main/java/com/google/cloud/executor/spanner/CloudClientExecutor.java

@@ -52,6 +52,7 @@
 import com.google.cloud.spanner.Mutation;
 import com.google.cloud.spanner.Mutation.WriteBuilder;
 import com.google.cloud.spanner.Options;
+import com.google.cloud.spanner.Options.RpcPriority;
 import com.google.cloud.spanner.Partition;
 import com.google.cloud.spanner.PartitionOptions;
 import com.google.cloud.spanner.ReadContext;
@@ -128,6 +129,8 @@
 import com.google.spanner.executor.v1.MutationAction.Mod;
 import com.google.spanner.executor.v1.MutationAction.UpdateArgs;
 import com.google.spanner.executor.v1.OperationResponse;
+import com.google.spanner.executor.v1.PartitionedUpdateAction;
+import com.google.spanner.executor.v1.PartitionedUpdateAction.ExecutePartitionedUpdateOptions;
 import com.google.spanner.executor.v1.QueryAction;
 import com.google.spanner.executor.v1.ReadAction;
 import com.google.spanner.executor.v1.RestoreCloudDatabaseAction;
@@ -886,6 +889,13 @@ private Status executeAction(
  } else if (action.hasExecutePartition()) {
  return executeExecutePartition(
  action.getExecutePartition(), outcomeSender, executionContext);
+ } else if (action.hasPartitionedUpdate()) {
+ if (dbPath == null) {
+ throw SpannerExceptionFactory.newSpannerException(
+ ErrorCode.INVALID_ARGUMENT, "Database path must be set for this action");
+ }
+ DatabaseClient dbClient = getClient().getDatabaseClient(DatabaseId.of(dbPath));
+ return executePartitionedUpdate(action.getPartitionedUpdate(), dbClient, outcomeSender);
  } else if (action.hasCloseBatchTxn()) {
  return executeCloseBatchTxn(action.getCloseBatchTxn(), outcomeSender, executionContext);
  } else if (action.hasExecuteChangeStreamQuery()) {
@@ -1974,6 +1984,33 @@ private Status executeExecutePartition(
  }
  }
 
+ /** Execute a partitioned update which runs different partitions in parallel. */
+ private Status executePartitionedUpdate(
+ PartitionedUpdateAction action, DatabaseClient dbClient, OutcomeSender sender) {
+ try {
+ ExecutePartitionedUpdateOptions options = action.getOptions();
+ Long count =
+ dbClient.executePartitionedUpdate(
+ Statement.of(action.getUpdate().getSql()),
+ Options.tag(options.getTag()),
+ Options.priority(RpcPriority.fromProto(options.getRpcPriority())));
+ SpannerActionOutcome outcome =
+ SpannerActionOutcome.newBuilder()
+ .setStatus(toProto(Status.OK))
+ .addDmlRowsModified(count)
+ .build();
+ sender.sendOutcome(outcome);
+ return sender.finishWithOK();
+ } catch (SpannerException e) {
+ return sender.finishWithError(toStatus(e));
+ } catch (Exception e) {
+ return sender.finishWithError(
+ toStatus(
+ SpannerExceptionFactory.newSpannerException(
+ ErrorCode.INVALID_ARGUMENT, "Unexpected error: " + e.getMessage())));
+ }
+ }
+
  /** Build a child partition record proto out of childPartitionRecord returned by client. */
  private ChildPartitionsRecord buildChildPartitionRecord(Struct childPartitionRecord)
  throws Exception {

google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java

@@ -32,13 +32,21 @@ public final class Options implements Serializable {
  public enum RpcPriority {
  LOW(Priority.PRIORITY_LOW),
  MEDIUM(Priority.PRIORITY_MEDIUM),
- HIGH(Priority.PRIORITY_HIGH);
+ HIGH(Priority.PRIORITY_HIGH),
+ UNSPECIFIED(Priority.PRIORITY_UNSPECIFIED);
 
  private final Priority proto;
 
  RpcPriority(Priority proto) {
  this.proto = Preconditions.checkNotNull(proto);
  }
+
+ public static RpcPriority fromProto(Priority proto) {
+ for (RpcPriority e : RpcPriority.values()) {
+ if (e.proto.equals(proto)) return e;
+ }
+ return RpcPriority.UNSPECIFIED;
+ }
  }
 
  /** Marker interface to mark options applicable to both Read and Query operations */

google-cloud-spanner/src/test/java/com/google/cloud/spanner/OptionsTest.java

@@ -300,6 +300,15 @@ public void testUpdateOptionsPriority() {
  assertEquals("priority: " + priority + " ", options.toString());
  }
 
+ @Test
+ public void testRpcPriorityEnumFromProto() {
+ assertEquals(RpcPriority.fromProto(Priority.PRIORITY_LOW), RpcPriority.LOW);
+ assertEquals(RpcPriority.fromProto(Priority.PRIORITY_MEDIUM), RpcPriority.MEDIUM);
+ assertEquals(RpcPriority.fromProto(Priority.PRIORITY_HIGH), RpcPriority.HIGH);
+ assertEquals(RpcPriority.fromProto(Priority.PRIORITY_UNSPECIFIED), RpcPriority.UNSPECIFIED);
+ assertEquals(RpcPriority.fromProto(null), RpcPriority.UNSPECIFIED);
+ }
+
  @Test
  public void testTransactionOptionsHashCode() {
  Options option1 = Options.fromTransactionOptions();