feat: Spanner non-awaiting DDL

Adds a method to execute DDL on Spanner without having the client awaiting the
long-running operation to finish. This makes it possible to for example start
the creation of a secondary index using the client library, without having the
library waiting for that operation to finish. Creating a secondary index on a
table with a large amount of data can take a very long time (several days).

Author: olavloite

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data.IntegrationTests/AdminTests.cs

@@ -239,17 +239,19 @@ AlbumTitle STRING(MAX),
             using (var connection = new SpannerConnection(builder.WithDatabase(dbName)))
             {
                 var dropSingersCmd = connection.CreateDdlCommand("DROP TABLE Singers");
-                var dropAlbumsCmd = connection.CreateDdlCommand("DROP TABLE Albums");
+                var dropBothTablesCmd = connection.CreateDdlCommand("DROP TABLE Albums", "DROP TABLE Singers");
 
                 await Assert.ThrowsAsync<SpannerException>(() => dropSingersCmd.ExecuteNonQueryAsync());
-                await dropAlbumsCmd.ExecuteNonQueryAsync();
-                await dropSingersCmd.ExecuteNonQueryAsync();
+                var operation = await dropBothTablesCmd.ExecuteDdlAsync();
+                Assert.NotNull(operation?.Name);
             }
 
             using (var connection = new SpannerConnection(builder))
             {
                 var dropCommand = connection.CreateDdlCommand($"DROP DATABASE {dbName}");
-                await dropCommand.ExecuteNonQueryAsync();
+                var operation = await dropCommand.ExecuteDdlAsync();
+                // DropDatabase does not return a long-running operation.
+                Assert.Null(operation);
             }
         }
     }

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data/SpannerCommand.ExecutableCommand.cs

@@ -17,6 +17,7 @@
 using Google.Cloud.Spanner.Admin.Database.V1;
 using Google.Cloud.Spanner.Common.V1;
 using Google.Cloud.Spanner.V1;
+using Google.LongRunning;
 using Google.Protobuf;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
@@ -181,7 +182,7 @@ internal Task<int> ExecuteNonQueryAsync(CancellationToken cancellationToken)
                 switch (CommandTextBuilder.SpannerCommandType)
                 {
                     case SpannerCommandType.Ddl:
-                        return ExecuteDdlAsync(cancellationToken);
+                        return ExecuteDdlAsync(pollUntilCompleted: true, cancellationToken).ContinueWith(_ => 1, TaskContinuationOptions.ExecuteSynchronously);
                     case SpannerCommandType.Delete:
                     case SpannerCommandType.Insert:
                     case SpannerCommandType.InsertOrUpdate:
@@ -243,12 +244,13 @@ private async Task<SpannerDataReader> ExecuteDmlReaderAsync(CommandBehavior beha
                 return new SpannerDataReader(Connection.Logger, resultSet, Transaction?.ReadTimestamp, resourceToClose, ConversionOptions, enableGetSchemaTable, CommandTimeout);
             }
 
-            private async Task<int> ExecuteDdlAsync(CancellationToken cancellationToken)
+            internal async Task<Operation> ExecuteDdlAsync(bool pollUntilCompleted, CancellationToken cancellationToken)
             {
                 string commandText = CommandTextBuilder.CommandText;
                 var builder = Connection.Builder;
                 var channelOptions = new SpannerClientCreationOptions(builder);
                 var credentials = await channelOptions.GetCredentialsAsync().ConfigureAwait(false);
+                Operation operation = null;
 
                 // Create the builder separately from actually building, so we can note the channel that it created.
                 // (This is fairly unpleasant, but we'll try to improve this in the next version of GAX.)
@@ -274,7 +276,11 @@ private async Task<int> ExecuteDdlAsync(CancellationToken cancellationToken)
                             ProtoDescriptors = CommandTextBuilder.ProtobufDescriptors?.ToByteString() ?? ByteString.Empty,
                         };
                         var response = await databaseAdminClient.CreateDatabaseAsync(request).ConfigureAwait(false);
-                        response = await response.PollUntilCompletedAsync().ConfigureAwait(false);
+                        operation = response.RpcMessage;
+                        if (pollUntilCompleted)
+                        {
+                            response = await response.PollUntilCompletedAsync().ConfigureAwait(false);
+                        }
                         if (response.IsFaulted)
                         {
                             throw SpannerException.FromOperationFailedException(response.Exception);
@@ -311,7 +317,11 @@ private async Task<int> ExecuteDdlAsync(CancellationToken cancellationToken)
                         };
 
                         var response = await databaseAdminClient.UpdateDatabaseDdlAsync(request).ConfigureAwait(false);
-                        response = await response.PollUntilCompletedAsync().ConfigureAwait(false);
+                        operation = response.RpcMessage;
+                        if (pollUntilCompleted)
+                        {
+                            response = await response.PollUntilCompletedAsync().ConfigureAwait(false);
+                        }
                         if (response.IsFaulted)
                         {
                             throw SpannerException.FromOperationFailedException(response.Exception);
@@ -328,7 +338,7 @@ private async Task<int> ExecuteDdlAsync(CancellationToken cancellationToken)
                     channel?.Shutdown();
                 }
 
-                return 0;
+                return operation;
             }
 
             private async Task<int> ExecuteMutationsAsync(CancellationToken cancellationToken)

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data/SpannerCommand.cs

