fix(rfc): own scan state per execution, lease connection slots, pin credentials

API_REFERENCE.md

@@ -1018,6 +1018,24 @@ SELECT * FROM sap_odp_get_subscriptions('ABAP_CDS', 'MY_CDS_VIEW$E');
 
 ---
 
+#### `sap_rfc_live_connections()` / `sap_rfc_connections_opened()` / `sap_rfc_connections_closed()`
+
+Scalar functions returning how many SAP RFC connections erpl has opened and closed in
+this process, and how many it currently holds open (`opened - closed`).
+
+Every open connection is a session and a work-process reservation on the SAP system, so
+`sap_rfc_live_connections()` is the number that matters to a Basis team. **Between
+queries it should be 0.** A non-zero value means erpl is still holding SAP sessions.
+
+```sql
+SELECT sap_rfc_live_connections();   -- 0 between queries
+```
+
+These are process-wide counters, not per-connection state, and they are `VOLATILE` so
+DuckDB never constant-folds them at bind time. Their intended use is asserting in tests
+and in the field that a scan released what it acquired — client-side timing shows nothing
+when a connection is never released, because the entire cost falls on the SAP system.
+
 #### `PRAGMA sap_odp_close_delta_cursor(odp_context, subscriber_process, odp_name [, secret=...])`
 
 Graceful counterpart to `sap_odp_drop`. Looks up the cursor for the given

CHANGELOG.md

@@ -61,6 +61,47 @@ LOAD erpl;
   partition worker. It now retries only when the selected function actually changed, and
   otherwise reports the original error.
 
+- **[rfc]** **A re-scanned `sap_read_table` silently returned nothing.** The column state
+  machines lived in *bind* data, which DuckDB reuses across executions of one bound plan,
+  so the second scan resumed from an exhausted cursor:
+
+  ```sql
+  PREPARE q AS SELECT count(*) FROM sap_read_table('SFLIGHT');
+  EXECUTE q;   -- 94
+  EXECUTE q;   --  0   <- no error
+  ```
+
+  Any re-scanned plan hit it: a prepared statement, a nested-loop join, an
+  un-materialised CTE referenced twice. Serial scans now build their machines in the
+  *global* state, which DuckDB rebuilds per execution — the pattern the partitioned path
+  already used, which is why `PARTITIONS` was never affected.
+
+  The same defect was present in `sap_show_tables`, `sap_odp_show_subscriptions` and the
+  internal table lister behind `ATTACH`; all three are fixed.
+
+- **[rfc]** **The persistent-connection budget was spent permanently.**
+  `erpl_rfc_max_persistent_connections` was a monotonic counter on bind data that never
+  released a slot, so a second execution of a bound plan began with the budget already
+  exhausted and every scan fell back to per-batch open/close. Slots are now leased:
+  released when the connection is dropped, reset per execution.
+
+- **[rfc]** **A long scan could follow a secret replaced underneath it.** The DuckDB
+  secret was re-resolved on *every* connection open, so replacing it mid-query could send
+  later windows of the same scan to a different SAP system, with no error. Credentials are
+  now resolved once per execution.
+
+- **[rfc]** **Setting a trace option while tracing was enabled deadlocked the process.**
+  `erpl_trace_level`, `erpl_trace_output` and the other setters logged their own change
+  while holding the tracer mutex, which the writer re-locks. The sequence documented for
+  diagnosing SAP communication was itself the trigger; it presents as a hung query.
+
+### Added
+
+- **[rfc]** `sap_rfc_live_connections()`, `sap_rfc_connections_opened()` and
+  `sap_rfc_connections_closed()` report how many SAP RFC connections erpl has opened,
+  closed and still holds. Between queries the live count should be 0 — a non-zero value
+  means SAP sessions are still reserved, which client-side timing cannot reveal.
+
 ## v2026.09.02 — a narrow extract can finally use more than one connection
 
 `sap_read_table` parallelised across *columns*, so reading a few columns out of a very

rfc/src/erpl_rfc_extension.cpp

@@ -190,6 +190,27 @@ namespace duckdb {
         ConstantVector::GetData<string_t>(result)[0] = StringVector::AddString(result, name);
     }
 
