feat(csharp): make StatementExecutionConnection inherit from AdbcConnection

Make StatementExecutionConnection a proper ADBC connection by inheriting
from AdbcConnection base class. This establishes the correct architecture
for the REST API implementation.

Changes:
- StatementExecutionConnection now inherits from AdbcConnection
- Added StatementExecutionStatement stub class implementing AdbcStatement
- CreateStatement() now returns a StatementExecutionStatement instance
- Implemented required abstract methods (GetObjects, GetTableSchema, GetTableTypes)
  with NotImplemented exceptions (metadata operations to be added in future)
- Statement execution methods throw NotImplementedException with note about
  PECO-2791-B implementation

Testing:
- Added 6 new unit tests verifying AdbcConnection implementation
- Updated E2E test to verify CreateStatement returns a statement
- All 42 tests pass (26 unit + 16 E2E)

This is a minimal implementation (Option 1) that establishes the proper
inheritance hierarchy while deferring full query execution to PECO-2791-B.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Jade Wang

csharp/src/StatementExecution/StatementExecutionConnection.cs

@@ -21,14 +21,15 @@
 using System.Threading.Tasks;
 using Apache.Arrow.Adbc.Drivers.Apache;
 using Apache.Arrow.Adbc.Drivers.Apache.Spark;
+using Apache.Arrow.Ipc;
 
 namespace Apache.Arrow.Adbc.Drivers.Databricks.StatementExecution
 {
     /// <summary>
     /// Connection implementation for Databricks Statement Execution REST API.
     /// Manages SQL sessions and statement creation using the REST protocol.
     /// </summary>
-    internal class StatementExecutionConnection : IDisposable
+    internal class StatementExecutionConnection : AdbcConnection
     {
         private readonly IStatementExecutionClient _client;
         private readonly IReadOnlyDictionary<string, string> _properties;
@@ -105,12 +106,12 @@ public async Task OpenAsync(CancellationToken cancellationToken = default)
         /// </summary>
         /// <returns>A new statement instance.</returns>
         /// <remarks>
-        /// This will be implemented in PECO-2791-B to return StatementExecutionStatement.
-        /// For now, it throws NotImplementedException.
+        /// Returns a StatementExecutionStatement instance. Full query execution
+        /// implementation will be completed in PECO-2791-B.
         /// </remarks>
-        public AdbcStatement CreateStatement()
+        public override AdbcStatement CreateStatement()
         {
-            throw new NotImplementedException("Statement creation will be implemented in PECO-2791-B");
+            return new StatementExecutionStatement(_client, _warehouseId, _sessionId);
         }
 
         /// <summary>
@@ -138,6 +139,51 @@ public async Task CloseAsync(CancellationToken cancellationToken = default)
             }
         }
 
+        /// <summary>
+        /// Get a hierarchical view of all catalogs, database schemas, tables, and columns.
+        /// </summary>
+        /// <remarks>
+        /// Metadata operations via Statement Execution API will be implemented in a future release.
+        /// </remarks>
+        public override IArrowArrayStream GetObjects(
+            GetObjectsDepth depth,
+            string? catalogPattern,
+            string? dbSchemaPattern,
+            string? tableNamePattern,
+            IReadOnlyList<string>? tableTypes,
+            string? columnNamePattern)
+        {
+            throw AdbcException.NotImplemented(
+                "GetObjects not yet implemented for Statement Execution API. " +
+                "Metadata operations will be added in a future release.");
+        }
+
+        /// <summary>
+        /// Get the Arrow schema of a database table.
+        /// </summary>
+        /// <remarks>
+        /// Metadata operations via Statement Execution API will be implemented in a future release.
+        /// </remarks>
+        public override Schema GetTableSchema(string? catalog, string? dbSchema, string tableName)
+        {
+            throw AdbcException.NotImplemented(
+                "GetTableSchema not yet implemented for Statement Execution API. " +
+                "Metadata operations will be added in a future release.");
+        }
+
+        /// <summary>
+        /// Get a list of table types supported by the database.
+        /// </summary>
+        /// <remarks>
+        /// Metadata operations via Statement Execution API will be implemented in a future release.
+        /// </remarks>
+        public override IArrowArrayStream GetTableTypes()
+        {
+            throw AdbcException.NotImplemented(
+                "GetTableTypes not yet implemented for Statement Execution API. " +
+                "Metadata operations will be added in a future release.");
+        }
+
         /// <summary>
         /// Extracts the warehouse ID from the http_path property (SparkParameters.Path).
         /// </summary>
