feat: add EnableImplicitSession support

src/Ydb.Sdk/src/Ado/PoolManager.cs

@@ -33,10 +33,22 @@ CancellationToken cancellationToken
                          !cacheDriver.IsDisposed
                 ? cacheDriver
                 : Drivers[settings.GrpcConnectionString] = await settings.BuildDriver();
-            driver.RegisterOwner();
 
-            var factory = new PoolingSessionFactory(driver, settings);
-            var newSessionPool = new PoolingSessionSource<PoolingSession>(factory, settings);
+            ISessionSource newSessionPool;
+            if (settings.EnableImplicitSession)
+            {
+                var key = settings.ConnectionString;
+                newSessionPool = new ImplicitSessionSource(
+                    driver,
+                    onEmpty: () => Pools.TryRemove(key, out _)
+                );
+            }
+            else
+            {
+                driver.RegisterOwner();
+                var factory = new PoolingSessionFactory(driver, settings);
+                newSessionPool = new PoolingSessionSource<PoolingSession>(factory, settings);
+            }
 
             Pools[settings.ConnectionString] = newSessionPool;
 

src/Ydb.Sdk/src/Ado/Session/ImplicitSession.cs

@@ -5,9 +5,12 @@ namespace Ydb.Sdk.Ado.Session;
 
 internal class ImplicitSession : ISession
 {
-    public ImplicitSession(IDriver driver)
+    private readonly Action? _onClose;
+
+    public ImplicitSession(IDriver driver, Action? onClose = null)
     {
         Driver = driver;
+        _onClose = onClose;
     }
 
     public IDriver Driver { get; }
@@ -47,9 +50,7 @@ public void OnNotSuccessStatusCode(StatusCode code)
     {
     }
 
-    public void Close()
-    {
-    }
+    public void Close() => _onClose?.Invoke();
 
     private static YdbException NotSupportedTransaction =>
         new(StatusCode.BadRequest, "Transactions are not supported in implicit sessions");

src/Ydb.Sdk/src/Ado/Session/ImplicitSessionSource.cs

@@ -0,0 +1,61 @@
+namespace Ydb.Sdk.Ado.Session;
+
+internal sealed class ImplicitSessionSource : ISessionSource
+{
+    private readonly IDriver _driver;
+    private readonly Action? _onBecameEmpty;
+    private int _isDisposed;
+    private int _activeLeaseCount;
+
+    internal ImplicitSessionSource(IDriver driver, Action? onEmpty = null)
+    {
+        _driver = driver ?? throw new ArgumentNullException(nameof(driver));
+        _onBecameEmpty = onEmpty;
+    }
+
+    public ValueTask<ISession> OpenSession(CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (!TryAcquireLease())
+            throw new ObjectDisposedException(nameof(ImplicitSessionSource));
+
+        return new ValueTask<ISession>(new ImplicitSession(_driver, ReleaseLease));
+    }
+
+    private bool TryAcquireLease()
+    {
+        if (Volatile.Read(ref _isDisposed) != 0)
+            return false;
+
+        Interlocked.Increment(ref _activeLeaseCount);
+
+        if (Volatile.Read(ref _isDisposed) != 0)
+        {
+            Interlocked.Decrement(ref _activeLeaseCount);
+            return false;
+        }
+
+        return true;
+    }
+
+    private void ReleaseLease()
+    {
+        if (Interlocked.Decrement(ref _activeLeaseCount) == 0)
+        {
+            _onBecameEmpty?.Invoke();
+        }
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        Interlocked.Exchange(ref _isDisposed, 1);
+
+        if (Volatile.Read(ref _activeLeaseCount) == 0)
+        {
+            _onBecameEmpty?.Invoke();
+        }
+
+        return ValueTask.CompletedTask;
+    }
+}

src/Ydb.Sdk/src/Ado/YdbCommand.cs

@@ -216,9 +216,19 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
             throw new InvalidOperationException("Transaction mismatched! (Maybe using another connection)");
         }
 
-        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(await YdbConnection.Session.ExecuteQuery(
-            preparedSql.ToString(), ydbParameters, execSettings, transaction?.TransactionControl
-        ), YdbConnection.OnNotSuccessStatusCode, transaction, cancellationToken);
+        var execResult = await YdbConnection.Session.ExecuteQuery(
+            preparedSql.ToString(),
+            ydbParameters,
+            execSettings,
+            transaction?.TransactionControl
+        );
+
+        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(
+            execResult,
+            YdbConnection.OnNotSuccessStatusCode,
+            transaction,
+            cancellationToken
+        );
 
         YdbConnection.LastReader = ydbDataReader;
         YdbConnection.LastCommand = CommandText;

src/Ydb.Sdk/src/Ado/YdbConnection.cs

@@ -148,7 +148,7 @@ public override string ConnectionString
 
     public override ConnectionState State => ConnectionState;
 
