feat: Spanner.DATA impl and adapt integration tests to multiplex sessions

test: Fix Mutation tests to be in keeping with mux sessions

test: Fix unit tests

Author: purvavasudeo

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data.CommonTesting/SpannerTestDatabaseBase.cs

@@ -200,7 +200,7 @@ public async Task<ManagedSession> GetManagedSession()
     private async Task<ManagedSession> CreateMultiplexSession(ManagedSessionOptions options)
     {
         var poolManager = SessionPoolManager.Create(options);
-        var muxSession = await poolManager.AcquireMultiplexSessionAsync(SpannerClientCreationOptions, DatabaseName, null);
+        var muxSession = await poolManager.AcquireManagedSessionAsync(SpannerClientCreationOptions, DatabaseName, null);
 
         return muxSession;
     }

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

@@ -95,8 +95,6 @@ StringValue         STRING(MAX),
                 var dropCommand = connection.CreateDdlCommand($"DROP DATABASE {dbName}");
                 await dropCommand.ExecuteNonQueryAsync();
             }
-
-            await SessionPoolHelpers.ShutdownPoolAsync(builder.WithDatabase(dbName));
         }
 
         [Fact]
@@ -146,8 +144,6 @@ StringValue         STRING(MAX),
                 var dropCommand = connection.CreateDdlCommand($"DROP DATABASE {dbName}");
                 await dropCommand.ExecuteNonQueryAsync();
             }
-
-            await SessionPoolHelpers.ShutdownPoolAsync(builder.WithDatabase(dbName));
         }
 
         [Fact]

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

@@ -1,4 +1,4 @@
-// Copyright 2018 Google LLC
+// Copyright 2018 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

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

@@ -65,7 +65,8 @@ private async Task AssertEmptyMutationsFailAsync(SpannerCommand emptyMutation)
         // "the amount of values does not match the number of columns in the key".
         if (!_fixture.RunningOnEmulator) // The message is different on the emulator.
         {
-            Assert.Contains("does not specify any value", exception.RpcException.Message);
+            // This error is expected for Multiplex Sesions as they expect a mutation key during commit for mutation only transactions
+            Assert.Contains("Failed to initialize transaction due to invalid mutation key.", exception.RpcException.Message);
         }
     }
 

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

@@ -71,10 +71,6 @@ public async Task BadDbName()
                 Assert.Equal(ErrorCode.NotFound, e.ErrorCode);
                 Assert.False(e.IsTransientSpannerFault());
             }
-
-            // Shut the pool associated with the bad database down, to avoid seeing spurious connection failures
-            // later in the log.
-            await SessionPoolHelpers.ShutdownPoolAsync(connectionString);
         }
 
         [Fact]

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

@@ -150,30 +150,21 @@ private async Task RunStress(Func<SpannerConnectionStringBuilder, Task> func, in
             // The maximum roundtrip time for Spanner (and MySQL) is about 200ms per
             // write. If we initialize with the target sustained # sessions,
             // we shouldn't see any more sessions created.
-            int countToPreWarm = Math.Min(TargetQps / 4, 800);
-            var options = new SessionPoolOptions
-            {
-                MaximumActiveSessions = Math.Max(countToPreWarm + 50, 400),
-                MinimumPooledSessions = countToPreWarm,
-                MaximumConcurrentSessionCreates = Math.Min(countToPreWarm, 50)
-            };
+            var options = new ManagedSessionOptions();
 
             var sessionPoolManager = SessionPoolManager.Create(options);
             var connectionStringBuilder = new SpannerConnectionStringBuilder(_fixture.ConnectionString)
             {
                 SessionPoolManager = sessionPoolManager,
                 MaximumGrpcChannels = Math.Max(4, 8 * TargetQps / 2000)
             };
-            var pool = await connectionStringBuilder.AcquireSessionPoolAsync();
+            var managedSession = await connectionStringBuilder.AcquireManagedSessionAsync();
             var logger = Logger.DefaultLogger;
             logger.ResetPerformanceData();
 
             logger.Info("Prewarming session pool for stress test");
             // Prewarm step: allow up to 30 seconds for the session pool to be populated.
             var cancellationToken = new CancellationTokenSource(30000).Token;
-            await pool.WhenPoolReady(_fixture.DatabaseName, cancellationToken);
-
-            logger.Info($"Prewarm complete. Pool stats: {pool.GetSegmentStatisticsSnapshot(_fixture.DatabaseName)}");
 
             // Now run the test, with performance logging enabled, but without debug logging.
             // (Debug logging can write a lot to our log file, breaking the test.)
@@ -193,8 +184,6 @@ private async Task RunStress(Func<SpannerConnectionStringBuilder, Task> func, in
             }
             logger.Info($"Spanner latency = {latencyMs}ms");
 
-            await SessionPoolHelpers.ShutdownPoolAsync(connectionStringBuilder);
-
             // Spanner latency with 100 qps simulated is usually around 75ms.
             // We allow for a latency multiplier from callers, because callers may be executing,
             // more than one command. In particular, with inline transactions, mutation commits

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

@@ -86,50 +86,6 @@ private async Task IncrementByOneAsync(SpannerConnection connection, bool orphan
             }
         }
 
