fix(csharp): emit correct metadata telemetry for statement-level metadata commands

Statement-level metadata commands (getcatalogs, gettables, getcolumns, etc.)
executed via DatabricksStatement.ExecuteQuery were incorrectly tagged as
StatementType.Query/OperationType.ExecuteStatement. This fix correctly emits
StatementType.Metadata with the appropriate OperationType (ListCatalogs,
ListTables, ListColumns, etc.), aligning with the connection-level GetObjects
telemetry. The two paths remain distinguishable via sql_statement_id (populated
for statement path, empty for GetObjects path).

Co-authored-by: Isaac

Author: Jade Wang

csharp/src/DatabricksStatement.cs

@@ -118,6 +118,41 @@ public DatabricksStatement(DatabricksConnection connection)
             return ctx;
         }
 
+        /// <summary>
+        /// Maps a metadata SQL command to the corresponding telemetry operation type.
+        /// Returns null if the command is not a recognized metadata command.
+        /// </summary>
+        internal static OperationType? GetMetadataOperationType(string? sqlQuery)
+        {
+            return sqlQuery?.ToLowerInvariant() switch
+            {
+                "getcatalogs" => OperationType.ListCatalogs,
+                "getschemas" => OperationType.ListSchemas,
+                "gettables" => OperationType.ListTables,
+                "getcolumns" or "getcolumnsextended" => OperationType.ListColumns,
+                "gettabletypes" => OperationType.ListTableTypes,
+                "getprimarykeys" => OperationType.ListPrimaryKeys,
+                "getcrossreference" => OperationType.ListCrossReferences,
+                _ => null
+            };
+        }
+
+        private StatementTelemetryContext? CreateMetadataTelemetryContext()
+        {
+            var session = ((DatabricksConnection)Connection).TelemetrySession;
+            if (session?.TelemetryClient == null) return null;
+
+            var operationType = GetMetadataOperationType(SqlQuery) ?? OperationType.Unspecified;
+
+            var ctx = new StatementTelemetryContext(session);
+            ctx.OperationType = operationType;
+            ctx.StatementType = Telemetry.Proto.Statement.Types.Type.Metadata;
+            ctx.ResultFormat = ExecutionResultFormat.InlineArrow;
+            ctx.IsCompressed = false;
+            ctx.IsInternalCall = IsInternalCall;
+            return ctx;
+        }
+
         private void RecordSuccess(StatementTelemetryContext ctx)
         {
             ctx.RecordFirstBatchReady();
@@ -136,7 +171,9 @@ private void RecordError(StatementTelemetryContext ctx, Exception ex)
 
         public override QueryResult ExecuteQuery()
         {
-            var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
+            var ctx = IsMetadataCommand
+                ? CreateMetadataTelemetryContext()
+                : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
             if (ctx == null) return base.ExecuteQuery();
 
             try
@@ -159,7 +196,9 @@ public override QueryResult ExecuteQuery()
 
         public override async ValueTask<QueryResult> ExecuteQueryAsync()
         {
-            var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
+            var ctx = IsMetadataCommand
+                ? CreateMetadataTelemetryContext()
+                : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
             if (ctx == null) return await base.ExecuteQueryAsync();
 
             try

csharp/test/E2E/Telemetry/StatementMetadataTelemetryTests.cs

@@ -0,0 +1,247 @@
+/*
+* 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;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using AdbcDrivers.Databricks.Telemetry;
+using AdbcDrivers.HiveServer2;
+using Apache.Arrow.Adbc;
+using Apache.Arrow.Adbc.Tests;
+using Xunit;
+using Xunit.Abstractions;
+using OperationType = AdbcDrivers.Databricks.Telemetry.Proto.Operation.Types.Type;
+using StatementType = AdbcDrivers.Databricks.Telemetry.Proto.Statement.Types.Type;
+
+namespace AdbcDrivers.Databricks.Tests.E2E.Telemetry
+{
+    /// <summary>
+    /// E2E tests for statement-level metadata command telemetry.
+    /// Validates that metadata commands executed via DatabricksStatement.ExecuteQuery
+    /// (e.g., SqlQuery = "getcatalogs") emit telemetry with correct StatementType.Metadata
+    /// and the appropriate OperationType, rather than StatementType.Query/OperationType.ExecuteStatement.
+    /// </summary>
+    public class StatementMetadataTelemetryTests : TestBase<DatabricksTestConfiguration, DatabricksTestEnvironment>
+    {
+        // Filters to scope metadata queries and avoid MaxMessageSize errors
+        private const string TestCatalog = "main";
+        private const string TestSchema = "adbc_testing";
+        private const string TestTable = "all_column_types";
+
+        public StatementMetadataTelemetryTests(ITestOutputHelper? outputHelper)
+            : base(outputHelper, new DatabricksTestEnvironment.Factory())
+        {
+            Skip.IfNot(Utils.CanExecuteTestConfig(TestConfigVariable));
+        }
+
+        [SkippableFact]
+        public async Task Telemetry_StatementGetCatalogs_EmitsMetadataWithListCatalogs()
+        {
+            await AssertStatementMetadataTelemetry(
+                command: "getcatalogs",
+                expectedOperationType: OperationType.ListCatalogs);
+        }
+
+        [SkippableFact]
+        public async Task Telemetry_StatementGetSchemas_EmitsMetadataWithListSchemas()
+        {
+            await AssertStatementMetadataTelemetry(
+                command: "getschemas",
+                expectedOperationType: OperationType.ListSchemas,
+                options: new Dictionary<string, string>
+                {
+                    [ApacheParameters.CatalogName] = TestCatalog,
+                });
+        }
+
+        [SkippableFact]
+        public async Task Telemetry_StatementGetTables_EmitsMetadataWithListTables()
+        {
+            await AssertStatementMetadataTelemetry(
+                command: "gettables",
+                expectedOperationType: OperationType.ListTables,
+                options: new Dictionary<string, string>
+                {
+                    [ApacheParameters.CatalogName] = TestCatalog,
+                    [ApacheParameters.SchemaName] = TestSchema,
+                });
+        }
+
+        [SkippableFact]
+        public async Task Telemetry_StatementGetColumns_EmitsMetadataWithListColumns()
+        {
+            await AssertStatementMetadataTelemetry(
+                command: "getcolumns",
+                expectedOperationType: OperationType.ListColumns,
+                options: new Dictionary<string, string>
+                {
+                    [ApacheParameters.CatalogName] = TestCatalog,
+                    [ApacheParameters.SchemaName] = TestSchema,
+                    [ApacheParameters.TableName] = TestTable,
+                });
+        }
+
+        [SkippableFact]
+        public async Task Telemetry_StatementMetadata_AllCommands_EmitCorrectOperationType()
+        {
+            CapturingTelemetryExporter exporter = null!;
+            AdbcConnection? connection = null;
+
+            try
+            {
+                var properties = TestEnvironment.GetDriverParameters(TestConfiguration);
+                (connection, exporter) = TelemetryTestHelpers.CreateConnectionWithCapturingTelemetry(properties);
+
+                var commandMappings = new (string Command, OperationType ExpectedOp, Dictionary<string, string>? Options)[]
+                {
+                    ("getcatalogs", OperationType.ListCatalogs, null),
+                    ("getschemas", OperationType.ListSchemas, new Dictionary<string, string>
+                    {
+                        [ApacheParameters.CatalogName] = TestCatalog,
+                    }),
+                    ("gettables", OperationType.ListTables, new Dictionary<string, string>
+                    {
+                        [ApacheParameters.CatalogName] = TestCatalog,
+                        [ApacheParameters.SchemaName] = TestSchema,
+                    }),
+                    ("getcolumns", OperationType.ListColumns, new Dictionary<string, string>
+                    {
+                        [ApacheParameters.CatalogName] = TestCatalog,
+                        [ApacheParameters.SchemaName] = TestSchema,
+                        [ApacheParameters.TableName] = TestTable,
+                    }),
+                };
+
+                foreach (var mapping in commandMappings)
+                {
+                    exporter.Reset();
+
+                    // Explicit using block so statement is disposed (and telemetry emitted) before we check
+                    using (var statement = connection.CreateStatement())
+                    {
+                        statement.SetOption(ApacheParameters.IsMetadataCommand, "true");
+                        statement.SqlQuery = mapping.Command;
+
+                        if (mapping.Options != null)
+                        {
+                            foreach (var opt in mapping.Options)
+                            {
+                                statement.SetOption(opt.Key, opt.Value);
+                            }
+                        }
+
+                        var result = statement.ExecuteQuery();
+                        result.Stream?.Dispose();
+                    }
+
+                    // Flush telemetry after statement disposal
+                    if (connection is DatabricksConnection dbConn && dbConn.TelemetrySession?.TelemetryClient != null)
+                    {
+                        await dbConn.TelemetrySession.TelemetryClient.FlushAsync(default);
+                    }
+
+                    var logs = await TelemetryTestHelpers.WaitForTelemetryEvents(exporter, expectedCount: 1, timeoutMs: 5000);
+                    Assert.NotEmpty(logs);
+
+                    var log = TelemetryTestHelpers.FindLog(logs, proto =>
+                        proto.SqlOperation?.OperationDetail?.OperationType == mapping.ExpectedOp);
+
+                    Assert.NotNull(log);
+
+                    var protoLog = TelemetryTestHelpers.GetProtoLog(log);
+                    Assert.Equal(StatementType.Metadata, protoLog.SqlOperation.StatementType);
+                    Assert.Equal(mapping.ExpectedOp, protoLog.SqlOperation.OperationDetail.OperationType);
+                }
+            }
+            finally
+            {
+                connection?.Dispose();
+                TelemetryTestHelpers.ClearExporterOverride();
+            }
+        }
+
+        /// <summary>
+        /// Helper method to test a single statement-level metadata command emits the correct telemetry.
+        /// </summary>
+        private async Task AssertStatementMetadataTelemetry(
+            string command,
+            OperationType expectedOperationType,
+            Dictionary<string, string>? options = null)
+        {
+            CapturingTelemetryExporter exporter = null!;
+            AdbcConnection? connection = null;
+
+            try
+            {
+                var properties = TestEnvironment.GetDriverParameters(TestConfiguration);
+                (connection, exporter) = TelemetryTestHelpers.CreateConnectionWithCapturingTelemetry(properties);
+
+                // Execute metadata command via statement path
+                // Explicit using block so statement is disposed (and telemetry emitted) before we check
+                using (var statement = connection.CreateStatement())
+                {
+                    statement.SetOption(ApacheParameters.IsMetadataCommand, "true");
+                    statement.SqlQuery = command;
+
+                    if (options != null)
+                    {
+                        foreach (var opt in options)
+                        {
+                            statement.SetOption(opt.Key, opt.Value);
+                        }
+                    }
+
+                    var result = statement.ExecuteQuery();
+                    result.Stream?.Dispose();
+                }
+
+                // Flush telemetry after statement disposal
+                if (connection is DatabricksConnection dbConn && dbConn.TelemetrySession?.TelemetryClient != null)
+                {
+                    await dbConn.TelemetrySession.TelemetryClient.FlushAsync(default);
+                }
+
+                // Wait for telemetry events
+                var logs = await TelemetryTestHelpers.WaitForTelemetryEvents(exporter, expectedCount: 1, timeoutMs: 5000);
+
+                Assert.NotEmpty(logs);
+
+                // Find the metadata telemetry log with correct operation type
+                var log = TelemetryTestHelpers.FindLog(logs, proto =>
+                    proto.SqlOperation?.OperationDetail?.OperationType == expectedOperationType);
+
+                Assert.NotNull(log);
+
+                var protoLog = TelemetryTestHelpers.GetProtoLog(log);
+
+                // Verify statement type is METADATA (not QUERY)
+                Assert.Equal(StatementType.Metadata, protoLog.SqlOperation.StatementType);
+
+                // Verify operation type matches the metadata command
+                Assert.Equal(expectedOperationType, protoLog.SqlOperation.OperationDetail.OperationType);
+
+                // Verify basic session-level telemetry fields are populated
+                TelemetryTestHelpers.AssertSessionFieldsPopulated(protoLog);
+            }
+            finally
+            {
+                connection?.Dispose();
+                TelemetryTestHelpers.ClearExporterOverride();
+            }
+        }
+    }
+}

csharp/test/Unit/DatabricksStatementUnitTests.cs

@@ -20,6 +20,7 @@
 using AdbcDrivers.HiveServer2.Spark;
 using AdbcDrivers.Databricks;
 using Xunit;
+using OperationType = AdbcDrivers.Databricks.Telemetry.Proto.Operation.Types.Type;
 
 namespace AdbcDrivers.Databricks.Tests.Unit
 {
@@ -126,5 +127,38 @@ public void CreateStatement_ConfOverlayInitiallyNull()
             var confOverlay = GetConfOverlay(statement);
             Assert.Null(confOverlay);
         }
+
+        [Theory]
+        [InlineData("getcatalogs", OperationType.ListCatalogs)]
+        [InlineData("getschemas", OperationType.ListSchemas)]
+        [InlineData("gettables", OperationType.ListTables)]
+        [InlineData("getcolumns", OperationType.ListColumns)]
+        [InlineData("getcolumnsextended", OperationType.ListColumns)]
+        [InlineData("gettabletypes", OperationType.ListTableTypes)]
+        [InlineData("getprimarykeys", OperationType.ListPrimaryKeys)]
+        [InlineData("getcrossreference", OperationType.ListCrossReferences)]
+        public void GetMetadataOperationType_ReturnsCorrectType(string command, OperationType expected)
+        {
+            Assert.Equal(expected, DatabricksStatement.GetMetadataOperationType(command));
+        }
+
+        [Theory]
+        [InlineData(null)]
+        [InlineData("")]
+        [InlineData("SELECT 1")]
+        [InlineData("unknown_command")]
+        public void GetMetadataOperationType_ReturnsNull_ForNonMetadataCommands(string? command)
+        {
+            Assert.Null(DatabricksStatement.GetMetadataOperationType(command));
+        }
+
+        [Theory]
+        [InlineData("GETCATALOGS")]
+        [InlineData("GetCatalogs")]
+        [InlineData("GetTables")]
+        public void GetMetadataOperationType_IsCaseInsensitive(string command)
+        {
+            Assert.NotNull(DatabricksStatement.GetMetadataOperationType(command));
+        }
     }
 }