fix(csharp): surface closed/expired sessions as a clear, typed error

csharp/src/DatabricksConnection.cs

@@ -132,6 +132,23 @@ internal class DatabricksConnection : SparkHttpConnection
         private bool _sessionDeleteTelemetryEmitted;
         internal TelemetrySessionContext? TelemetrySession => _telemetry.Session;
 
+        // Set once the server-side session is known to be closed/expired (e.g. inactivity
+        // timeout). The session handle held by this connection is then permanently stale, so
+        // every subsequent operation should fail fast rather than reusing the dead handle.
+        private volatile bool _sessionInvalid;
+
+        /// <summary>
+        /// True when the server-side session has been closed or has expired. Once set, the
+        /// connection can no longer execute statements and must be disposed and re-created.
+        /// </summary>
+        internal bool IsSessionInvalid => _sessionInvalid;
+
+        /// <summary>
+        /// Marks the connection's server-side session as closed/expired. Idempotent and
+        /// thread-safe. Subsequent operations fail fast with <see cref="DatabricksSessionExpiredException"/>.
+        /// </summary>
+        internal void MarkSessionInvalid() => _sessionInvalid = true;
+
         /// <summary>
         /// RecyclableMemoryStreamManager for LZ4 decompression.
         /// If provided by Database, this is shared across all connections for optimal pooling.

csharp/src/DatabricksSessionExpiredException.cs