@@ -183,7 +229,7 @@ private static string ExtractWarehouseId(IReadOnlyDictionary<string, string> pro
         /// <summary>
         /// Disposes the connection and releases resources.
         /// </summary>
-        public void Dispose()
+        public override void Dispose()
         {
             if (!_disposed)
             {
@@ -201,6 +247,7 @@ public void Dispose()
                     }
                 }
 
+                base.Dispose();
                 _disposed = true;
             }
         }

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -0,0 +1,78 @@
+/*
+* Copyright (c) 2025 ADBC Drivers Contributors
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+using System;
+
+namespace Apache.Arrow.Adbc.Drivers.Databricks.StatementExecution
+{
+    /// <summary>
+    /// Statement implementation for Databricks Statement Execution REST API.
+    /// This is a stub implementation - full query execution will be implemented in PECO-2791-B.
+    /// </summary>
+    internal class StatementExecutionStatement : AdbcStatement
+    {
+        private readonly IStatementExecutionClient _client;
+        private readonly string _warehouseId;
+        private readonly string? _sessionId;
+
+        /// <summary>
+        /// Initializes a new instance of the StatementExecutionStatement class.
+        /// </summary>
+        /// <param name="client">The Statement Execution API client.</param>
+        /// <param name="warehouseId">The warehouse ID for query execution.</param>
+        /// <param name="sessionId">Optional session ID for session-scoped execution.</param>
+        public StatementExecutionStatement(
+            IStatementExecutionClient client,
+            string warehouseId,
+            string? sessionId = null)
+        {
+            _client = client ?? throw new ArgumentNullException(nameof(client));
+            _warehouseId = warehouseId ?? throw new ArgumentNullException(nameof(warehouseId));
+            _sessionId = sessionId;
+        }
+
+        /// <summary>
+        /// Executes a query and returns the results.
+        /// </summary>
+        /// <returns>Query results with schema and data.</returns>
+        /// <remarks>
+        /// Full implementation will be added in PECO-2791-B.
+        /// This stub allows StatementExecutionConnection to properly inherit from AdbcConnection.
+        /// </remarks>
+        public override QueryResult ExecuteQuery()
+        {
+            throw new NotImplementedException(
+                "Query execution will be implemented in PECO-2791-B. " +
+                "This includes: executing statements via REST API, polling for completion, " +
+                "and fetching results via CloudFetch or inline disposition.");
+        }
+
+        /// <summary>
+        /// Executes an update statement (INSERT, UPDATE, DELETE, etc.) and returns affected row count.
+        /// </summary>
+        /// <returns>Update results with affected row count.</returns>
+        /// <remarks>
+        /// Full implementation will be added in PECO-2791-B.
+        /// This stub allows StatementExecutionConnection to properly inherit from AdbcConnection.
+        /// </remarks>
+        public override UpdateResult ExecuteUpdate()
+        {
+            throw new NotImplementedException(
+                "Update execution will be implemented in PECO-2791-B. " +
+                "This includes: executing DML statements via REST API and returning affected row counts.");
+        }
+    }
+}

csharp/test/E2E/StatementExecution/StatementExecutionConnectionE2ETests.cs

@@ -522,16 +522,16 @@ public async Task MultipleConnections_Independent_EachHasOwnSession()
         }
 
         [Fact]
-        public void CreateStatement_ThrowsNotImplementedException()
+        public void CreateStatement_ReturnsStatement()
         {
             _client = CreateClient();
             var properties = GetConnectionProperties(enableSessionManagement: true);
             var connection = new StatementExecutionConnection(_client, properties);
 
-            var exception = Assert.Throws<NotImplementedException>(() =>
-                connection.CreateStatement());
+            var statement = connection.CreateStatement();
 
-            Assert.Contains("PECO-2791-B", exception.Message);
+            Assert.NotNull(statement);
+            Assert.IsAssignableFrom<AdbcStatement>(statement);
         }
     }
 }

csharp/test/Unit/StatementExecution/StatementExecutionConnectionTests.cs

@@ -382,10 +382,10 @@ public async Task Dispose_WithActiveSession_ClosesSession()
 
         #endregion
 
-        #region CreateStatement Tests
+        #region AdbcConnection Implementation Tests
 
         [Fact]
-        public void CreateStatement_ThrowsNotImplementedException()
+        public void Connection_InheritsFromAdbcConnection()
         {
             var properties = new Dictionary<string, string>
             {
@@ -394,7 +394,107 @@ public void CreateStatement_ThrowsNotImplementedException()
 
             var connection = new StatementExecutionConnection(_mockClient.Object, properties);
 
-            var exception = Assert.Throws<NotImplementedException>(() => connection.CreateStatement());
+            Assert.IsAssignableFrom<AdbcConnection>(connection);
+        }
+
+        [Fact]
+        public void CreateStatement_ReturnsStatementExecutionStatement()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+            var statement = connection.CreateStatement();
+
+            Assert.NotNull(statement);
+            Assert.IsType<StatementExecutionStatement>(statement);
+        }
+
+        [Fact]
+        public void GetObjects_ThrowsNotImplemented()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+
+            var exception = Assert.Throws<AdbcException>(() =>
+                connection.GetObjects(
+                    AdbcConnection.GetObjectsDepth.All,
+                    null, null, null, null, null));
+
+            Assert.Contains("not yet implemented", exception.Message);
+        }
+
+        [Fact]
+        public void GetTableSchema_ThrowsNotImplemented()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+
+            var exception = Assert.Throws<AdbcException>(() =>
+                connection.GetTableSchema(null, null, "test_table"));
+
+            Assert.Contains("not yet implemented", exception.Message);
+        }
+
+        [Fact]
+        public void GetTableTypes_ThrowsNotImplemented()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+
+            var exception = Assert.Throws<AdbcException>(() =>
+                connection.GetTableTypes());
+
+            Assert.Contains("not yet implemented", exception.Message);
+        }
+
+        [Fact]
+        public void Statement_ExecuteQuery_ThrowsNotImplementedException()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+            var statement = connection.CreateStatement();
+            statement.SqlQuery = "SELECT 1";
+
+            var exception = Assert.Throws<NotImplementedException>(() =>
+                statement.ExecuteQuery());
+
+            Assert.Contains("PECO-2791-B", exception.Message);
+        }
+
+        [Fact]
+        public void Statement_ExecuteUpdate_ThrowsNotImplementedException()
+        {
+            var properties = new Dictionary<string, string>
+            {
+                { SparkParameters.Path, "/sql/1.0/warehouses/test-warehouse" }
+            };
+
+            var connection = new StatementExecutionConnection(_mockClient.Object, properties);
+            var statement = connection.CreateStatement();
+            statement.SqlQuery = "INSERT INTO test VALUES (1)";
+
+            var exception = Assert.Throws<NotImplementedException>(() =>
+                statement.ExecuteUpdate());
+
             Assert.Contains("PECO-2791-B", exception.Message);
         }