@@ -14,6 +14,7 @@
 
 using Google.Api.Gax;
 using Google.Cloud.Spanner.V1;
+using Google.LongRunning;
 using System;
 using System.Collections.Generic;
 using System.Data;
@@ -507,6 +508,39 @@ public long ExecutePartitionedUpdate() =>
         public Task<long> ExecutePartitionedUpdateAsync(CancellationToken cancellationToken = default) =>
             CreateExecutableCommand().ExecutePartitionedUpdateAsync(cancellationToken);
 
+        /// <summary>
+        /// Executes this command as DDL. The command must contain one or more DDL statements;
+        /// <see cref="SpannerConnection.CreateDdlCommand(string, string[])"/> for details.
+        /// </summary>
+        /// <param name="pollUntilCompleted">
+        /// If true, the returned task will wait for the DDL operation to finish execution on Spanner.
+        /// Otherwise, the returned task will finish as soon as the DDL operation has successfully been started on
+        /// Spanner, and the caller can monitor the progress of the long-running operation.
+        /// </param>
+        /// <returns>
+        /// A reference to the long-running operation that was started for the DDL statement(s).
+        /// Note: The operation is null for DropDatabase commands.
+        /// </returns>
+        public Operation ExecuteDdl(bool pollUntilCompleted = false) =>
+            Task.Run(() => ExecuteDdlAsync(pollUntilCompleted, _synchronousCancellationTokenSource.Token)).ResultWithUnwrappedExceptions();
+
+        /// <summary>
+        /// Executes this command as DDL. The command must contain one or more DDL statements;
+        /// <see cref="SpannerConnection.CreateDdlCommand(string, string[])"/> for details.
+        /// </summary>
+        /// <param name="pollUntilCompleted">
+        /// If true, the returned task will wait for the DDL operation to finish execution on Spanner.
+        /// Otherwise, the returned task will finish as soon as the DDL operation has successfully been started on
+        /// Spanner, and the caller can monitor the progress of the long-running operation.
+        /// </param>
+        /// <param name="cancellationToken">An optional token for canceling the call.</param>
+        /// <returns>
+        /// A reference to the long-running operation that was started for the DDL statement(s).
+        /// Note: The operation is null for DropDatabase commands.
+        /// </returns>
+        public Task<Operation> ExecuteDdlAsync(bool pollUntilCompleted = false, CancellationToken cancellationToken = default) =>
+            CreateExecutableCommand().ExecuteDdlAsync(pollUntilCompleted, cancellationToken);
+
         /// <summary>
         /// Creates an executable command that captures all the necessary information from this command.
         /// </summary>