@@ -0,0 +1,93 @@
+/*
+* 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 Apache.Arrow.Adbc;
+
+namespace AdbcDrivers.Databricks
+{
+    /// <summary>
+    /// Raised when the server-side session backing a connection has been closed or has
+    /// expired (e.g. due to an inactivity timeout) and the connection can no longer be used.
+    ///
+    /// The server reports this as HTTP 400 with a Thrift error message containing
+    /// "Invalid SessionHandle". Without this typed exception the condition surfaces as a
+    /// generic, misleading "An unexpected error occurred while fetching results / Couldn't
+    /// connect to server" error, which makes it impossible for callers to distinguish a
+    /// recoverable "reconnect" situation from a genuine network/transport failure.
+    ///
+    /// Callers that catch this exception should dispose the connection and open a new one.
+    /// </summary>
+    public class DatabricksSessionExpiredException : DatabricksException
+    {
+        /// <summary>
+        /// Substring present in the server's error message for a closed/expired session.
+        /// Both the inactivity-timeout variant ("Invalid SessionHandle: Session [..] is closed")
+        /// and the explicitly-closed variant ("Invalid SessionHandle: SessionHandle [..]")
+        /// contain this phrase.
+        /// </summary>
+        internal const string ServerErrorSignature = "Invalid SessionHandle";
+
+        /// <summary>
+        /// Message used when failing fast on a connection already known to have an invalid session.
+        /// </summary>
+        internal const string FastFailMessage =
+            "The Databricks session has expired or was closed by the server and is no longer usable. " +
+            "Open a new connection to continue.";
+
+        public DatabricksSessionExpiredException(string message)
+            : base(message, AdbcStatusCode.InvalidState)
+        {
+        }
+
+        public DatabricksSessionExpiredException(string message, Exception innerException)
+            : base(message, AdbcStatusCode.InvalidState, innerException)
+        {
+        }
+
+        /// <summary>
+        /// Determines whether the given exception (or any exception in its inner / aggregate
+        /// chain) represents a closed or expired server-side session.
+        /// </summary>
+        internal static bool IsSessionExpired(Exception? exception)
+        {
+            switch (exception)
+            {
+                case null:
+                    return false;
+                case DatabricksSessionExpiredException:
+                    return true;
+                case AggregateException aggregate:
+                    foreach (Exception inner in aggregate.InnerExceptions)
+                    {
+                        if (IsSessionExpired(inner))
+                        {
+                            return true;
+                        }
+                    }
+                    return false;
+            }
+
+            if (exception.Message != null &&
+                exception.Message.IndexOf(ServerErrorSignature, StringComparison.OrdinalIgnoreCase) >= 0)
+            {
+                return true;
+            }
+
+            return IsSessionExpired(exception.InnerException);
+        }
+    }
+}

csharp/src/DatabricksStatement.cs

@@ -211,60 +211,118 @@ private void RecordError(StatementTelemetryContext ctx, Exception ex)
             CaptureRetryCount(ctx);
         }
 
-        public override QueryResult ExecuteQuery()
+        /// <summary>
+        /// Tier 2 fast-fail: if the connection's server-side session is already known to be
+        /// closed/expired, throw immediately instead of issuing an RPC that would reuse the
+        /// stale handle and return the same opaque error.
+        /// </summary>
+        private void ThrowIfSessionInvalid()
+        {
+            if (((DatabricksConnection)Connection).IsSessionInvalid)
+            {
+                throw new DatabricksSessionExpiredException(DatabricksSessionExpiredException.FastFailMessage);
+            }
+        }
+
+        /// <summary>
+        /// Tier 1: when an execution failure indicates a closed/expired server-side session,
+        /// mark the connection invalid and rethrow as a clear <see cref="DatabricksSessionExpiredException"/>.
+        /// Otherwise returns so the caller can rethrow the original exception (preserving its stack).
+        /// </summary>
+        private void ThrowIfSessionExpired(Exception ex)
         {
-            var ctx = IsMetadataCommand
-                ? CreateMetadataTelemetryContext()
-                : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
-            if (ctx == null) return MaybeWrapComplexTypes(base.ExecuteQuery());
+            if (!DatabricksSessionExpiredException.IsSessionExpired(ex))
+            {
+                return;
+            }
+
+            ((DatabricksConnection)Connection).MarkSessionInvalid();
 
-            // Expose ctx to NewReader so the operation status poller can update PollCount/PollLatencyMs (PECO-2992).
-            PendingTelemetryContext = ctx;
+            // Already the clean typed exception (e.g. surfaced by a nested statement) — preserve it.
+            if (ex is DatabricksSessionExpiredException sessionEx)
+            {
+                throw sessionEx;
+            }
+
+            throw new DatabricksSessionExpiredException(
+                "The Databricks session has expired or was closed by the server " +
+                "(e.g. due to an inactivity timeout). The connection can no longer be used; " +
+                "open a new connection to continue. Server error: " + ex.Message,
+                ex);
+        }
+
+        public override QueryResult ExecuteQuery()
+        {
+            ThrowIfSessionInvalid();
             try
             {
-                QueryResult result = base.ExecuteQuery();
-                // Store the UNWRAPPED result for telemetry: EmitTelemetry inspects
-                // _lastQueryResult.Stream via `is CloudFetchReader/DatabricksCompositeReader`
-                // to read chunk metrics and IsCompressed/ResultFormat. ComplexTypeSerializingStream
-                // would mask those types, so keep the real reader here and wrap only on return.
-                _lastQueryResult = result;
-                RecordSuccess(ctx);
-                return MaybeWrapComplexTypes(result);
+                var ctx = IsMetadataCommand
+                    ? CreateMetadataTelemetryContext()
+                    : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
+                if (ctx == null) return MaybeWrapComplexTypes(base.ExecuteQuery());
+
+                // Expose ctx to NewReader so the operation status poller can update PollCount/PollLatencyMs (PECO-2992).
+                PendingTelemetryContext = ctx;
+                try
+                {
+                    QueryResult result = base.ExecuteQuery();
+                    // Store the UNWRAPPED result for telemetry: EmitTelemetry inspects
+                    // _lastQueryResult.Stream via `is CloudFetchReader/DatabricksCompositeReader`
+                    // to read chunk metrics and IsCompressed/ResultFormat. ComplexTypeSerializingStream
+                    // would mask those types, so keep the real reader here and wrap only on return.
+                    _lastQueryResult = result;
+                    RecordSuccess(ctx);
+                    return MaybeWrapComplexTypes(result);
+                }
+                catch (Exception ex)
+                {
+                    RecordError(ctx, ex);
+                    // Emit telemetry immediately on error (won't reach Dispose)
+                    EmitTelemetry(ctx);
+                    PendingTelemetryContext = null; // Clear to avoid double emission
+                    throw;
+                }
             }
             catch (Exception ex)
             {
-                RecordError(ctx, ex);
-                // Emit telemetry immediately on error (won't reach Dispose)
-                EmitTelemetry(ctx);
-                PendingTelemetryContext = null; // Clear to avoid double emission
+                ThrowIfSessionExpired(ex); // Tier 1: reclassify a closed/expired session; else rethrow as-is.
                 throw;
             }
         }
 
         public override async ValueTask<QueryResult> ExecuteQueryAsync()
         {
-            var ctx = IsMetadataCommand
-                ? CreateMetadataTelemetryContext()
-                : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
-            if (ctx == null) return MaybeWrapComplexTypes(await base.ExecuteQueryAsync());
-
-            // Expose ctx to NewReader so the operation status poller can update PollCount/PollLatencyMs (PECO-2992).
-            PendingTelemetryContext = ctx;
+            ThrowIfSessionInvalid();
             try
             {
-                QueryResult result = await base.ExecuteQueryAsync();
-                // Store the UNWRAPPED result for telemetry (see ExecuteQuery for rationale):
-                // the wrapper would mask CloudFetchReader/DatabricksCompositeReader from EmitTelemetry.
-                _lastQueryResult = result;
-                RecordSuccess(ctx);
-                return MaybeWrapComplexTypes(result);
+                var ctx = IsMetadataCommand
+                    ? CreateMetadataTelemetryContext()
+                    : CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Query);
+                if (ctx == null) return MaybeWrapComplexTypes(await base.ExecuteQueryAsync());
+
+                // Expose ctx to NewReader so the operation status poller can update PollCount/PollLatencyMs (PECO-2992).
+                PendingTelemetryContext = ctx;
+                try
+                {
+                    QueryResult result = await base.ExecuteQueryAsync();
+                    // Store the UNWRAPPED result for telemetry (see ExecuteQuery for rationale):
+                    // the wrapper would mask CloudFetchReader/DatabricksCompositeReader from EmitTelemetry.
+                    _lastQueryResult = result;
+                    RecordSuccess(ctx);
+                    return MaybeWrapComplexTypes(result);
+                }
+                catch (Exception ex)
+                {
+                    RecordError(ctx, ex);
+                    // Emit telemetry immediately on error (won't reach Dispose)
+                    EmitTelemetry(ctx);
+                    PendingTelemetryContext = null; // Clear to avoid double emission
+                    throw;
+                }
             }
             catch (Exception ex)
             {
-                RecordError(ctx, ex);
-                // Emit telemetry immediately on error (won't reach Dispose)
-                EmitTelemetry(ctx);
-                PendingTelemetryContext = null; // Clear to avoid double emission
+                ThrowIfSessionExpired(ex); // Tier 1: reclassify a closed/expired session; else rethrow as-is.
                 throw;
             }
         }
@@ -286,44 +344,62 @@ private QueryResult MaybeWrapComplexTypes(QueryResult result)
 
         public override UpdateResult ExecuteUpdate()
         {
-            var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Update);
-            if (ctx == null) return base.ExecuteUpdate();
-
-            PendingTelemetryContext = ctx;
+            ThrowIfSessionInvalid();
             try
             {
-                UpdateResult result = base.ExecuteUpdate();
-                RecordSuccess(ctx);
-                return result;
+                var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Update);
+                if (ctx == null) return base.ExecuteUpdate();
+
+                PendingTelemetryContext = ctx;
+                try
+                {
+                    UpdateResult result = base.ExecuteUpdate();
+                    RecordSuccess(ctx);
+                    return result;
+                }
+                catch (Exception ex)
+                {
+                    RecordError(ctx, ex);
+                    // Emit telemetry immediately on error (won't reach Dispose)
+                    EmitTelemetry(ctx);
+                    PendingTelemetryContext = null; // Clear to avoid double emission
+                    throw;
+                }
             }
             catch (Exception ex)
             {
-                RecordError(ctx, ex);
-                // Emit telemetry immediately on error (won't reach Dispose)
-                EmitTelemetry(ctx);
-                PendingTelemetryContext = null; // Clear to avoid double emission
+                ThrowIfSessionExpired(ex); // Tier 1: reclassify a closed/expired session; else rethrow as-is.
                 throw;
             }
         }
 
         public override async Task<UpdateResult> ExecuteUpdateAsync()
         {
-            var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Update);
-            if (ctx == null) return await base.ExecuteUpdateAsync();
-
-            PendingTelemetryContext = ctx;
+            ThrowIfSessionInvalid();
             try
             {
-                UpdateResult result = await base.ExecuteUpdateAsync();
-                RecordSuccess(ctx);
-                return result;
+                var ctx = CreateTelemetryContext(Telemetry.Proto.Statement.Types.Type.Update);
+                if (ctx == null) return await base.ExecuteUpdateAsync();
+
+                PendingTelemetryContext = ctx;
+                try
+                {
+                    UpdateResult result = await base.ExecuteUpdateAsync();
+                    RecordSuccess(ctx);
+                    return result;
+                }
+                catch (Exception ex)
+                {
+                    RecordError(ctx, ex);
+                    // Emit telemetry immediately on error (won't reach Dispose)
+                    EmitTelemetry(ctx);
+                    PendingTelemetryContext = null; // Clear to avoid double emission
+                    throw;
+                }
             }
             catch (Exception ex)
             {
-                RecordError(ctx, ex);
-                // Emit telemetry immediately on error (won't reach Dispose)
-                EmitTelemetry(ctx);
-                PendingTelemetryContext = null; // Clear to avoid double emission
+                ThrowIfSessionExpired(ex); // Tier 1: reclassify a closed/expired session; else rethrow as-is.
                 throw;
             }
         }