-        [Fact]
-        public async Task Commit_ReturnsToPool()
-        {
-            using var connection = new SpannerConnection(_fixture.ConnectionString);
-            await connection.OpenAsync();
-
-            using var transaction = await connection.BeginTransactionAsync();
-            using var command = connection.CreateSelectCommand($"SELECT Int64Value FROM {_fixture.TableName} WHERE K=@k");
-            command.Parameters.Add("k", SpannerDbType.String, _key);
-            command.Transaction = transaction;
-
-            var value = await command.ExecuteScalarAsync();
-
-            transaction.Commit();
-
-            var poolStatistics = connection.GetSessionPoolSegmentStatistics();
-
-            // Because the session is eagerly returned to the pool after a commit, there shouldn't
-            // be any active sessions even before we dispose of the transaction explicitly.
-            Assert.Equal(0, poolStatistics.ActiveSessionCount);
-        }
-
-        [Fact]
-        public async Task Rollback_ReturnsToPool()
-        {
-            using var connection = new SpannerConnection(_fixture.ConnectionString);
-            await connection.OpenAsync();
-
-            using var transaction = await connection.BeginTransactionAsync();
-            using var command = connection.CreateSelectCommand($"SELECT Int64Value FROM {_fixture.TableName} WHERE K=@k");
-            command.Parameters.Add("k", SpannerDbType.String, _key);
-            command.Transaction = transaction;
-
-            var value = await command.ExecuteScalarAsync();
-
-            transaction.Rollback();
-
-            var poolStatistics = connection.GetSessionPoolSegmentStatistics();
-
-            // Because the session is eagerly returned to the pool after a rollback, there shouldn't
-            // be any active sessions even before we dispose of the transaction explicitly.
-            Assert.Equal(0, poolStatistics.ActiveSessionCount);
-        }
-
         [Fact]
         public async Task DetachOnDisposeTransactionIsDetached()
         {

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data.IntegrationTests/V1/Internal/ExecuteHelperTests.cs

@@ -30,37 +30,17 @@ public class ExecuteHelperTests
     public ExecuteHelperTests(SpannerDatabaseFixture fixture) =>
         _fixture = fixture;
 
-    [Fact]
-    public Task SessionNotFound() => WithSessionPool(async pool =>
-    {
-        var session = await pool.Client.CreateSessionAsync(_fixture.DatabaseName);
-        await pool.Client.DeleteSessionAsync(session.SessionName);
-
-        // The session doesn't become invalid immediately after deletion.
-        // Wait for a minute to ensure the session is really expired.
-        await Task.Delay(TimeSpan.FromMinutes(1));
-
-        var request = new ExecuteSqlRequest
-        {
-            Sql = $"SELECT 1",
-            Session = session.Name
-        };
-        var exception = await Assert.ThrowsAsync<RpcException>(() => pool.Client.ExecuteSqlAsync(request));
-        Assert.True(ExecuteHelper.IsSessionExpiredError(exception));
-    });
-
     // This code is separated out in case we need more tests. It's really just fluff.
-    private async Task WithSessionPool(Func<SessionPool, Task> action)
+    private async Task WithManagedSession(Func<ManagedSession, Task> action)
     {
         var builder = new SpannerConnectionStringBuilder(_fixture.ConnectionString);
-        var pool = await builder.AcquireSessionPoolAsync();
+        var managedSession = await builder.AcquireManagedSessionAsync();
         try
         {
-            await action(pool);
+            await action(managedSession);
         }
         finally
         {
-            builder.SessionPoolManager.Release(pool);
         }
     }
 }

apis/Google.Cloud.Spanner.Data/Google.Cloud.Spanner.Data.IntegrationTests/V1/ReliableStreamReaderTests.cs

@@ -1,4 +1,4 @@
-// Copyright 2019 Google LLC
+// Copyright 2019 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -14,6 +14,7 @@
 
 using Google.Cloud.Spanner.Data;
 using Google.Cloud.Spanner.Data.IntegrationTests;
+using Google.Cloud.Spanner.V1.Internal.Logging;
 using Google.Protobuf.WellKnownTypes;
 using System.Threading.Tasks;
 using Xunit;
@@ -41,12 +42,14 @@ public async Task HasDataAsync(string key, int expectedValueCount)
                 ParamTypes = { { "Key", new Type { Code = TypeCode.String } } }
             };
             var builder = new SpannerConnectionStringBuilder(_fixture.ConnectionString);
-            var pool = await builder.AcquireSessionPoolAsync();
+            var managedSession = await builder.AcquireManagedSessionAsync();
             try
             {
-                using (var pooledSession = await pool.AcquireSessionAsync(_fixture.DatabaseName, null, default))
+                using (SpannerConnection connection = new SpannerConnection(builder))
                 {
-                    using (var reader = pooledSession.ExecuteSqlStreamReader(request, null))
+                    await connection.OpenAsync(default);
+                    ManagedTransaction managedTransaction = connection.AcquireManagedTransaction(null, out _);
+                    using (var reader = managedTransaction.ExecuteSqlStreamReader(request, null))
                     {
                         // While there are more values to read, HasDataAsync should return true
                         for (int valuesRead = 0; valuesRead < expectedValueCount; valuesRead++)
@@ -69,7 +72,7 @@ public async Task HasDataAsync(string key, int expectedValueCount)
             }
             finally
             {
-                builder.SessionPoolManager.Release(pool);
+                // Nothing to clean here
             }
         }
     }

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

@@ -334,7 +334,7 @@ public async Task BadColumnType()
                 cmd.Parameters.Add("K", SpannerDbType.Float64, 0.1);
                 var e = await Assert.ThrowsAsync<SpannerException>(() => cmd.ExecuteNonQueryAsyncWithRetry());
                 Logger.DefaultLogger.Debug($"BadColumnType: Caught error code: {e.ErrorCode}");
-                Assert.Equal(ErrorCode.FailedPrecondition, e.ErrorCode);
+                Assert.Equal(ErrorCode.InvalidArgument, e.ErrorCode);
                 Assert.False(e.IsTransientSpannerFault());
             }
         }