From 15916b524ae87bb0441e1b96a9eb9cafa095274c Mon Sep 17 00:00:00 2001 From: Joachim Rosskopf Date: Fri, 4 Sep 2026 05:28:18 +0200 Subject: [PATCH 1/3] fix(rfc): stop the tracer deadlocking when a setter logs its own change SetLevel, SetTraceDirectory, SetOutputMode, SetMaxFileSize and SetRotation each held trace_mutex across an Info() call. Info() reaches WriteToFile(), which locks the same plain std::mutex, so any of these called while file tracing was active deadlocked the calling thread -- and every later tracer call behind it. It presents as a hung query rather than a lock error, which is why it survived: the sequence CLAUDE.md documents for debugging SAP communication SET erpl_trace_enabled = TRUE; SET erpl_trace_level = 'DEBUG'; SET erpl_trace_output = 'file'; is exactly the sequence that triggers it. I hit this earlier in this work and wrote it off as trace verbosity. SetEnabled already scoped its lock and logged afterwards; the others now do the same. EnsureTraceFile stays inside the lock in SetTraceDirectory because it touches trace_file. The test runs each setter on a DETACHED thread with a deadline. It cannot use std::async: that future joins in its destructor, so the watchdog would hang on the very deadlock it is meant to catch -- which it did on the first attempt. --- rfc/src/erpl_tracing.cpp | 49 ++++++++++++------ rfc/test/cpp/CMakeLists.txt | 1 + rfc/test/cpp/test_tracing_locking.cpp | 73 +++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 16 deletions(-) create mode 100644 rfc/test/cpp/test_tracing_locking.cpp diff --git a/rfc/src/erpl_tracing.cpp b/rfc/src/erpl_tracing.cpp index 8874d79..dadde5f 100644 --- a/rfc/src/erpl_tracing.cpp +++ b/rfc/src/erpl_tracing.cpp @@ -53,43 +53,60 @@ void ErplTracer::SetEnabled(bool enable_flag) void ErplTracer::SetLevel(TraceLevel trace_level) { - std::lock_guard 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 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 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 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 lock(trace_mutex); - output_mode = mode; + { + std::lock_guard lock(trace_mutex); + output_mode = mode; + } Info("TRACER", "Trace output mode set to: " + mode); } void ErplTracer::SetMaxFileSize(int64_t max_size) { - std::lock_guard lock(trace_mutex); - max_file_size = max_size; + { + std::lock_guard 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 lock(trace_mutex); - rotation_enabled = rotation; + { + std::lock_guard lock(trace_mutex); + rotation_enabled = rotation; + } Info("TRACER", "Trace rotation " + std::string(rotation ? "enabled" : "disabled")); } diff --git a/rfc/test/cpp/CMakeLists.txt b/rfc/test/cpp/CMakeLists.txt index aef2c23..69ccc79 100644 --- a/rfc/test/cpp/CMakeLists.txt +++ b/rfc/test/cpp/CMakeLists.txt @@ -25,6 +25,7 @@ set(TEST_SOURCES test_read_table_batching.cpp test_read_table_filters.cpp test_row_window_scheduler.cpp + test_tracing_locking.cpp test_connection_close.cpp test_sap_secret.cpp test_select_supported_args.cpp diff --git a/rfc/test/cpp/test_tracing_locking.cpp b/rfc/test/cpp/test_tracing_locking.cpp new file mode 100644 index 0000000..7aa83df --- /dev/null +++ b/rfc/test/cpp/test_tracing_locking.cpp @@ -0,0 +1,73 @@ +#include "catch.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include "erpl_tracing.hpp" + +using namespace erpl; + +namespace { + +// Run `fn` on a DETACHED thread and report whether it finished in time. +// +// It must be detached: a std::async future joins in its destructor, so a watchdog +// built on it cannot escape the very deadlock it is meant to detect -- it hangs the +// test binary instead, with no diagnosis. That is also how this defect hid in +// production: it looks like a slow query, not a lock cycle. +bool CompletesWithin(std::chrono::milliseconds budget, std::function fn) +{ + auto done = std::make_shared>(false); + std::thread([done, fn = std::move(fn)]() mutable { + fn(); + done->store(true, std::memory_order_release); + }).detach(); + + auto deadline = std::chrono::steady_clock::now() + budget; + while (std::chrono::steady_clock::now() < deadline) { + if (done->load(std::memory_order_acquire)) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return done->load(std::memory_order_acquire); +} + +} // namespace + +TEST_CASE("Tracer setters do not deadlock while tracing is enabled", "[erpl_rfc][tracing]") { + auto &tracer = ErplTracer::Instance(); + + // The setters log their own change through Info(). SetEnabled scopes its lock and + // logs after the scope; the others held trace_mutex across the Info() call, and + // WriteToFile locks the same non-recursive mutex -- so any setter called while + // tracing was writing to a file deadlocked the process. + tracer.SetTraceDirectory("./trace"); + tracer.SetOutputMode("file"); + tracer.SetEnabled(true); + + REQUIRE(CompletesWithin(std::chrono::seconds(5), [&] { + tracer.SetLevel(TraceLevel::DEBUG_LEVEL); + })); + REQUIRE(CompletesWithin(std::chrono::seconds(5), [&] { + tracer.SetOutputMode("both"); + })); + REQUIRE(CompletesWithin(std::chrono::seconds(5), [&] { + tracer.SetMaxFileSize(4 * 1024 * 1024); + })); + REQUIRE(CompletesWithin(std::chrono::seconds(5), [&] { + tracer.SetTraceDirectory("./trace"); + })); + + // The tracer must still work after all that. + REQUIRE(CompletesWithin(std::chrono::seconds(5), [&] { + tracer.Info("TEST", "still alive"); + })); + + tracer.SetEnabled(false); +} From dd41d77e300890452e8115f3a1710e01f5f76500 Mon Sep 17 00:00:00 2001 From: Joachim Rosskopf Date: Fri, 4 Sep 2026 07:10:37 +0200 Subject: [PATCH 2/3] fix(rfc): own scan state per execution, lease connection slots, pin credentials Three defects with one root cause: mutable per-scan state living on RfcReadTableBindData, which DuckDB reuses across executions of a bound plan. 1. A re-scanned sap_read_table returned nothing, silently. PREPARE q AS SELECT count(*) FROM sap_read_table('SFLIGHT'); EXECUTE q; -- 94 EXECUTE q; -- 0 The column state machines were bind-owned, so the second scan resumed from an exhausted cursor. Every 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 is created per execution -- the pattern the partitioned path already used, which is why PARTITIONS was never affected. Step() and HasMoreResults() now take the machine set explicitly instead of reading it from bind data. That is what found the other three sites with the same defect: sap_show_tables, sap_odp_show_subscriptions and the table lister behind ATTACH. A convenience overload reading bind data would have left all three broken. 2. The persistent-connection budget was spent permanently. TryReservePersistentSlot was a monotonic fetch_add on bind data that never released a slot -- not on failure, not when the connection was dropped -- so once `cap` attempts had been made no machine ever won a slot again, including in later executions. Slots are now leased: returned on a failed reservation, returned in InvalidateCachedConnection, reset per execution. 3. A long scan could follow a secret replaced underneath it. OpenNewConnection resolved the DuckDB secret on EVERY open, so replacing it mid-query sent later windows to a different SAP system with no error and nothing in the result to show it. Credentials are resolved once per execution and pinned. Adds sap_rfc_live_connections() / _opened() / _closed(). 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. The live count must be 0 between queries. Tests, written red first: sap_read_table_rescan.test -- prepared statement, content checksum (a count-only assertion passes while rows differ), two scans in one statement, early LIMIT, and the partitioned path as non-regression guard sap_rfc_connection_release.test -- live count returns to baseline after a plain scan, a partitioned scan, an early LIMIT, a FAILING scan, and repeated executions; plus a guard that the counters actually move, so a stub returning 0 would fail Suites: offline partition 11/11, batching 6/6, tracing 1/1. Live, all with zero failures and zero known gaps: RFC nwrfc, RFC proto, ODP nwrfc, ODP proto, BICS nwrfc. --- API_REFERENCE.md | 18 +++ CHANGELOG.md | 45 +++++++ odp | 2 +- rfc/src/erpl_rfc_extension.cpp | 49 +++++++ rfc/src/include/sap_connection.hpp | 18 +++ rfc/src/include/sap_rfc.hpp | 38 +++++- rfc/src/sap_connection.cpp | 31 +++++ rfc/src/sap_rfc.cpp | 83 +++++++++--- rfc/src/sap_storage.cpp | 8 +- rfc/src/scanner_read_table.cpp | 24 +++- rfc/src/scanner_show_tables.cpp | 12 +- rfc/test/sql/sap_read_table_rescan.test | 133 +++++++++++++++++++ rfc/test/sql/sap_rfc_connection_release.test | 118 ++++++++++++++++ 13 files changed, 551 insertions(+), 28 deletions(-) create mode 100644 rfc/test/sql/sap_read_table_rescan.test create mode 100644 rfc/test/sql/sap_rfc_connection_release.test diff --git a/API_REFERENCE.md b/API_REFERENCE.md index cadc1f2..1a8eb71 100644 --- a/API_REFERENCE.md +++ b/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 diff --git a/CHANGELOG.md b/CHANGELOG.md index f6eda60..7a280c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,51 @@ LOAD erpl; --- +## Unreleased + +### Fixed + +- **[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 diff --git a/odp b/odp index 0d66415..ba5c8c0 160000 --- a/odp +++ b/odp @@ -1 +1 @@ -Subproject commit 0d664158bf346a089aa3b767f63d84285b8de7e2 +Subproject commit ba5c8c0b4c9dda9b35ebf052f72b143d26341584 diff --git a/rfc/src/erpl_rfc_extension.cpp b/rfc/src/erpl_rfc_extension.cpp index 87c13a9..14836cf 100644 --- a/rfc/src/erpl_rfc_extension.cpp +++ b/rfc/src/erpl_rfc_extension.cpp @@ -190,6 +190,27 @@ namespace duckdb { ConstantVector::GetData(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(result)[0] = RfcConnectionStats::Live(); + } + + static void RfcConnectionsOpenedFunction(DataChunk &, ExpressionState &, Vector &result) { + result.SetVectorType(VectorType::CONSTANT_VECTOR); + ConstantVector::GetData(result)[0] = (int64_t)RfcConnectionStats::Opened(); + } + + static void RfcConnectionsClosedFunction(DataChunk &, ExpressionState &, Vector &result) { + result.SetVectorType(VectorType::CONSTANT_VECTOR); + ConstantVector::GetData(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; diff --git a/rfc/src/include/sap_connection.hpp b/rfc/src/include/sap_connection.hpp index 069ebb6..127fa7b 100644 --- a/rfc/src/include/sap_connection.hpp +++ b/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; diff --git a/rfc/src/include/sap_rfc.hpp b/rfc/src/include/sap_rfc.hpp index 858fbfd..d445e67 100644 --- a/rfc/src/include/sap_rfc.hpp +++ b/rfc/src/include/sap_rfc.hpp @@ -153,8 +153,17 @@ namespace duckdb void ResolveReadTableImportParams(std::shared_ptr connection); bool ReadTableHasParam(const std::string ¶m_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 &machines); + void Step(ClientContext &context, DataChunk &output, + std::vector &machines); + bool AreActiveStateMachineCaridnalitiesEqual(std::vector &machines); + unsigned int FirstActiveStateMachineCardinality(std::vector &machines); // Per-scan ceiling for the warm-up batch doubling, capped so that // (projected columns x batch_size) stays within a fixed row budget @@ -190,17 +199,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 pinned_auth; + std::mutex pinned_auth_lock; std::vector column_names; std::vector 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 client_columns; std::vector 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 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. @@ -223,8 +250,6 @@ namespace duckdb std::vector CreateReadColumnStateMachines(); unsigned int NActiveStateMachines(); - unsigned int FirstActiveStateMachineCardinality(); - bool AreActiveStateMachineCaridnalitiesEqual(); public: static std::vector GetTableFieldMetas(std::shared_ptr connection, std::string table_name); static RfcType GetRfcTypeForFieldMeta(Value &DFIES_entry); @@ -459,6 +484,11 @@ namespace duckdb duckdb::unique_ptr 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 serial_machines; + idx_t MaxThreads() const override { return max_threads; } bool IsPartitioned() const { return scheduler != nullptr; } diff --git a/rfc/src/sap_connection.cpp b/rfc/src/sap_connection.cpp index efab42d..28fedba 100644 --- a/rfc/src/sap_connection.cpp +++ b/rfc/src/sap_connection.cpp @@ -1,4 +1,6 @@ #include "duckdb.hpp" +#include + #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(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 g_connections_opened{0}; + std::atomic 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 \ No newline at end of file diff --git a/rfc/src/sap_rfc.cpp b/rfc/src/sap_rfc.cpp index 5056712..e94e5b2 100644 --- a/rfc/src/sap_rfc.cpp +++ b/rfc/src/sap_rfc.cpp @@ -218,9 +218,32 @@ namespace duckdb return ret; } + void RfcReadTableBindData::PinAuthParams() + { + if (secret_name.empty()) { + return; + } + // Resolve once per execution. Re-resolving on every open let a scan follow a + // secret that was replaced mid-query -- half its windows read one system and + // half another, with no error and no way to tell from the result. + auto resolved = std::make_shared( + RfcAuthParams::FromContext(client_context, secret_name)); + std::lock_guard guard(pinned_auth_lock); + pinned_auth = std::move(resolved); + } + std::shared_ptr RfcReadTableBindData::OpenNewConnection() { if (!secret_name.empty()) { + std::shared_ptr auth; + { + std::lock_guard guard(pinned_auth_lock); + auth = pinned_auth; + } + if (auth) { + return auth->Connect(); + } + // No pin (a helper that never went through init-global): resolve now. return RfcAuthParams::FromContext(client_context, secret_name).Connect(); } return connection_factory(client_context); @@ -264,11 +287,30 @@ namespace duckdb } auto previous = persistent_slots_used.fetch_add(1, std::memory_order_relaxed); if (previous >= cap) { + // Give the slot straight back. Leaving it counted made the budget + // monotonic: once `cap` attempts had been made the counter never fell + // below the cap again, so every later machine -- and every later + // execution, since this lives on bind data -- was denied. + persistent_slots_used.fetch_sub(1, std::memory_order_relaxed); return false; } return true; } + void RfcReadTableBindData::ReleasePersistentSlot() + { + auto previous = persistent_slots_used.load(std::memory_order_relaxed); + while (previous > 0 && + !persistent_slots_used.compare_exchange_weak(previous, previous - 1, + std::memory_order_relaxed)) { + } + } + + void RfcReadTableBindData::ResetPersistentSlots() + { + persistent_slots_used.store(0, std::memory_order_relaxed); + } + void RfcReadTableBindData::ValidateReadTableFunctionName() { static const std::set allowed_functions = { @@ -1246,9 +1288,9 @@ namespace duckdb active, budget); } - bool RfcReadTableBindData::HasMoreResults() + bool RfcReadTableBindData::HasMoreResults(std::vector &machines) { - for (auto &sm : column_state_machines) { + for (auto &sm : machines) { if (sm.Active() && !sm.Finished()) { return true; } @@ -1314,15 +1356,16 @@ namespace duckdb output.SetCardinality(cardinality); } - void RfcReadTableBindData::Step(ClientContext &context, DataChunk &output) + void RfcReadTableBindData::Step(ClientContext &context, DataChunk &output, + std::vector &machines) { auto &scheduler = TaskScheduler::GetScheduler(context); // Snapshot the active state machines so we can throttle scheduling - // without iterating column_state_machines twice. + // without iterating the machine set twice. std::vector active; - active.reserve(column_state_machines.size()); - for (auto &sm : column_state_machines) { + active.reserve(machines.size()); + for (auto &sm : machines) { if (sm.Active()) { active.push_back(&sm); } @@ -1364,11 +1407,11 @@ namespace duckdb executor.WorkOnTasks(); } - if (! AreActiveStateMachineCaridnalitiesEqual()) { + if (! AreActiveStateMachineCaridnalitiesEqual(machines)) { throw std::runtime_error("Cardinality of column state machines is not the same. This should not happen."); } - auto cardinality = FirstActiveStateMachineCardinality(); + auto cardinality = FirstActiveStateMachineCardinality(machines); output.SetCardinality(cardinality); } @@ -1379,9 +1422,10 @@ namespace duckdb } - unsigned int RfcReadTableBindData::FirstActiveStateMachineCardinality() + unsigned int RfcReadTableBindData::FirstActiveStateMachineCardinality( + std::vector &machines) { - for (auto &sm : column_state_machines) { + for (auto &sm : machines) { if (sm.Active()) { return sm.GetCardinality(); } @@ -1390,19 +1434,20 @@ namespace duckdb } - bool RfcReadTableBindData::AreActiveStateMachineCaridnalitiesEqual() + bool RfcReadTableBindData::AreActiveStateMachineCaridnalitiesEqual( + std::vector &machines) { - if (column_state_machines.empty()) { + if (machines.empty()) { return false; } - auto ref_state_machine = std::find_if(column_state_machines.begin(), column_state_machines.end(), + auto ref_state_machine = std::find_if(machines.begin(), machines.end(), [](auto &sm) { return sm.Active(); }); - if (ref_state_machine == column_state_machines.end()) { + if (ref_state_machine == machines.end()) { throw std::runtime_error("No active state machine found. This should not happen."); } - for (auto &sm : column_state_machines) { + for (auto &sm : machines) { if (! sm.Active()) { continue; } @@ -1665,6 +1710,14 @@ namespace duckdb // Caller already holds thread_lock. cached_function.reset(); cached_function_name.clear(); + // The slot is a lease on a live cached connection. Dropping the connection + // returns it, so a partitioned worker that finishes a window does not keep the + // budget reserved for a connection it no longer holds, and the next window (or + // another machine) can win it. + if (persistent_decision == PersistentDecision::APPROVED && bind_data != nullptr) { + bind_data->ReleasePersistentSlot(); + persistent_decision = PersistentDecision::UNDECIDED; + } if (cached_connection) { try { cached_connection->Close(); diff --git a/rfc/src/sap_storage.cpp b/rfc/src/sap_storage.cpp index fdc5d7a..6d15f1d 100644 --- a/rfc/src/sap_storage.cpp +++ b/rfc/src/sap_storage.cpp @@ -347,9 +347,13 @@ static vector ResolveTablePatterns(ClientContext &context, const string DataChunk chunk; chunk.Initialize(Allocator::Get(context), bind_data->GetReturnTypes()); - while (bind_data->HasMoreResults()) { + // Local machine set: this helper builds fresh bind data per call, so there is + // no re-scan hazard here, but the state machines are per-scan state and are + // now owned as such rather than living on the shared bind data. + auto machines = bind_data->CreateWindowStateMachines(); + while (bind_data->HasMoreResults(machines)) { chunk.Reset(); - bind_data->Step(context, chunk); + bind_data->Step(context, chunk, machines); for (idx_t i = 0; i < chunk.size(); i++) { auto val = chunk.GetValue(0, i); if (val.IsNull()) { diff --git a/rfc/src/scanner_read_table.cpp b/rfc/src/scanner_read_table.cpp index 7cb7dd8..28d18a4 100644 --- a/rfc/src/scanner_read_table.cpp +++ b/rfc/src/scanner_read_table.cpp @@ -98,6 +98,10 @@ namespace duckdb bind_data.ActivateColumns(column_ids); bind_data.AddOptionsFromFilters(input.filters); + // Per-execution setup: pin the credentials for this run, and hand back every + // persistent-connection slot the previous execution left counted. + bind_data.PinAuthParams(); + bind_data.ResetPersistentSlots(); // The serial path computes this inside Step(); a partitioned scan needs it // before any worker starts, and it must not be written afterwards. bind_data.ResolveEffectiveMaxBatchSize(); @@ -107,7 +111,13 @@ namespace duckdb // runs the column-parallel path exactly as it always has. auto partitions = bind_data.GetPartitionCount(); if (partitions <= 1) { - return make_uniq(1, nullptr); + // Build this execution's OWN machines. Global state is created per + // execution; bind data is not. Reusing the bind-owned set made the second + // EXECUTE of a prepared statement resume from an exhausted cursor and + // return zero rows with no error. + auto gstate = make_uniq(1, nullptr); + gstate->serial_machines = bind_data.CreateWindowStateMachines(); + return std::move(gstate); } // Use the same batch size the unpartitioned path warms up to, not @@ -229,6 +239,14 @@ namespace duckdb } } + // Serial path. The machines belong to this execution's global state, so a + // re-scan of the same bound plan starts from INIT instead of resuming an + // exhausted cursor. + if (data.global_state == nullptr) { + throw InternalException("sap_read_table: serial scan without global state"); + } + auto &serial_machines = data.global_state->Cast().serial_machines; + // Loop, because an empty chunk is how a table function says "scan finished". // Residual filtering can legitimately reject every row of a batch, and returning // that empty chunk would end the scan and silently discard everything still @@ -236,7 +254,7 @@ namespace duckdb // and returned 25,578 rows instead of 114,566. Keep pulling until a batch has a // surviving row or the table is genuinely exhausted. while (true) { - if (! bind_data.HasMoreResults()) { + if (! bind_data.HasMoreResults(serial_machines)) { #ifdef __GLIBC__ // Scan finished: per-column SDK handles were released at FINISHED // and the streaming reader holds no whole-batch buffers, so hand @@ -247,7 +265,7 @@ namespace duckdb return; } - bind_data.Step(context, output); + bind_data.Step(context, output, serial_machines); if (! bind_data.HasResidualFilters()) { return; } diff --git a/rfc/src/scanner_show_tables.cpp b/rfc/src/scanner_show_tables.cpp index 88f5a2c..8c0db07 100644 --- a/rfc/src/scanner_show_tables.cpp +++ b/rfc/src/scanner_show_tables.cpp @@ -63,7 +63,12 @@ static unique_ptr RfcShowTablesInitGlobalState(ClientC bind_data.ActivateColumns(column_ids); - return make_uniq(); + // Own the state machines per EXECUTION, not per bind: DuckDB reuses bind data + // across executions of a bound plan, so bind-owned machines make a re-scan resume + // from an exhausted cursor and return nothing. + auto gstate = make_uniq(1, nullptr); + gstate->serial_machines = bind_data.CreateWindowStateMachines(); + return std::move(gstate); } static void RfcShowTablesScan(ClientContext &context, @@ -71,12 +76,13 @@ static void RfcShowTablesScan(ClientContext &context, DataChunk &output) { auto &bind_data = data.bind_data->CastNoConst(); + auto &machines = data.global_state->Cast().serial_machines; - if (! bind_data.HasMoreResults()) { + if (! bind_data.HasMoreResults(machines)) { return; } - bind_data.Step(context, output); + bind_data.Step(context, output, machines); } TableFunction CreateRfcShowTablesScanFunction() diff --git a/rfc/test/sql/sap_read_table_rescan.test b/rfc/test/sql/sap_read_table_rescan.test new file mode 100644 index 0000000..9c753b1 --- /dev/null +++ b/rfc/test/sql/sap_read_table_rescan.test @@ -0,0 +1,133 @@ +# name: test/sql/sap_read_table_rescan.test +# description: A plan may scan one table function more than once. Every execution must +# return the full result. +# +# sap_read_table kept its column state machines in BIND data, which DuckDB +# shares across executions of the same bound plan. The second scan therefore +# resumed from an exhausted cursor and returned nothing: +# +# PREPARE q AS SELECT count(*) FROM sap_read_table('SFLIGHT'); +# EXECUTE q; -- 94 +# EXECUTE q; -- 0 <-- silent, no error +# +# Wrong results with no error, on ordinary usage: a prepared statement, a +# nested-loop join, or an un-materialised CTE referenced twice. +# +# The PARTITIONS cases are the non-regression guard: that path already +# builds its machines into per-worker local state and was always correct, +# and it is the pattern the fix follows. +# group: [rfc] + +require erpl_rfc + +require-env ERPL_SAP_ASHOST + +require-env ERPL_SAP_SYSNR + +require-env ERPL_SAP_USER + +require-env ERPL_SAP_PASSWORD + +require-env ERPL_SAP_CLIENT + +require-env ERPL_SAP_LANG + +statement ok +CREATE SECRET abap_trial ( + TYPE sap_rfc, + ASHOST '${ERPL_SAP_ASHOST}', + SYSNR '${ERPL_SAP_SYSNR}', + CLIENT '${ERPL_SAP_CLIENT}', + USER '${ERPL_SAP_USER}', + PASSWD '${ERPL_SAP_PASSWORD}', + LANG '${ERPL_SAP_LANG}' +); + +# ------------------------------------------------------------------------- +# 1. A prepared statement executed repeatedly. +# ------------------------------------------------------------------------- +statement ok +PREPARE q_count AS SELECT count(*) FROM sap_read_table('SFLIGHT'); + +query I +EXECUTE q_count; +---- +94 + +query I +EXECUTE q_count; +---- +94 + +query I +EXECUTE q_count; +---- +94 + +# ------------------------------------------------------------------------- +# 2. Row VALUES must be stable across executions, not merely the count. +# A count-only assertion passes while the rows differ, so this hashes the +# content. sum(hash(..)) is order-insensitive, which matters because +# RFC_READ_TABLE gives no ordering guarantee. +# ------------------------------------------------------------------------- +statement ok +PREPARE q_ck AS SELECT sum(hash(CARRID || CONNID)::HUGEINT) AS ck FROM sap_read_table('SFLIGHT'); + +query I +EXECUTE q_ck; +---- +635163459561984897428 + +query I +EXECUTE q_ck; +---- +635163459561984897428 + +# ------------------------------------------------------------------------- +# 3. One statement that scans the same function twice. +# ------------------------------------------------------------------------- +query I +SELECT (SELECT count(*) FROM sap_read_table('SFLIGHT')) + + (SELECT count(*) FROM sap_read_table('SFLIGHT')); +---- +188 + +# ------------------------------------------------------------------------- +# 4. A LIMIT that stops the scan early must not poison the next execution. +# This is the sharper case: the first execution leaves the machines mid-table +# rather than exhausted. +# ------------------------------------------------------------------------- +statement ok +PREPARE q_limited AS SELECT count(*) FROM (SELECT CARRID FROM sap_read_table('SFLIGHT') LIMIT 2); + +query I +EXECUTE q_limited; +---- +2 + +query I +EXECUTE q_limited; +---- +2 + +query I +EXECUTE q_limited; +---- +2 + +# ------------------------------------------------------------------------- +# 5. Non-regression: the partitioned path builds per-worker local state and +# was already correct. It must stay correct. +# ------------------------------------------------------------------------- +statement ok +PREPARE q_part AS SELECT count(*) FROM sap_read_table('SFLIGHT', PARTITIONS=4); + +query I +EXECUTE q_part; +---- +94 + +query I +EXECUTE q_part; +---- +94 diff --git a/rfc/test/sql/sap_rfc_connection_release.test b/rfc/test/sql/sap_rfc_connection_release.test new file mode 100644 index 0000000..5f08fed --- /dev/null +++ b/rfc/test/sql/sap_rfc_connection_release.test @@ -0,0 +1,118 @@ +# name: test/sql/sap_rfc_connection_release.test +# description: A scan must release every RFC connection it opens. +# +# Each open connection is a session and a work-process reservation on the +# SAP system. Client-side timing shows nothing when one is never released +# -- the whole cost lands on the source system -- so this asserts the +# connection count directly. +# +# sap_rfc_live_connections() is opened-minus-closed, process-wide. Between +# statements it must be 0. It is deliberately asserted as a DELTA against a +# baseline taken in the same session rather than as an absolute, so the +# file does not depend on what ran before it. +# group: [rfc] + +require erpl_rfc + +require-env ERPL_SAP_ASHOST + +require-env ERPL_SAP_SYSNR + +require-env ERPL_SAP_USER + +require-env ERPL_SAP_PASSWORD + +require-env ERPL_SAP_CLIENT + +require-env ERPL_SAP_LANG + +statement ok +CREATE SECRET abap_trial ( + TYPE sap_rfc, + ASHOST '${ERPL_SAP_ASHOST}', + SYSNR '${ERPL_SAP_SYSNR}', + CLIENT '${ERPL_SAP_CLIENT}', + USER '${ERPL_SAP_USER}', + PASSWD '${ERPL_SAP_PASSWORD}', + LANG '${ERPL_SAP_LANG}' +); + +statement ok +CREATE TABLE baseline AS SELECT sap_rfc_live_connections() AS live; + +# ------------------------------------------------------------------------- +# 1. A plain scan releases everything it opened. +# ------------------------------------------------------------------------- +statement ok +SELECT count(*) FROM sap_read_table('SFLIGHT'); + +query I +SELECT sap_rfc_live_connections() - (SELECT live FROM baseline); +---- +0 + +# ------------------------------------------------------------------------- +# 2. A partitioned scan opens several connections; all must come back. +# ------------------------------------------------------------------------- +statement ok +SELECT count(*) FROM sap_read_table('SFLIGHT', PARTITIONS=4); + +query I +SELECT sap_rfc_live_connections() - (SELECT live FROM baseline); +---- +0 + +# ------------------------------------------------------------------------- +# 3. A LIMIT abandons the scan early -- the connections still must not leak. +# This is the path where a worker never reaches a natural end. +# ------------------------------------------------------------------------- +statement ok +SELECT CARRID FROM sap_read_table('SFLIGHT', PARTITIONS=4) LIMIT 1; + +query I +SELECT sap_rfc_live_connections() - (SELECT live FROM baseline); +---- +0 + +# ------------------------------------------------------------------------- +# 4. A failing scan must not strand a connection either. +# ------------------------------------------------------------------------- +statement error +SELECT count(*) FROM sap_read_table('THIS_TABLE_DOES_NOT_EXIST_XYZ'); +---- + +query I +SELECT sap_rfc_live_connections() - (SELECT live FROM baseline); +---- +0 + +# ------------------------------------------------------------------------- +# 5. Repeated executions of one prepared plan must not accumulate connections. +# The persistent-slot budget used to be monotonic and bind-scoped, so it +# behaved differently on the second execution than on the first. +# ------------------------------------------------------------------------- +statement ok +PREPARE q_rel AS SELECT count(*) FROM sap_read_table('SFLIGHT'); + +statement ok +EXECUTE q_rel; + +statement ok +EXECUTE q_rel; + +statement ok +EXECUTE q_rel; + +query I +SELECT sap_rfc_live_connections() - (SELECT live FROM baseline); +---- +0 + +# ------------------------------------------------------------------------- +# 6. And the counters must actually be moving -- a stub returning 0 would pass +# every assertion above. +# ------------------------------------------------------------------------- +query I +SELECT sap_rfc_connections_opened() > 0 AND sap_rfc_connections_closed() > 0; +---- +true From 07e34b3a2253de90cc4f93698de1853215e3cb70 Mon Sep 17 00:00:00 2001 From: Joachim Rosskopf Date: Fri, 4 Sep 2026 08:20:41 +0200 Subject: [PATCH 3/3] chore: point odp at the merged per-execution scan-state fix (erpl-odp#11) --- odp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/odp b/odp index ba5c8c0..bc04904 160000 --- a/odp +++ b/odp @@ -1 +1 @@ -Subproject commit ba5c8c0b4c9dda9b35ebf052f72b143d26341584 +Subproject commit bc04904f00724dbc7418bf7dacd8b611e7555eba