-    private ConnectionState ConnectionState { get; set; } = ConnectionState.Closed; // Invoke AsyncOpen()
+    private ConnectionState ConnectionState { get; set; } = ConnectionState.Closed;
 
     internal void OnNotSuccessStatusCode(StatusCode code)
     {

src/Ydb.Sdk/src/Ado/YdbConnectionStringBuilder.cs

@@ -40,6 +40,7 @@ private void InitDefaultValues()
         _maxReceiveMessageSize = GrpcDefaultSettings.MaxReceiveMessageSize;
         _disableDiscovery = GrpcDefaultSettings.DisableDiscovery;
         _disableServerBalancer = false;
+        _enableImplicitSession = false;
     }
 
     public string Host
@@ -314,6 +315,18 @@ public int CreateSessionTimeout
 
     private int _createSessionTimeout;
 
+    public bool EnableImplicitSession
+    {
+        get => _enableImplicitSession;
+        set
+        {
+            _enableImplicitSession = value;
+            SaveValue(nameof(EnableImplicitSession), value);
+        }
+    }
+
+    private bool _enableImplicitSession;
+
     public ILoggerFactory LoggerFactory { get; init; } = NullLoggerFactory.Instance;
 
     public ICredentialsProvider? CredentialsProvider { get; init; }
@@ -495,6 +508,9 @@ static YdbConnectionOption()
             AddOption(new YdbConnectionOption<bool>(BoolExtractor,
                     (builder, disableServerBalancer) => builder.DisableServerBalancer = disableServerBalancer),
                 "DisableServerBalancer", "Disable Server Balancer");
+            AddOption(new YdbConnectionOption<bool>(BoolExtractor,
+                    (builder, enableImplicitSession) => builder.EnableImplicitSession = enableImplicitSession),
+                "EnableImplicitSession", "ImplicitSession");
         }
 
         private static void AddOption(YdbConnectionOption option, params string[] keys)

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/PoolManagerTests.cs

@@ -3,8 +3,10 @@
 
 namespace Ydb.Sdk.Ado.Tests;
 
-[Collection("PoolManagerTests")]
 [CollectionDefinition("PoolManagerTests", DisableParallelization = true)]
+public sealed class PoolManagerCollection;
+
+[Collection("PoolManagerTests")]
 public class PoolManagerTests
 {
     [Theory]

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/YdbCommandTests.cs

@@ -204,6 +204,127 @@ public async Task ExecuteScalar_WhenSelectNoRows_ReturnNull()
             .ExecuteScalarAsync());
     }
 
