feat: support ADO.NET GetSchema & dev: fixes (#247)

- Supported in ADO.NET GetSchema(Async). CollectionNames: 
   * Tables
   * TablesWithCollections
   * DataSourceInformation
   * MetaDataCollections
   * Restrictions
- Rename field _onStatus -> _onNotSuccessStatus in YdbDataReader
- If session is not active, do not invoke DeleteSession(session)
- AttachStream: connect stream using NodeId
- PoolManager: change pool properties on field
- Delete *Settings.DefaultInstance because it's a singleton object that's changed by tasks when NodeId is set
- DbConnection.Session.OnStatus(status) in YdbTransaction

Author: KirillKurdyukov

CHANGELOG.md

@@ -1,3 +1,16 @@
+- Supported in ADO.NET GetSchema(Async). CollectionNames: 
+   * Tables
+   * TablesWithCollections
+   * DataSourceInformation
+   * MetaDataCollections
+   * Restrictions
+- Rename field _onStatus -> _onNotSuccessStatus in YdbDataReader
+- If session is not active, do not invoke DeleteNotActiveSession(session)
+- AttachStream: connect stream using NodeId
+- PoolManager: change pool properties on field
+- Delete *Settings.DefaultInstance because it's a singleton object that's changed by tasks when NodeId is set
+- DbConnection.Session.OnStatus(status) in YdbTransaction
+
 ## v0.9.4
 - Do not pessimize the node on Grpc.Core.StatusCode.Cancelled and Grpc.Core.StatusCode.DeadlineExceeded.
 - Dispose of WriterSession using dispose CancellationToken.

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

@@ -6,8 +6,7 @@ namespace Ydb.Sdk.Ado;
 internal static class PoolManager
 {
     private static readonly SemaphoreSlim SemaphoreSlim = new(1); // async mutex
-
-    private static ConcurrentDictionary<string, SessionPool> Pools { get; } = new();
+    private static readonly ConcurrentDictionary<string, SessionPool> Pools = new();
 
     internal static async Task<Session> GetSession(YdbConnectionStringBuilder connectionString)
     {

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

@@ -181,7 +181,7 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
 
         var execSettings = CommandTimeout > 0
             ? new ExecuteQuerySettings { TransportTimeout = TimeSpan.FromSeconds(CommandTimeout) }
-            : ExecuteQuerySettings.DefaultInstance;
+            : new ExecuteQuerySettings();
 
         var ydbDataReader = await YdbDataReader.CreateYdbDataReader(YdbConnection.Session.ExecuteQuery(
                 preparedSql.ToString(), ydbParameters, execSettings, Transaction?.TransactionControl),

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

@@ -164,6 +164,39 @@ protected override YdbCommand CreateDbCommand()
         return CreateDbCommand();
     }
 
+    public override DataTable GetSchema()
+    {
+        return GetSchemaAsync().GetAwaiter().GetResult();
+    }
+
+    public override DataTable GetSchema(string collectionName)
+    {
+        return GetSchemaAsync(collectionName).GetAwaiter().GetResult();
+    }
+
+    public override DataTable GetSchema(string collectionName, string?[] restrictionValues)
+    {
+        return GetSchemaAsync(collectionName, restrictionValues).GetAwaiter().GetResult();
+    }
+
+    public override Task<DataTable> GetSchemaAsync(CancellationToken cancellationToken = default)
+    {
+        return GetSchemaAsync("MetaDataCollections", cancellationToken);
+    }
+
+    public override Task<DataTable> GetSchemaAsync(string collectionName, CancellationToken cancellationToken = default)
+    {
+        return GetSchemaAsync(collectionName, new string[4], cancellationToken);
+    }
+
+    public override Task<DataTable> GetSchemaAsync(
+        string collectionName,
+        string?[] restrictionValues,
+        CancellationToken cancellationToken = default)
+    {
+        return YdbSchema.GetSchemaAsync(this, collectionName, restrictionValues, cancellationToken);
+    }
+
     internal void EnsureConnectionOpen()
     {
         if (ConnectionState == ConnectionState.Closed)

src/Ydb.Sdk/src/Ado/YdbDataReader.cs

@@ -11,7 +11,7 @@ public sealed class YdbDataReader : DbDataReader, IAsyncEnumerable<YdbDataRecord
     private readonly IAsyncEnumerator<ExecuteQueryResponsePart> _stream;
     private readonly YdbTransaction? _ydbTransaction;
     private readonly RepeatedField<IssueMessage> _issueMessagesInStream = new();
-    private readonly Action<Status> _onStatus;
+    private readonly Action<Status> _onNotSuccessStatus;
 
     private int _currentRowIndex = -1;
     private long _resultSetIndex = -1;
@@ -51,11 +51,11 @@ private interface IMetadata
 
     private YdbDataReader(
         IAsyncEnumerator<ExecuteQueryResponsePart> resultSetStream,
-        Action<Status> onStatus,
+        Action<Status> onNotSuccessStatus,
         YdbTransaction? ydbTransaction)
     {
         _stream = resultSetStream;
-        _onStatus = onStatus;
+        _onNotSuccessStatus = onNotSuccessStatus;
         _ydbTransaction = ydbTransaction;
     }
 
@@ -489,7 +489,7 @@ private async Task<State> NextExecPart()
 
                 var status = Status.FromProto(part.Status, _issueMessagesInStream);
 
-                _onStatus(status);
+                _onNotSuccessStatus(status);
 
                 throw new YdbException(status);
             }
@@ -515,7 +515,7 @@ private async Task<State> NextExecPart()
         {
             OnFailReadStream();
 
-            _onStatus(e.Status);
+            _onNotSuccessStatus(e.Status);
 
             throw new YdbException(e.Status);
         }