+    // Number of RFC connections erpl currently holds open: opened minus closed.
+    //
+    // Exists so a test can assert that a scan RELEASES what it acquired. Every open
+    // connection is a session and a work-process reservation on the SAP system, and
+    // client-side timing shows nothing when one is never released -- the cost is
+    // entirely on the source system.
+    static void RfcLiveConnectionsFunction(DataChunk &, ExpressionState &, Vector &result) {
+        result.SetVectorType(VectorType::CONSTANT_VECTOR);
+        ConstantVector::GetData<int64_t>(result)[0] = RfcConnectionStats::Live();
+    }
+
+    static void RfcConnectionsOpenedFunction(DataChunk &, ExpressionState &, Vector &result) {
+        result.SetVectorType(VectorType::CONSTANT_VECTOR);
+        ConstantVector::GetData<int64_t>(result)[0] = (int64_t)RfcConnectionStats::Opened();
+    }
+
+    static void RfcConnectionsClosedFunction(DataChunk &, ExpressionState &, Vector &result) {
+        result.SetVectorType(VectorType::CONSTANT_VECTOR);
+        ConstantVector::GetData<int64_t>(result)[0] = (int64_t)RfcConnectionStats::Closed();
+    }
+
     static void RegisterConfiguration(ExtensionLoader &loader)
     {
         auto &instance = loader.GetDatabaseInstance();
@@ -366,6 +387,34 @@ namespace duckdb {
             loader.RegisterFunction(std::move(info));
         }
 
+        {
+            // Registered together: same shape, same volatility, same reason to exist.
+            using StatFnPtr = void (*)(DataChunk &, ExpressionState &, Vector &);
+            struct StatFn { const char *name; StatFnPtr fn; const char *doc; };
+            const StatFn stat_fns[] = {
+                {"sap_rfc_live_connections", RfcLiveConnectionsFunction,
+                 "Number of SAP RFC connections erpl currently holds open (opened minus closed). "
+                 "Should be 0 between queries; a non-zero value means SAP sessions are still reserved."},
+                {"sap_rfc_connections_opened", RfcConnectionsOpenedFunction,
+                 "Total SAP RFC connections erpl has opened in this process."},
+                {"sap_rfc_connections_closed", RfcConnectionsClosedFunction,
+                 "Total SAP RFC connections erpl has closed in this process."},
+            };
+            for (auto &sf : stat_fns) {
+                ScalarFunction f(sf.name, {}, LogicalType::BIGINT, sf.fn);
+                // Process-wide state, not a function of the arguments: must never be
+                // constant-folded at bind time or reused across statements.
+                f.stability = FunctionStability::VOLATILE;
+                CreateScalarFunctionInfo info(f);
+                FunctionDescription desc;
+                desc.description = sf.doc;
+                desc.examples    = {std::string("SELECT ") + sf.name + "()"};
+                desc.categories  = {"sap"};
+                info.descriptions.push_back(std::move(desc));
+                loader.RegisterFunction(std::move(info));
+            }
+        }
+
         {
             CreateTableFunctionInfo info(CreateRfcReadTableScanFunction());
             FunctionDescription desc;

rfc/src/erpl_tracing.cpp

@@ -53,43 +53,60 @@ void ErplTracer::SetEnabled(bool enable_flag)
 
 void ErplTracer::SetLevel(TraceLevel trace_level)
 {
-    std::lock_guard<std::mutex> lock(trace_mutex);
-    level = trace_level;
+    // Log AFTER releasing the lock. Info() -> Trace() -> WriteToFile() takes
+    // trace_mutex, which is a plain std::mutex, so logging while holding it
+    // deadlocks the calling thread and every later tracer call with it. That
+    // presents as a hung query, not as a lock error. SetEnabled already scopes
+    // its lock this way; these setters did not.
+    {
+        std::lock_guard<std::mutex> lock(trace_mutex);
+        level = trace_level;
+    }
     Info("TRACER", "Trace level set to: " + LevelToString(trace_level));
 }
 
 void ErplTracer::SetTraceDirectory(const std::string &directory)
 {
-    std::lock_guard<std::mutex> lock(trace_mutex);
-    trace_directory = directory;
-    std::filesystem::path path(directory);
-    if (!std::filesystem::exists(path)) {
-        std::filesystem::create_directories(path);
+    // EnsureTraceFile must stay INSIDE the lock (it touches trace_file); only the
+    // Info() call moves out, for the reason given on SetLevel.
+    {
+        std::lock_guard<std::mutex> lock(trace_mutex);
+        trace_directory = directory;
+        std::filesystem::path path(directory);
+        if (!std::filesystem::exists(path)) {
+            std::filesystem::create_directories(path);
+        }
+        if (enabled) {
+            EnsureTraceFile();
+        }
     }
     Info("TRACER", "Trace directory set to: " + directory);
-    if (enabled) {
-        EnsureTraceFile();
-    }
 }
 
 void ErplTracer::SetOutputMode(const std::string &mode)
 {
-    std::lock_guard<std::mutex> lock(trace_mutex);
-    output_mode = mode;
+    {
+        std::lock_guard<std::mutex> lock(trace_mutex);
+        output_mode = mode;
+    }
     Info("TRACER", "Trace output mode set to: " + mode);
 }
 
 void ErplTracer::SetMaxFileSize(int64_t max_size)
 {
-    std::lock_guard<std::mutex> lock(trace_mutex);
-    max_file_size = max_size;
+    {
+        std::lock_guard<std::mutex> lock(trace_mutex);
+        max_file_size = max_size;
+    }
     Info("TRACER", "Trace max file size set to: " + std::to_string(max_size));
 }
 
 void ErplTracer::SetRotation(bool rotation)
 {
-    std::lock_guard<std::mutex> lock(trace_mutex);
-    rotation_enabled = rotation;
+    {
+        std::lock_guard<std::mutex> lock(trace_mutex);
+        rotation_enabled = rotation;
+    }
     Info("TRACER", "Trace rotation " + std::string(rotation ? "enabled" : "disabled"));
 }
 

rfc/src/include/sap_connection.hpp

@@ -6,6 +6,24 @@
 
 namespace duckdb 
 {
+
+    // Process-wide count of RFC connections erpl has opened and closed.
+    //
+    // Exists so a test can assert that a scan releases what it acquired -- every open
+    // connection is a session and a work-process reservation on the SAP system, and
+    // wall-clock timing does not reveal one that is never released. Deliberately a
+    // plain counter rather than a registry: it must be cheap enough to leave on in
+    // release builds, and it is read by `PRAGMA sap_rfc_connection_stats`.
+    struct RfcConnectionStats {
+        static void NoteOpened();
+        static void NoteClosed();
+        static uint64_t Opened();
+        static uint64_t Closed();
+        // Opened - Closed. Non-zero after a query has finished means erpl is still
+        // holding SAP sessions.
+        static int64_t Live();
+        static void Reset();
+    };
 	typedef struct RfcConnectionAttributes
 	{
 		std::string destination;

rfc/src/include/sap_rfc.hpp

@@ -159,8 +159,17 @@ namespace duckdb
 			void ResolveReadTableImportParams(std::shared_ptr<RfcConnection> connection);
 			bool ReadTableHasParam(const std::string &param_name);
 
-			bool HasMoreResults();
-			void Step(ClientContext &context, DataChunk &output);
+			// The machine set is passed in rather than read from bind data. DuckDB
+			// shares bind data across executions of one bound plan, so state machines
+			// living there make a second scan resume from an exhausted cursor and
+			// return nothing -- silently. The partitioned path already owned its
+			// machines per worker; these overloads let the serial path own them per
+			// execution, in the global state.
+			bool HasMoreResults(std::vector<RfcReadColumnStateMachine> &machines);
+			void Step(ClientContext &context, DataChunk &output,
+			          std::vector<RfcReadColumnStateMachine> &machines);
+			bool AreActiveStateMachineCaridnalitiesEqual(std::vector<RfcReadColumnStateMachine> &machines);
+			unsigned int FirstActiveStateMachineCardinality(std::vector<RfcReadColumnStateMachine> &machines);
 
 			// Per-scan ceiling for the warm-up batch doubling, capped so that
 			// (projected columns x batch_size) stays within a fixed row budget
@@ -196,17 +205,35 @@ namespace duckdb
 			// released until the bind data dies (the connections live for
 			// the whole scan), so the counter is monotonically increasing.
 			bool TryReservePersistentSlot();
+			void ReleasePersistentSlot();
+			void ResetPersistentSlots();
+
+			// Resolve the SAP credentials ONCE per execution and reuse them.
+			// OpenNewConnection used to re-resolve the DuckDB secret on every open, so a
+			// scan that opens a connection per window could silently move to a different
+			// system mid-query if the secret was replaced underneath it.
+			void PinAuthParams();
 
 		private:
 			std::string secret_name;
 			RfcConnectionFactory_t connection_factory;
 			ClientContext &client_context;
+			// Set by PinAuthParams() at init-global; read by OpenNewConnection().
+			std::shared_ptr<RfcAuthParams> pinned_auth;
+			std::mutex pinned_auth_lock;
 			std::vector<std::string> column_names;
 			std::vector<RfcType> column_types;
 			// Columns whose DDIC type is CLNT.  RFC_READ_TABLE's OPTIONS parser rejects
 			// any clause naming the client field, so these are never pushed.
 			std::set<std::string> client_columns;
 			std::vector<RfcReadColumnStateMachine> column_state_machines;
+			// Slots are LEASED, not consumed. The counter used to be a monotonic
+			// fetch_add that never gave a slot back, and it lives on bind data -- which
+			// DuckDB reuses across executions -- so a second execution of a bound plan
+			// started with the budget already spent and every machine was denied a
+			// persistent connection. ResetPersistentSlots() is called once per
+			// execution from init-global; ReleasePersistentSlot() returns a slot when
+			// its connection is dropped.
 			std::atomic<unsigned int> persistent_slots_used{0};
 			// Filters that could not be translated into OPTIONS, keyed by projected
 			// column index.  Copied because the TableFilterSet belongs to the plan.
@@ -229,8 +256,6 @@ namespace duckdb
 
 			std::vector<RfcReadColumnStateMachine> CreateReadColumnStateMachines();
 			unsigned int NActiveStateMachines();
-			unsigned int FirstActiveStateMachineCardinality();
-			bool AreActiveStateMachineCaridnalitiesEqual();
 		public:
 			static std::vector<Value> GetTableFieldMetas(std::shared_ptr<RfcConnection> connection, std::string table_name);
 			static RfcType GetRfcTypeForFieldMeta(Value &DFIES_entry);
@@ -465,6 +490,11 @@ namespace duckdb
 			                        duckdb::unique_ptr<RfcRowWindowScheduler> scheduler_p)
 				: max_threads(max_threads_p), scheduler(std::move(scheduler_p)) { }
 
+			// Serial (unpartitioned) scans own their state machines here, one set per
+			// EXECUTION. Global state is rebuilt for every execution of a bound plan;
+			// bind data is not, which is why these cannot live there.
+			std::vector<RfcReadColumnStateMachine> serial_machines;
+
 			idx_t MaxThreads() const override { return max_threads; }
 			bool IsPartitioned() const { return scheduler != nullptr; }
 

rfc/src/sap_connection.cpp

@@ -1,4 +1,6 @@
 #include "duckdb.hpp"
+#include <atomic>
+
 #include "sap_connection.hpp"
 #include "sap_type_conversion.hpp"
 #include "sap_secret.hpp"
@@ -172,6 +174,7 @@ namespace duckdb
         }
         // Telemetry: feature_used {feature="connection_opened", auth_kind}.
         erpl_telemetry::CaptureConnectionOpened(auth);
+        RfcConnectionStats::NoteOpened();
         return std::make_shared<RfcConnection>(connection_handle);
     }
 
@@ -234,6 +237,11 @@ namespace duckdb
         RFC_ERROR_INFO error_info;
 
         rc = RfcCloseConnection(handle, &error_info);
+        // Counted regardless of rc: the handle is released either way (it is nulled
+        // below), so for "what does erpl still hold open" purposes this connection is
+        // gone. Counting only RFC_OK would report a permanent leak whenever the
+        // gateway had already dropped the connection.
+        RfcConnectionStats::NoteClosed();
         // Regardless of the outcome the handle must not be reused: a second
         // RfcCloseConnection on the same handle yields RFC_INVALID_HANDLE.
         // Nulling here prevents the double-close path (issue #78).
@@ -299,4 +307,27 @@ namespace duckdb
 
     // RfcConnnection -----------------------------------------------------------
 
+
+    // --- RfcConnectionStats ----------------------------------------------------
+
+    namespace {
+        std::atomic<uint64_t> g_connections_opened{0};
+        std::atomic<uint64_t> g_connections_closed{0};
+    }
+
+    void RfcConnectionStats::NoteOpened() { g_connections_opened.fetch_add(1, std::memory_order_relaxed); }
+    void RfcConnectionStats::NoteClosed() { g_connections_closed.fetch_add(1, std::memory_order_relaxed); }
+    uint64_t RfcConnectionStats::Opened()  { return g_connections_opened.load(std::memory_order_relaxed); }
+    uint64_t RfcConnectionStats::Closed()  { return g_connections_closed.load(std::memory_order_relaxed); }
+    int64_t RfcConnectionStats::Live()
+    {
+        return (int64_t)g_connections_opened.load(std::memory_order_relaxed) -
+               (int64_t)g_connections_closed.load(std::memory_order_relaxed);
+    }
+    void RfcConnectionStats::Reset()
+    {
+        g_connections_opened.store(0, std::memory_order_relaxed);
+        g_connections_closed.store(0, std::memory_order_relaxed);
+    }
+
 } // namespace duckdb