feat(csharp): wire telemetry for SEA/REST protocol connections

csharp/src/StatementExecution/StatementExecutionConnection.cs

@@ -16,11 +16,13 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
 using AdbcDrivers.Databricks.Http;
+using AdbcDrivers.Databricks.Telemetry;
 using AdbcDrivers.HiveServer2.Hive2;
 using AdbcDrivers.Databricks.StatementExecution.MetadataCommands;
 using AdbcDrivers.HiveServer2.Spark;
@@ -54,6 +56,10 @@ internal class StatementExecutionConnection : TracingConnection, IGetObjectsData
         private string? _sessionId;
         private readonly SemaphoreSlim _sessionLock = new SemaphoreSlim(1, 1);
 
+        // Telemetry
+        private IConnectionTelemetry _telemetry = NoOpConnectionTelemetry.Instance;
+        internal TelemetrySessionContext? TelemetrySession => _telemetry.Session;
+
         // Configuration for statement creation
         private readonly string _resultDisposition;
         private readonly string _resultFormat;
@@ -411,6 +417,9 @@ public async Task OpenAsync(CancellationToken cancellationToken = default)
                         {
                             _catalog = GetCurrentCatalog();
                         }
+
+                        // Initialize telemetry after successful session creation
+                        InitializeTelemetry();
                     }
                 }
                 finally
@@ -420,6 +429,20 @@ public async Task OpenAsync(CancellationToken cancellationToken = default)
             }
         }
 
+        /// <summary>
+        /// Initializes telemetry via the SeaConnectionTelemetry factory.
+        /// Returns NoOpConnectionTelemetry if telemetry is disabled or initialization fails.
+        /// </summary>
+        private void InitializeTelemetry()
+        {
+            _telemetry = SeaConnectionTelemetry.Create(
+                properties: _properties,
+                host: GetHost(_properties),
+                assemblyVersion: AssemblyVersion,
+                sessionId: _sessionId,
+                activity: Activity.Current);
+        }
+
         /// <summary>
         /// Creates a new statement for query execution.
         /// </summary>
@@ -440,7 +463,8 @@ public override AdbcStatement CreateStatement()
                 _recyclableMemoryStreamManager,
                 _lz4BufferPool,
                 _cloudFetchHttpClient,
-                this); // Pass connection as TracingConnection for tracing support
+                this, // Pass connection as TracingConnection for tracing support
+                _telemetry.Session); // Pass telemetry session context
         }
 
         public override void SetOption(string key, string? value)
@@ -902,6 +926,9 @@ public override void Dispose()
                 _cloudFetchHttpClient.Dispose();
 
                 _sessionLock.Dispose();
+
+                // Dispose telemetry (flush + release client)
+                _telemetry.DisposeAsync().GetAwaiter().GetResult();
             });
         }