+    [Fact]
+    public async Task ImplicitSession_SimpleScalar_Works()
+    {
+        await using var connection = CreateConnection();
+        connection.ConnectionString += ";EnableImplicitSession=true";
+        await connection.OpenAsync();
+
+        var cmd = connection.CreateCommand();
+        cmd.CommandText = "SELECT 40 + 2;";
+        var scalar = await cmd.ExecuteScalarAsync();
+        Assert.Equal(42, Convert.ToInt32(scalar));
+    }
+
+    [Fact]
+    public async Task ImplicitSession_RepeatedScalars_WorksManyTimes()
+    {
+        await using var connection = CreateConnection();
+        connection.ConnectionString += ";EnableImplicitSession=true";
+        await connection.OpenAsync();
+
+        for (var i = 0; i < 30; i++)
+        {
+            var cmd = connection.CreateCommand();
+            cmd.CommandText = $"SELECT {i};";
+            var scalar = await cmd.ExecuteScalarAsync();
+            Assert.Equal(i, Convert.ToInt32(scalar));
+        }
+    }
+
+    [Fact]
+    public void ImplicitSession_ConcurrentCommand_IsStillBlockedByBusyCheck()
+    {
+        using var connection = CreateConnection();
+        connection.ConnectionString += ";EnableImplicitSession=true";
+        connection.Open();
+
+        var cmd = connection.CreateCommand();
+        cmd.CommandText = "SELECT 1; SELECT 1;";
+        using var reader = cmd.ExecuteReader();
+
+        var ex = Assert.Throws<YdbOperationInProgressException>(() => cmd.ExecuteReader());
+        Assert.Equal("A command is already in progress: SELECT 1; SELECT 1;", ex.Message);
+    }
+
+    [Fact]
+    public async Task ImplicitSession_DisallowsTransactions_And_AllowsNonTransactionalCommands()
+    {
+        var table = $"Implicit_{Guid.NewGuid():N}";
+
+        await using var connection = CreateConnection();
+        connection.ConnectionString += ";EnableImplicitSession=true";
+        await connection.OpenAsync();
+
+        try
+        {
+            await using (var create = connection.CreateCommand())
+            {
+                create.CommandText = $"""
+                                      CREATE TABLE {table} (
+                                          Id Int32,
+                                          Name Text,
+                                          PRIMARY KEY (Id)
+                                      )
+                                      """;
+                await create.ExecuteNonQueryAsync();
+            }
+
+            await using (var insert = connection.CreateCommand())
+            {
+                insert.CommandText = $"INSERT INTO {table} (Id, Name) VALUES (1, 'A');";
+                await insert.ExecuteNonQueryAsync();
+            }
+
+            var tx = connection.BeginTransaction();
+            await using (var insertTx = connection.CreateCommand())
+            {
+                insertTx.Transaction = tx;
+                insertTx.CommandText = $"INSERT INTO {table} (Id, Name) VALUES (2, 'B');";
+                var ex = await Assert.ThrowsAsync<YdbException>(async () => await insertTx.ExecuteNonQueryAsync());
+                Assert.Contains("Transactions are not supported in implicit sessions", ex.Message);
+            }
+
+            await tx.RollbackAsync();
+
+            await using (var check = connection.CreateCommand())
+            {
+                check.CommandText = $"SELECT COUNT(*) FROM {table};";
+                var count = Convert.ToInt32(await check.ExecuteScalarAsync());
+                Assert.Equal(1, count);
+            }
+        }
+        finally
+        {
+            await using var drop = connection.CreateCommand();
+            drop.CommandText = $"DROP TABLE {table}";
+            await drop.ExecuteNonQueryAsync();
+        }
+    }
+
+    [Fact]
+    public async Task ImplicitSession_Cancellation_AfterFirstResult_StillReturnsFirst()
+    {
+        await using var connection = CreateConnection();
+        connection.ConnectionString += ";EnableImplicitSession=true";
+        await connection.OpenAsync();
+
+        var cmd = new YdbCommand(connection) { CommandText = "SELECT 1; SELECT 1;" };
+        using var cts = new CancellationTokenSource();
+
+        var reader = await cmd.ExecuteReaderAsync(cts.Token);
+
+        await reader.ReadAsync(cts.Token);
+        Assert.Equal(1, reader.GetValue(0));
+        Assert.True(await reader.NextResultAsync(cts.Token));
+
+        await cts.CancelAsync();
+
+        await reader.ReadAsync(cts.Token);
+        Assert.Equal(1, reader.GetValue(0));
+        Assert.False(await reader.NextResultAsync());
+    }
 
     public class Data<T>(DbType dbType, T expected, bool isNullable = false)
     {

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/YdbConnectionStringBuilderTests.cs

@@ -30,6 +30,7 @@ public void InitDefaultValues_WhenEmptyConstructorInvoke_ReturnDefaultConnection
         Assert.False(ydbConnectionStringBuilder.DisableDiscovery);
         Assert.False(ydbConnectionStringBuilder.DisableServerBalancer);
         Assert.False(ydbConnectionStringBuilder.UseTls);
+        Assert.False(ydbConnectionStringBuilder.EnableImplicitSession);
 
         Assert.Equal("UseTls=False;Host=localhost;Port=2136;Database=/local;User=;Password=;ConnectTimeout=5;" +
                      "KeepAlivePingDelay=10;KeepAlivePingTimeout=10;EnableMultipleHttp2Connections=False;" +
@@ -54,7 +55,7 @@ public void InitConnectionStringBuilder_WhenExpectedKeys_ReturnUpdatedConnection
             "Host=server;Port=2135;Database=/my/path;User=Kirill;UseTls=true;MinSessionPool=10;MaxSessionPool=50;" +
             "CreateSessionTimeout=30;SessionIdleTimeout=600;ConnectTimeout=30;KeepAlivePingDelay=30;" +
             "KeepAlivePingTimeout=60;EnableMultipleHttp2Connections=true;MaxSendMessageSize=1000000;" +
-            "MaxReceiveMessageSize=1000000;DisableDiscovery=true;DisableServerBalancer=true;"
+            "MaxReceiveMessageSize=1000000;DisableDiscovery=true;DisableServerBalancer=true;EnableImplicitSession=true;"
         );
 
         Assert.Equal(2135, ydbConnectionStringBuilder.Port);
@@ -78,9 +79,11 @@ public void InitConnectionStringBuilder_WhenExpectedKeys_ReturnUpdatedConnection
                      "ConnectTimeout=30;KeepAlivePingDelay=30;KeepAlivePingTimeout=60;" +
                      "EnableMultipleHttp2Connections=True;" +
                      "MaxSendMessageSize=1000000;MaxReceiveMessageSize=1000000;" +
-                     "DisableDiscovery=True;DisableServerBalancer=True", ydbConnectionStringBuilder.ConnectionString);
+                     "DisableDiscovery=True;DisableServerBalancer=True;EnableImplicitSession=True",
+            ydbConnectionStringBuilder.ConnectionString);
         Assert.True(ydbConnectionStringBuilder.DisableDiscovery);
         Assert.True(ydbConnectionStringBuilder.DisableServerBalancer);
+        Assert.True(ydbConnectionStringBuilder.EnableImplicitSession);
         Assert.Equal("UseTls=True;Host=server;Port=2135;Database=/my/path;User=Kirill;Password=;ConnectTimeout=30;" +
                      "KeepAlivePingDelay=30;KeepAlivePingTimeout=60;EnableMultipleHttp2Connections=True;" +
                      "MaxSendMessageSize=1000000;MaxReceiveMessageSize=1000000;DisableDiscovery=True",