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 2e3796b..9790f64 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/odp b/odp index 0d66415..bc04904 160000 --- a/odp +++ b/odp @@ -1 +1 @@ -Subproject commit 0d664158bf346a089aa3b767f63d84285b8de7e2 +Subproject commit bc04904f00724dbc7418bf7dacd8b611e7555eba 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/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/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 b4e7b01..fcec9ca 100644 --- a/rfc/src/include/sap_rfc.hpp +++ b/rfc/src/include/sap_rfc.hpp @@ -159,8 +159,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 @@ -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 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. @@ -229,8 +256,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); @@ -465,6 +490,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 947f4f1..9a0f948 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 = { @@ -1258,9 +1300,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; } @@ -1326,15 +1368,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); } @@ -1376,11 +1419,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); } @@ -1391,9 +1434,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(); } @@ -1402,19 +1446,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; } @@ -1685,6 +1730,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 d12314d..027575d 100644 --- a/rfc/src/scanner_read_table.cpp +++ b/rfc/src/scanner_read_table.cpp @@ -99,6 +99,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(); @@ -108,7 +112,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 @@ -248,6 +258,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 @@ -255,7 +273,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 @@ -266,7 +284,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/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); +} 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