.Net MEVD: SqlServerVectorStore should accept a connection string to be thread safe (#11042)

adamsitnik · web-flow · commit f34bcf5c2cee · 2025-03-18T16:08:34.000Z
By accepting a connection string rather than a connection itself we can
allow the store to be thread safe and we don't need to worry about the
lifetime of provided connection.

It's a breaking change, but we really want to steer the users to do the
right thing, so the sooner we release it, the better.
diff --git a/dotnet/src/Connectors/Connectors.Memory.SqlServer/ExceptionWrapper.cs b/dotnet/src/Connectors/Connectors.Memory.SqlServer/ExceptionWrapper.cs
@@ -33,6 +33,12 @@ internal static async Task<T> WrapAsync<T>(
         }
         catch (Exception ex)
         {
+#if NET
+            await connection.DisposeAsync().ConfigureAwait(false);
+#else
+            connection.Dispose();
+#endif
+
             throw new VectorStoreOperationException(ex.Message, ex)
             {
                 OperationName = operationName,
diff --git a/dotnet/src/Connectors/Connectors.Memory.SqlServer/SqlServerVectorStore.cs b/dotnet/src/Connectors/Connectors.Memory.SqlServer/SqlServerVectorStore.cs
@@ -1,6 +1,5 @@
 ﻿// Copyright (c) Microsoft. All rights reserved.
 
-using System;
 using System.Collections.Generic;
 using System.Runtime.CompilerServices;
 using System.Threading;
@@ -12,36 +11,35 @@ namespace Microsoft.SemanticKernel.Connectors.SqlServer;
 /// <summary>
 /// An implementation of <see cref="IVectorStore"/> backed by a SQL Server or Azure SQL database.
 /// </summary>
-public sealed class SqlServerVectorStore : IVectorStore, IDisposable
+public sealed class SqlServerVectorStore : IVectorStore
 {
-    private readonly SqlConnection _connection;
+    private readonly string _connectionString;
     private readonly SqlServerVectorStoreOptions _options;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="SqlServerVectorStore"/> class.
     /// </summary>
-    /// <param name="connection">Database connection.</param>
+    /// <param name="connectionString">The connection string.</param>
     /// <param name="options">Optional configuration options.</param>
-    public SqlServerVectorStore(SqlConnection connection, SqlServerVectorStoreOptions? options = null)
+    public SqlServerVectorStore(string connectionString, SqlServerVectorStoreOptions? options = null)
     {
-        this._connection = connection;
+        Verify.NotNullOrWhiteSpace(connectionString);
+
+        this._connectionString = connectionString;
         // We need to create a copy, so any changes made to the option bag after
         // the ctor call do not affect this instance.
         this._options = options is not null
             ? new() { Schema = options.Schema }
             : SqlServerVectorStoreOptions.Defaults;
     }
 
-    /// <inheritdoc/>
-    public void Dispose() => this._connection.Dispose();
-
     /// <inheritdoc/>
     public IVectorStoreRecordCollection<TKey, TRecord> GetCollection<TKey, TRecord>(string name, VectorStoreRecordDefinition? vectorStoreRecordDefinition = null) where TKey : notnull
     {
         Verify.NotNull(name);
 
         return new SqlServerVectorStoreRecordCollection<TKey, TRecord>(
-            this._connection,
+            this._connectionString,
             name,
             new()
             {
@@ -53,9 +51,10 @@ public IVectorStoreRecordCollection<TKey, TRecord> GetCollection<TKey, TRecord>(
     /// <inheritdoc/>
     public async IAsyncEnumerable<string> ListCollectionNamesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        using SqlCommand command = SqlServerCommandBuilder.SelectTableNames(this._connection, this._options.Schema);
+        using SqlConnection connection = new(this._connectionString);
+        using SqlCommand command = SqlServerCommandBuilder.SelectTableNames(connection, this._options.Schema);
 
-        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(this._connection, command,
+        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteReaderAsync(ct),
             cancellationToken, "ListCollection").ConfigureAwait(false);
 
diff --git a/dotnet/src/Connectors/Connectors.Memory.SqlServer/SqlServerVectorStoreRecordCollection.cs b/dotnet/src/Connectors/Connectors.Memory.SqlServer/SqlServerVectorStoreRecordCollection.cs
@@ -22,23 +22,23 @@ public sealed class SqlServerVectorStoreRecordCollection<TKey, TRecord>
     private static readonly VectorSearchOptions<TRecord> s_defaultVectorSearchOptions = new();
     private static readonly SqlServerVectorStoreRecordCollectionOptions<TRecord> s_defaultOptions = new();
 
-    private readonly SqlConnection _sqlConnection;
+    private readonly string _connectionString;
     private readonly SqlServerVectorStoreRecordCollectionOptions<TRecord> _options;
     private readonly VectorStoreRecordPropertyReader _propertyReader;
     private readonly IVectorStoreRecordMapper<TRecord, IDictionary<string, object?>> _mapper;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="SqlServerVectorStoreRecordCollection{TKey, TRecord}"/> class.
     /// </summary>
-    /// <param name="connection">Database connection.</param>
+    /// <param name="connectionString">Database connection string.</param>
     /// <param name="name">The name of the collection.</param>
     /// <param name="options">Optional configuration options.</param>
     public SqlServerVectorStoreRecordCollection(
-        SqlConnection connection,
+        string connectionString,
         string name,
         SqlServerVectorStoreRecordCollectionOptions<TRecord>? options = null)
     {
-        Verify.NotNull(connection);
+        Verify.NotNullOrWhiteSpace(connectionString);
         Verify.NotNull(name);
 
         VectorStoreRecordPropertyReader propertyReader = new(typeof(TRecord),
@@ -61,7 +61,7 @@ public SqlServerVectorStoreRecordCollection(
         propertyReader.VerifyDataProperties(SqlServerConstants.SupportedDataTypes, supportEnumerable: false);
         propertyReader.VerifyVectorProperties(SqlServerConstants.SupportedVectorTypes);
 
-        this._sqlConnection = connection;
+        this._connectionString = connectionString;
         this.CollectionName = name;
         // We need to create a copy, so any changes made to the option bag after
         // the ctor call do not affect this instance.
@@ -96,10 +96,11 @@ public SqlServerVectorStoreRecordCollection(
     /// <inheritdoc/>
     public async Task<bool> CollectionExistsAsync(CancellationToken cancellationToken = default)
     {
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.SelectTableName(
-            this._sqlConnection, this._options.Schema, this.CollectionName);
+            connection, this._options.Schema, this.CollectionName);
 
-        return await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        return await ExceptionWrapper.WrapAsync(connection, command,
             static async (cmd, ct) =>
             {
                 using SqlDataReader reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
@@ -125,27 +126,29 @@ private async Task CreateCollectionAsync(bool ifNotExists, CancellationToken can
             }
         }
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.CreateTable(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             ifNotExists,
             this._propertyReader.KeyProperty,
             this._propertyReader.DataProperties,
             this._propertyReader.VectorProperties);
 
-        await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteNonQueryAsync(ct),
             cancellationToken, "CreateCollection", this.CollectionName).ConfigureAwait(false);
     }
 
     /// <inheritdoc/>
     public async Task DeleteCollectionAsync(CancellationToken cancellationToken = default)
     {
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.DropTableIfExists(
-            this._sqlConnection, this._options.Schema, this.CollectionName);
+            connection, this._options.Schema, this.CollectionName);
 
-        await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteNonQueryAsync(ct),
             cancellationToken, "DeleteCollection", this.CollectionName).ConfigureAwait(false);
     }
@@ -155,14 +158,15 @@ public async Task DeleteAsync(TKey key, CancellationToken cancellationToken = de
     {
         Verify.NotNull(key);
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.DeleteSingle(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
             key);
 
-        await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteNonQueryAsync(ct),
             cancellationToken, "Delete", this.CollectionName).ConfigureAwait(false);
     }
@@ -172,8 +176,9 @@ public async Task DeleteBatchAsync(IEnumerable<TKey> keys, CancellationToken can
     {
         Verify.NotNull(keys);
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand? command = SqlServerCommandBuilder.DeleteMany(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
@@ -184,7 +189,7 @@ public async Task DeleteBatchAsync(IEnumerable<TKey> keys, CancellationToken can
             return; // keys is empty, there is nothing to delete
         }
 
-        await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteNonQueryAsync(ct),
             cancellationToken, "DeleteBatch", this.CollectionName).ConfigureAwait(false);
     }
@@ -196,16 +201,17 @@ await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
 
         bool includeVectors = options?.IncludeVectors is true;
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.SelectSingle(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
             this._propertyReader.Properties,
             key,
             includeVectors);
 
-        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(connection, command,
             static async (cmd, ct) =>
             {
                 SqlDataReader reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
@@ -228,8 +234,9 @@ public async IAsyncEnumerable<TRecord> GetBatchAsync(IEnumerable<TKey> keys, Get
 
         bool includeVectors = options?.IncludeVectors is true;
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand? command = SqlServerCommandBuilder.SelectMany(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
@@ -242,7 +249,7 @@ public async IAsyncEnumerable<TRecord> GetBatchAsync(IEnumerable<TKey> keys, Get
             yield break; // keys is empty
         }
 
-        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteReaderAsync(ct),
             cancellationToken, "GetBatch", this.CollectionName).ConfigureAwait(false);
 
@@ -259,15 +266,16 @@ public async Task<TKey> UpsertAsync(TRecord record, CancellationToken cancellati
     {
         Verify.NotNull(record);
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand command = SqlServerCommandBuilder.MergeIntoSingle(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
             this._propertyReader.Properties,
             this._mapper.MapFromDataToStorageModel(record));
 
-        return await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        return await ExceptionWrapper.WrapAsync(connection, command,
             async static (cmd, ct) =>
             {
                 using SqlDataReader reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
@@ -282,8 +290,9 @@ public async IAsyncEnumerable<TKey> UpsertBatchAsync(IEnumerable<TRecord> record
     {
         Verify.NotNull(records);
 
+        using SqlConnection connection = new(this._connectionString);
         using SqlCommand? command = SqlServerCommandBuilder.MergeIntoMany(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             this._propertyReader.KeyProperty,
@@ -295,7 +304,7 @@ public async IAsyncEnumerable<TKey> UpsertBatchAsync(IEnumerable<TRecord> record
             yield break; // records is empty
         }
 
-        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        using SqlDataReader reader = await ExceptionWrapper.WrapAsync(connection, command,
             static (cmd, ct) => cmd.ExecuteReaderAsync(ct),
             cancellationToken, "GetBatch", this.CollectionName).ConfigureAwait(false);
 
@@ -326,8 +335,13 @@ public async Task<VectorSearchResults<TRecord>> VectorizedSearchAsync<TVector>(T
         var searchOptions = options ?? s_defaultVectorSearchOptions;
         var vectorProperty = this._propertyReader.GetVectorPropertyOrSingle(searchOptions);
 
+#pragma warning disable CA2000 // Dispose objects before losing scope
+        // This connection will be disposed by the ReadVectorSearchResultsAsync
+        // when the user is done with the results.
+        SqlConnection connection = new(this._connectionString);
+#pragma warning restore CA2000 // Dispose objects before losing scope
         using SqlCommand command = SqlServerCommandBuilder.SelectVector(
-            this._sqlConnection,
+            connection,
             this._options.Schema,
             this.CollectionName,
             vectorProperty,
@@ -336,34 +350,42 @@ public async Task<VectorSearchResults<TRecord>> VectorizedSearchAsync<TVector>(T
             searchOptions,
             allowed);
 
-        return await ExceptionWrapper.WrapAsync(this._sqlConnection, command,
+        return await ExceptionWrapper.WrapAsync(connection, command,
             (cmd, ct) =>
             {
-                var results = this.ReadVectorSearchResultsAsync(cmd, searchOptions.IncludeVectors, ct);
+                var results = this.ReadVectorSearchResultsAsync(connection, cmd, searchOptions.IncludeVectors, ct);
                 return Task.FromResult(new VectorSearchResults<TRecord>(results));
             }, cancellationToken, "VectorizedSearch", this.CollectionName).ConfigureAwait(false);
     }
 
     private async IAsyncEnumerable<VectorSearchResult<TRecord>> ReadVectorSearchResultsAsync(
+        SqlConnection connection,
         SqlCommand command,
         bool includeVectors,
         [EnumeratorCancellation] CancellationToken cancellationToken)
     {
-        StorageToDataModelMapperOptions options = new() { IncludeVectors = includeVectors };
-        var vectorPropertyStoragePropertyNames = includeVectors ? this._propertyReader.VectorPropertyStoragePropertyNames : [];
-        using SqlDataReader reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
-
-        int scoreIndex = -1;
-        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
+        try
         {
-            if (scoreIndex < 0)
+            StorageToDataModelMapperOptions options = new() { IncludeVectors = includeVectors };
+            var vectorPropertyStoragePropertyNames = includeVectors ? this._propertyReader.VectorPropertyStoragePropertyNames : [];
+            using SqlDataReader reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
+
+            int scoreIndex = -1;
+            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
             {
-                scoreIndex = reader.GetOrdinal("score");
+                if (scoreIndex < 0)
+                {
+                    scoreIndex = reader.GetOrdinal("score");
+                }
+
+                yield return new VectorSearchResult<TRecord>(
+                    this._mapper.MapFromStorageToDataModel(new SqlDataReaderDictionary(reader, vectorPropertyStoragePropertyNames), options),
+                    reader.GetDouble(scoreIndex));
             }
-
-            yield return new VectorSearchResult<TRecord>(
-                this._mapper.MapFromStorageToDataModel(new SqlDataReaderDictionary(reader, vectorPropertyStoragePropertyNames), options),
-                reader.GetDouble(scoreIndex));
+        }
+        finally
+        {
+            connection.Dispose();
         }
     }
 }
diff --git a/dotnet/src/VectorDataIntegrationTests/SqlServerIntegrationTests/SqlServerVectorStoreTests.cs b/dotnet/src/VectorDataIntegrationTests/SqlServerIntegrationTests/SqlServerVectorStoreTests.cs
@@ -159,8 +159,7 @@ public async Task CustomMapper()
         {
             Mapper = mapper
         };
-        using SqlConnection connection = new(SqlServerTestEnvironment.ConnectionString);
-        SqlServerVectorStoreRecordCollection<string, TestModel> collection = new(connection, collectionName, options);
+        SqlServerVectorStoreRecordCollection<string, TestModel> collection = new(SqlServerTestEnvironment.ConnectionString!, collectionName, options);
 
         try
         {
diff --git a/dotnet/src/VectorDataIntegrationTests/SqlServerIntegrationTests/Support/SqlServerTestStore.cs b/dotnet/src/VectorDataIntegrationTests/SqlServerIntegrationTests/Support/SqlServerTestStore.cs

Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,12 @@ internal static async Task<T> WrapAsync<T>(`
`33`	`33`	`}`
`34`	`34`	`catch (Exception ex)`
`35`	`35`	`{`
	`36`	`+#if NET`
	`37`	`+ await connection.DisposeAsync().ConfigureAwait(false);`
	`38`	`+#else`
	`39`	`+ connection.Dispose();`
	`40`	`+#endif`
	`41`	`+`
`36`	`42`	`throw new VectorStoreOperationException(ex.Message, ex)`
`37`	`43`	`{`
`38`	`44`	`OperationName = operationName,`
Original file line number	Diff line number	Diff line change
`@@ -159,8 +159,7 @@ public async Task CustomMapper()`
`159`	`159`	`{`
`160`	`160`	`Mapper = mapper`
`161`	`161`	`};`
`162`		`- using SqlConnection connection = new(SqlServerTestEnvironment.ConnectionString);`
`163`		`- SqlServerVectorStoreRecordCollection<string, TestModel> collection = new(connection, collectionName, options);`
	`162`	`+ SqlServerVectorStoreRecordCollection<string, TestModel> collection = new(SqlServerTestEnvironment.ConnectionString!, collectionName, options);`
`164`	`163`
`165`	`164`	`try`
`166`	`165`	`{`