diff --git a/.mapping.json b/.mapping.json index 684fa4ba249d..19dc4816636b 100644 --- a/.mapping.json +++ b/.mapping.json @@ -3640,14 +3640,23 @@ "odbc/functional_tests/basic_chaos/schemas/postgresql/key_value.sql":"taxi/uservices/userver/odbc/functional_tests/basic_chaos/schemas/postgresql/key_value.sql", "odbc/functional_tests/basic_chaos/static_config.yaml":"taxi/uservices/userver/odbc/functional_tests/basic_chaos/static_config.yaml", "odbc/functional_tests/basic_chaos/tests/conftest.py":"taxi/uservices/userver/odbc/functional_tests/basic_chaos/tests/conftest.py", + "odbc/functional_tests/basic_chaos/tests/test_metrics.py":"taxi/uservices/userver/odbc/functional_tests/basic_chaos/tests/test_metrics.py", "odbc/functional_tests/basic_chaos/tests/test_odbc.py":"taxi/uservices/userver/odbc/functional_tests/basic_chaos/tests/test_odbc.py", + "odbc/functional_tests/secdist_update/CMakeLists.txt":"taxi/uservices/userver/odbc/functional_tests/secdist_update/CMakeLists.txt", + "odbc/functional_tests/secdist_update/odbc_service.cpp":"taxi/uservices/userver/odbc/functional_tests/secdist_update/odbc_service.cpp", + "odbc/functional_tests/secdist_update/schemas/postgresql/key_value.sql":"taxi/uservices/userver/odbc/functional_tests/secdist_update/schemas/postgresql/key_value.sql", + "odbc/functional_tests/secdist_update/static_config.yaml":"taxi/uservices/userver/odbc/functional_tests/secdist_update/static_config.yaml", + "odbc/functional_tests/secdist_update/tests/conftest.py":"taxi/uservices/userver/odbc/functional_tests/secdist_update/tests/conftest.py", + "odbc/functional_tests/secdist_update/tests/test_secdist.py":"taxi/uservices/userver/odbc/functional_tests/secdist_update/tests/test_secdist.py", "odbc/include/userver/storages/odbc.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc.hpp", "odbc/include/userver/storages/odbc/cluster.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/cluster.hpp", "odbc/include/userver/storages/odbc/cluster_types.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/cluster_types.hpp", + "odbc/include/userver/storages/odbc/command_control.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/command_control.hpp", "odbc/include/userver/storages/odbc/component.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/component.hpp", "odbc/include/userver/storages/odbc/exception.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/exception.hpp", "odbc/include/userver/storages/odbc/execution_result.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/execution_result.hpp", "odbc/include/userver/storages/odbc/field.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/field.hpp", + "odbc/include/userver/storages/odbc/impl/parameter.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/impl/parameter.hpp", "odbc/include/userver/storages/odbc/impl/tracing_tags.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/impl/tracing_tags.hpp", "odbc/include/userver/storages/odbc/odbc_fwd.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/odbc_fwd.hpp", "odbc/include/userver/storages/odbc/query.hpp":"taxi/uservices/userver/odbc/include/userver/storages/odbc/query.hpp", @@ -3691,7 +3700,6 @@ "odbc/src/storages/odbc/dsn.hpp":"taxi/uservices/userver/odbc/src/storages/odbc/dsn.hpp", "odbc/src/storages/odbc/exception.cpp":"taxi/uservices/userver/odbc/src/storages/odbc/exception.cpp", "odbc/src/storages/odbc/field.cpp":"taxi/uservices/userver/odbc/src/storages/odbc/field.cpp", - "odbc/src/storages/odbc/odbc_config.hpp":"taxi/uservices/userver/odbc/src/storages/odbc/odbc_config.hpp", "odbc/src/storages/odbc/odbc_secdist.cpp":"taxi/uservices/userver/odbc/src/storages/odbc/odbc_secdist.cpp", "odbc/src/storages/odbc/odbc_secdist.hpp":"taxi/uservices/userver/odbc/src/storages/odbc/odbc_secdist.hpp", "odbc/src/storages/odbc/result_set.cpp":"taxi/uservices/userver/odbc/src/storages/odbc/result_set.cpp", @@ -6670,4 +6678,4 @@ "ydb/utest/include/userver/ydb/tests/topic_writer_mock.hpp":"taxi/uservices/userver/ydb/utest/include/userver/ydb/tests/topic_writer_mock.hpp", "ydb/utest/include/userver/ydb/tests/write_session_mock.hpp":"taxi/uservices/userver/ydb/utest/include/userver/ydb/tests/write_session_mock.hpp", "ydb/utest/src/utest/topic_writer_mock.cpp":"taxi/uservices/userver/ydb/utest/src/utest/topic_writer_mock.cpp" -} \ No newline at end of file +} diff --git a/odbc/README.md b/odbc/README.md index e1f9e26ac6da..689babcf0114 100644 --- a/odbc/README.md +++ b/odbc/README.md @@ -1,229 +1,4 @@ -# userver: ODBC Driver Wrapper [WIP] - -ODBC storage wrapper for `userver` (cluster + connection pool + query execution). - -Under active development! - -## Quick start - -Create a `storages::odbc::Cluster` with ODBC DSN and execute a query: - -```cpp -#include - -using namespace std::chrono_literals; - -storages::odbc::settings::PoolSettings pool_settings{ - .min_size=1, - .max_size=5, -}; - -storages::odbc::settings::HostSettings host_settings{ - .dsn="DRIVER={PostgreSQL Unicode};SERVER=localhost;PORT=15433;DATABASE=postgres;UID=testsuite;PWD=password;", - .pool=pool_settings, -}; - -storages::odbc::settings::ODBCClusterSettings cluster_settings{ - .pools={host_settings}, -}; - -storages::odbc::Cluster cluster{cluster_settings}; - -auto rs = cluster.Execute(storages::odbc::ClusterHostType::kMaster, "SELECT 1"); -auto row = rs[0]; -auto field = row[0]; -// field.GetInt32() / GetInt64() / GetString() / ... -``` - -### Reading results - -`Execute(...)` returns `storages::odbc::ResultSet`. Each row is `storages::odbc::Row`, and each field is `storages::odbc::Field`: - -```cpp -auto rs = cluster.Execute(storages::odbc::ClusterHostType::kMaster, - "SELECT 42, 'test', 1.0, false, null, true"); - -const auto row = rs[0]; -const auto i32 = row[0].GetInt32(); -const auto str = row[1].GetString(); -if (row[4].IsNull()) { - // ... -} -``` - -## Deadlines - -ODBC operations can be aborted when a deadline is reached. - -### Explicit deadline - -Use the overloads that accept `engine::Deadline`: - -```cpp -#include -#include - -using namespace std::chrono_literals; - -auto deadline = engine::Deadline::FromDuration(200ms); -auto rs = cluster.Execute(deadline, storages::odbc::ClusterHostType::kMaster, "SELECT 1"); -``` - -Deadlines are also applied to transactions started with `Begin(deadline, ...)`: - -```cpp -auto tx = cluster.Begin(deadline, storages::odbc::ClusterHostType::kMaster); -auto rs = tx.Execute("SELECT 1"); -tx.Commit(); // deadline is honored internally -``` - -### Deadline resolution in ODBC - -ODBC driver statement timeout (`SQL_ATTR_QUERY_TIMEOUT`) is configured in whole seconds. -As a result, when converting `engine::Deadline` to the ODBC timeout, sub-second deadlines are rounded up to the next full second (so the operation may run slightly longer than the exact deadline). - -### Request deadline propagation - -If you call ODBC from a request task, the task-inherited request deadline is automatically merged into ODBC deadlines. -If it expires, `storages::odbc::OperationInterrupted` is thrown. - -## Transactions - -Transactions are created via `Cluster::Begin(...)`. -They execute statements via `Transaction::Execute(...)`, then finish with `Commit()` or `Rollback()`. -If neither commit nor rollback was called, the transaction rolls back on destruction (RAII). - -```cpp -auto tx = cluster.Begin(storages::odbc::ClusterHostType::kMaster); - -tx.Execute("INSERT INTO t(a) VALUES (1)"); -tx.Execute("UPDATE t SET a = a + 1 WHERE a = 1"); - -tx.Commit(); -``` - -## Exceptions - -Common exceptions from `userver::storages::odbc`: - -- `storages::odbc::OperationInterrupted` — deadline expired -- `storages::odbc::ConnectionError` — connection / driver failures -- `storages::odbc::StatementError` — statement-level execution errors - -## Component Configuration - -The ODBC component can be configured in the static config file. Below is the full schema: - -### Single Pool Configuration - -```yaml -components_manager: - components: - odbc: - dsn: "DRIVER={PostgreSQL Unicode};SERVER=localhost;PORT=5432;DATABASE=mydb;UID=user;PWD=password" - min_pool_size: 1 # optional, default: 1 - max_pool_size: 10 # optional, default: 10 - dns_resolver: async # optional, default: async (options: async, getaddrinfo) -``` - -### Multi-Pool Configuration - -For master-replica setups or multiple database hosts: - -```yaml -components_manager: - components: - odbc: - dns_resolver: async - pools: - - dsn: "DRIVER={PostgreSQL Unicode};SERVER=master.db.local;PORT=5432;DATABASE=mydb;UID=user;PWD=password" - min_pool_size: 2 - max_pool_size: 15 - - dsn: "DRIVER={PostgreSQL Unicode};SERVER=replica.db.local;PORT=5432;DATABASE=mydb;UID=user;PWD=password" - min_pool_size: 1 - max_pool_size: 10 -``` - -### Configuration Options - -| Option | Type | Default | Description | -|--------|------|---------|-------------| -| `secdist_alias` | string | — | Name of the database in secdist config (for secure credential storage) | -| `dsn` | string | — | ODBC connection string (for single-pool mode) | -| `min_pool_size` | integer | 1 | Minimum number of connections kept in the pool | -| `max_pool_size` | integer | 10 | Maximum number of connections in the pool | -| `dns_resolver` | string | `async` | DNS resolution mode: `async` (non-blocking) or `getaddrinfo` (blocking) | -| `pools` | array | — | List of pool configurations (for multi-pool mode) | - -### Secdist Integration - -For secure credential storage, you can use secdist instead of putting DSN strings directly in the static config: - -```yaml -components_manager: - components: - odbc: - secdist_alias: my_database - min_pool_size: 1 - max_pool_size: 10 - dns_resolver: async -``` - -The secdist JSON file should contain: - -```json -{ - "odbc_settings": { - "databases": { - "my_database": { - "dsn": "DRIVER={PostgreSQL Unicode};SERVER=localhost;PORT=5432;DATABASE=mydb;UID=user;PWD=secret" - } - } - } -} -``` - -For multiple hosts (master-replica setup): - -```json -{ - "odbc_settings": { - "databases": { - "my_database": { - "hosts": [ - "DRIVER={PostgreSQL Unicode};SERVER=master.db.local;PORT=5432;DATABASE=mydb;UID=user;PWD=secret", - "DRIVER={PostgreSQL Unicode};SERVER=replica.db.local;PORT=5432;DATABASE=mydb;UID=user;PWD=secret" - ] - } - } - } -} -``` - -### DNS Resolution - -The `dns_resolver` option controls how hostnames in DSN strings are resolved: - -- **`async`** (default): Uses userver's asynchronous DNS resolver. Hostnames are resolved at component startup and replaced with IP addresses in the DSN. This is non-blocking and recommended for production. - -- **`getaddrinfo`**: Uses the system's blocking `getaddrinfo()` call. The DSN is passed to the ODBC driver as-is, and hostname resolution happens during connection establishment. - -When using `async` mode, the SERVER/HOST parameter in your DSN will be automatically resolved to an IP address before connecting. This allows for proper integration with service discovery and DNS-based load balancing. - - -### Programmatic Access - -You can access the current dynamic config values programmatically: - -```cpp -#include - -// Get current default timeouts from dynamic config -auto network_timeout = cluster->GetDefaultNetworkTimeout(); -auto statement_timeout = cluster->GetDefaultStatementTimeout(); - -if (network_timeout.has_value()) { - // Use the configured timeout -} -``` +# userver ODBC driver +The driver documentation is maintained in +[`scripts/docs/en/userver/odbc.md`](../scripts/docs/en/userver/odbc.md). diff --git a/odbc/dynamic_configs/USERVER_ODBC_CONNECTION_POOL_SETTINGS.yaml b/odbc/dynamic_configs/USERVER_ODBC_CONNECTION_POOL_SETTINGS.yaml index 075d26ff8e9f..e1f9b51cf836 100644 --- a/odbc/dynamic_configs/USERVER_ODBC_CONNECTION_POOL_SETTINGS.yaml +++ b/odbc/dynamic_configs/USERVER_ODBC_CONNECTION_POOL_SETTINGS.yaml @@ -6,8 +6,9 @@ description: | The latter configuration is applied for every non-matching ODBC component of the service. - Note: Pool size changes require component restart to take effect, - as the underlying connection pool does not support dynamic resizing. + Updates atomically replace pools for new operations. Queries and + transactions already in progress keep their old pools alive until they + finish. schema: type: object example: | diff --git a/odbc/dynamic_configs/USERVER_ODBC_DEFAULT_COMMAND_CONTROL.yaml b/odbc/dynamic_configs/USERVER_ODBC_DEFAULT_COMMAND_CONTROL.yaml index 5ddc6456589b..8db2cd4bb8a5 100644 --- a/odbc/dynamic_configs/USERVER_ODBC_DEFAULT_COMMAND_CONTROL.yaml +++ b/odbc/dynamic_configs/USERVER_ODBC_DEFAULT_COMMAND_CONTROL.yaml @@ -1,8 +1,9 @@ default: {} description: | Dynamic config that controls default network and statement timeouts for ODBC driver. - Overrides the built-in timeouts, but could be overridden by explicit - engine::Deadline passed to Execute/Begin methods. + Overrides the built-in timeouts. Individual operations can override these + values with storages::odbc::OptionalCommandControl. A task-inherited + request deadline always caps the resulting deadline. schema: type: object additionalProperties: false @@ -12,8 +13,11 @@ schema: minimum: 1 x-usrv-cpp-type: std::chrono::milliseconds description: | - Network timeout in milliseconds. Controls how long to wait for - network operations (connection establishment, data transfer). + Overall operation budget in milliseconds. It starts before + connection acquisition, configures the ODBC login timeout, and + caps statement and transaction deadlines. Blocking ODBC calls + are ultimately subject to timeout support and whole-second + resolution of the selected ODBC driver. statement_timeout_ms: type: integer minimum: 1 diff --git a/odbc/dynamic_configs/USERVER_ODBC_HANDLERS_COMMAND_CONTROL.yaml b/odbc/dynamic_configs/USERVER_ODBC_HANDLERS_COMMAND_CONTROL.yaml new file mode 100644 index 000000000000..5f30d3f0f1b2 --- /dev/null +++ b/odbc/dynamic_configs/USERVER_ODBC_HANDLERS_COMMAND_CONTROL.yaml @@ -0,0 +1,27 @@ +default: {} +description: | + Dynamic config for per-HTTP-handler ODBC timeouts. Keys are handler paths, + then HTTP methods. Values override only their specified fields from + USERVER_ODBC_DEFAULT_COMMAND_CONTROL and are overridden by named-query and + explicit per-call command controls. +schema: + type: object + additionalProperties: + $ref: "#/definitions/CommandControlByMethodMap" + definitions: + CommandControlByMethodMap: + type: object + additionalProperties: + $ref: "#/definitions/CommandControl" + CommandControl: + type: object + additionalProperties: false + properties: + network_timeout_ms: + type: integer + minimum: 1 + x-usrv-cpp-type: std::chrono::milliseconds + statement_timeout_ms: + type: integer + minimum: 1 + x-usrv-cpp-type: std::chrono::milliseconds diff --git a/odbc/dynamic_configs/USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS.yaml b/odbc/dynamic_configs/USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS.yaml new file mode 100644 index 000000000000..a283d8baa8ef --- /dev/null +++ b/odbc/dynamic_configs/USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS.yaml @@ -0,0 +1,42 @@ +default: {} +description: | + Dynamic config that controls the per-connection prepared statement cache for + ODBC components. + + Dictionary keys can be either an ODBC component name or `__default__`. + An exact component entry takes precedence over `__default__`, which takes + precedence over the component's static `max_prepared_cache_size` value. + Zero disables the cache and clears retained prepared statements before the + next operation on each physical connection. +schema: + type: object + example: | + { + "odbc-orders": { + "max_prepared_cache_size": 50 + }, + "__default__": { + "max_prepared_cache_size": 10 + } + } + properties: + __default__: + $ref: "#/definitions/PreparedStatementCacheSettings" + additionalProperties: + $ref: "#/definitions/PreparedStatementCacheSettings" + definitions: + PreparedStatementCacheSettings: + type: object + additionalProperties: false + properties: + max_prepared_cache_size: + type: integer + minimum: 0 + default: 0 + x-usrv-cpp-type: std::size_t + description: | + Maximum number of prepared parameterized SQL statements + retained per physical ODBC connection. Zero disables and + clears the cache. + required: + - max_prepared_cache_size diff --git a/odbc/dynamic_configs/USERVER_ODBC_QUERIES_COMMAND_CONTROL.yaml b/odbc/dynamic_configs/USERVER_ODBC_QUERIES_COMMAND_CONTROL.yaml new file mode 100644 index 000000000000..c92a82faa940 --- /dev/null +++ b/odbc/dynamic_configs/USERVER_ODBC_QUERIES_COMMAND_CONTROL.yaml @@ -0,0 +1,23 @@ +default: {} +description: | + Dynamic config for named ODBC query timeouts. Keys are + storages::odbc::Query names. Values override only their specified fields + from default and handler command controls and are overridden by explicit + per-call command controls. Unnamed queries do not use this config. +schema: + type: object + additionalProperties: + $ref: "#/definitions/CommandControl" + definitions: + CommandControl: + type: object + additionalProperties: false + properties: + network_timeout_ms: + type: integer + minimum: 1 + x-usrv-cpp-type: std::chrono::milliseconds + statement_timeout_ms: + type: integer + minimum: 1 + x-usrv-cpp-type: std::chrono::milliseconds diff --git a/odbc/dynamic_configs/USERVER_ODBC_STATEMENT_METRICS_SETTINGS.yaml b/odbc/dynamic_configs/USERVER_ODBC_STATEMENT_METRICS_SETTINGS.yaml new file mode 100644 index 000000000000..7b4b4601ea6d --- /dev/null +++ b/odbc/dynamic_configs/USERVER_ODBC_STATEMENT_METRICS_SETTINGS.yaml @@ -0,0 +1,43 @@ +default: {} +description: | + Dynamic config that controls named query metrics for ODBC components. + + Dictionary keys can be either an ODBC component name or `__default__`. + An exact component entry takes precedence over `__default__`, which takes + precedence over the component's static `max_statement_metrics` value. + + The bound is applied independently to every pool labelled by `odbc_pool`. + A value of 0 disables accounting and clears retained named query-name entries. + Each retained entry exports three metric series. +schema: + type: object + example: | + { + "odbc-orders": { + "max_statement_metrics": 50 + }, + "__default__": { + "max_statement_metrics": 10 + } + } + properties: + __default__: + $ref: "#/definitions/StatementMetricsSettings" + additionalProperties: + $ref: "#/definitions/StatementMetricsSettings" + definitions: + StatementMetricsSettings: + type: object + additionalProperties: false + properties: + max_statement_metrics: + type: integer + minimum: 0 + default: 0 + x-usrv-cpp-type: std::size_t + description: | + Maximum number of named query-name entries retained per pool. + Each entry exports three metric series. Zero disables named + query metrics. + required: + - max_statement_metrics diff --git a/odbc/functional_tests/CMakeLists.txt b/odbc/functional_tests/CMakeLists.txt index 6d5e1f776a94..0a208dc02779 100644 --- a/odbc/functional_tests/CMakeLists.txt +++ b/odbc/functional_tests/CMakeLists.txt @@ -4,3 +4,6 @@ add_custom_target(${PROJECT_NAME}) add_subdirectory(basic_chaos) add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-basic-chaos) + +add_subdirectory(secdist_update) +add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-secdist-update) diff --git a/odbc/functional_tests/basic_chaos/odbc_service.cpp b/odbc/functional_tests/basic_chaos/odbc_service.cpp index 3887a6d23c2b..0f838381e7a7 100644 --- a/odbc/functional_tests/basic_chaos/odbc_service.cpp +++ b/odbc/functional_tests/basic_chaos/odbc_service.cpp @@ -1,12 +1,17 @@ #include #include +#include #include #include +#include #include +#include +#include #include #include #include +#include #include namespace chaos { @@ -43,10 +48,8 @@ class KeyValue final : public server::handlers::HttpHandlerBase { private: std::string GetValue(std::string_view key, const server::http::HttpRequest& request) const { - auto result = odbc_->Execute( - storages::odbc::ClusterHostType::kMaster, - fmt::format("SELECT value FROM kv WHERE key = '{}'", key) - ); + auto result = + odbc_->Execute(storages::odbc::ClusterHostType::kMaster, "SELECT value FROM kv WHERE key = ?", key); if (result.IsEmpty()) { request.SetResponseStatus(server::http::HttpStatus::kNotFound); @@ -64,12 +67,10 @@ class KeyValue final : public server::handlers::HttpHandlerBase { odbc_->Execute( storages::odbc::ClusterHostType::kMaster, - fmt::format( - "INSERT INTO kv(key, value) VALUES ('{}', '{}') " - "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", - key, - value - ) + "INSERT INTO kv(key, value) VALUES (?, ?) " + "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + key, + value ); request.SetResponseStatus(server::http::HttpStatus::kCreated); @@ -77,7 +78,7 @@ class KeyValue final : public server::handlers::HttpHandlerBase { } std::string DeleteValue(std::string_view key) const { - odbc_->Execute(storages::odbc::ClusterHostType::kMaster, fmt::format("DELETE FROM kv WHERE key = '{}'", key)); + odbc_->Execute(storages::odbc::ClusterHostType::kMaster, "DELETE FROM kv WHERE key = ?", key); return {}; } @@ -118,7 +119,7 @@ class KeyValueTrx final : public server::handlers::HttpHandlerBase { private: std::string GetValue(std::string_view key, const server::http::HttpRequest& request) const { auto trx = odbc_->Begin(storages::odbc::ClusterHostType::kMaster); - auto result = trx.Execute(fmt::format("SELECT value FROM kv WHERE key = '{}'", key)); + auto result = trx.Execute("SELECT value FROM kv WHERE key = ?", key); trx.Commit(); if (result.IsEmpty()) { @@ -136,12 +137,12 @@ class KeyValueTrx final : public server::handlers::HttpHandlerBase { } auto trx = odbc_->Begin(storages::odbc::ClusterHostType::kMaster); - trx.Execute(fmt::format( - "INSERT INTO kv(key, value) VALUES ('{}', '{}') " + trx.Execute( + "INSERT INTO kv(key, value) VALUES (?, ?) " "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", key, value - )); + ); trx.Commit(); request.SetResponseStatus(server::http::HttpStatus::kCreated); @@ -150,7 +151,7 @@ class KeyValueTrx final : public server::handlers::HttpHandlerBase { std::string DeleteValue(std::string_view key) const { auto trx = odbc_->Begin(storages::odbc::ClusterHostType::kMaster); - trx.Execute(fmt::format("DELETE FROM kv WHERE key = '{}'", key)); + trx.Execute("DELETE FROM kv WHERE key = ?", key); trx.Commit(); return {}; @@ -159,13 +160,64 @@ class KeyValueTrx final : public server::handlers::HttpHandlerBase { const std::shared_ptr odbc_; }; +class CommandControl final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName{"handler-command-control"}; + + CommandControl(const components::ComponentConfig& config, const components::ComponentContext& context) + : server::handlers::HttpHandlerBase{config, context}, + odbc_{context.FindComponent("key-value-db").GetCluster()} + {} + + std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override { + const storages::odbc::Query query{ + "SELECT pg_sleep(2)", + storages::odbc::Query::Name{"odbc-functional-sleep"}, + }; + odbc_->Execute(storages::odbc::ClusterHostType::kMaster, query); + return "ok"; + } + +private: + const std::shared_ptr odbc_; +}; + +class StatementMetrics final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName{"handler-statement-metrics"}; + + StatementMetrics(const components::ComponentConfig& config, const components::ComponentContext& context) + : server::handlers::HttpHandlerBase{config, context}, + odbc_{context.FindComponent("key-value-db").GetCluster()} + {} + + std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override { + const storages::odbc::Query query{ + "SELECT 1", + storages::odbc::Query::Name{"odbc-functional-statement-metrics"}, + }; + odbc_->Execute(storages::odbc::ClusterHostType::kMaster, query); + return "ok"; + } + +private: + const std::shared_ptr odbc_; +}; + } // namespace chaos int main(int argc, char* argv[]) { const auto component_list = components::MinimalServerComponentList() + .AppendComponentList(USERVER_NAMESPACE::dynamic_config::updater::ComponentList()) + .AppendComponentList(clients::http::ComponentList()) .Append() .Append() + .Append() + .Append() + .Append() + .Append() + .Append() .Append() .Append() .Append() diff --git a/odbc/functional_tests/basic_chaos/static_config.yaml b/odbc/functional_tests/basic_chaos/static_config.yaml index 73edd4461515..65f59b367a0c 100644 --- a/odbc/functional_tests/basic_chaos/static_config.yaml +++ b/odbc/functional_tests/basic_chaos/static_config.yaml @@ -10,10 +10,25 @@ components_manager: task_processor: main-task-processor method: GET,POST,DELETE + handler-command-control: + path: /command-control + task_processor: main-task-processor + method: GET + + handler-statement-metrics: + path: /statement-metrics + task_processor: main-task-processor + method: GET + + # [ODBC component config] key-value-db: + blocking_task_processor: fs-task-processor secdist_alias: key-value-db min_pool_size: 1 max_pool_size: 1 + max_statement_metrics: 2 + max_prepared_cache_size: 2 + # [ODBC component config] secdist: {} default-secdist-provider: @@ -25,6 +40,44 @@ components_manager: listener: port: 8097 task_processor: main-task-processor + listener-monitor: + port: $monitor-server-port + port#fallback: 8098 + task_processor: main-task-processor + + handler-server-monitor: + path: /service/monitor + method: GET + task_processor: main-task-processor + + testsuite-support: + + dynamic-config-client: + config-url: $config-server-url + http-retries: 5 + http-timeout: 20s + service-name: testsuite-support + dynamic-config-client-updater: + config-settings: false + first-update-fail-ok: true + full-update-interval: 1m + update-interval: 5s + dynamic-config: + updates-enabled: true + + http-client: {} + http-client-core: + destination-metrics-auto-max-size: 0 + fs-task-processor: fs-task-processor + + tests-control: + method: POST + path: /tests/{action} + task_processor: main-task-processor + testpoint-url: $mockserver/testpoint + testpoint-timeout: 10s + skip-unregistered-testpoints: true + throttling_enabled: false logging: fs-task-processor: fs-task-processor loggers: diff --git a/odbc/functional_tests/basic_chaos/tests/test_metrics.py b/odbc/functional_tests/basic_chaos/tests/test_metrics.py new file mode 100644 index 000000000000..f9cda421cab2 --- /dev/null +++ b/odbc/functional_tests/basic_chaos/tests/test_metrics.py @@ -0,0 +1,182 @@ +import asyncio + + +_STATEMENT_QUERY_LABEL = 'odbc_query=odbc-functional-statement-metrics' + + +async def _statement_metric_lines(monitor_client): + metrics = await monitor_client.metrics_raw(output_format='pretty') + return [ + line + for line in metrics.splitlines() + if line.startswith('odbc.statement_') + and _STATEMENT_QUERY_LABEL in line + ] + + +async def _wait_for_statement_metrics(monitor_client): + for _ in range(50): + lines = await _statement_metric_lines(monitor_client) + if lines: + return lines + await asyncio.sleep(0.02) + return [] + + +async def _prepared_cache_metric_lines(monitor_client): + metrics = await monitor_client.metrics_raw(output_format='pretty') + prefixes = ( + 'odbc.queries.prepared-cache-hits', + 'odbc.queries.prepared-cache-misses', + 'odbc.queries.prepared-cache-evictions', + 'odbc.connections.prepared-statements', + ) + return [line for line in metrics.splitlines() if line.startswith(prefixes)] + + +def _prepared_cache_current(lines): + line = next( + line + for line in lines + if line.startswith('odbc.connections.prepared-statements') + ) + return int(line.rsplit('\t', 1)[-1]) + + +async def test_odbc_metrics_smoke(service_client, monitor_client): + response = await service_client.post('/chaos?key=metrics&value=value') + assert response.status == 201 + + response = await service_client.get('/chaos/trx?key=metrics') + assert response.status == 200 + + metrics = await monitor_client.metrics_raw(output_format='pretty') + odbc_metrics = [line for line in metrics.splitlines() if line.startswith('odbc.')] + + assert odbc_metrics + assert any('component=key-value-db' in line for line in odbc_metrics) + assert any('odbc_pool=0' in line for line in odbc_metrics) + assert any(line.startswith('odbc.queries.executed') for line in odbc_metrics) + assert any(line.startswith('odbc.transactions.committed') for line in odbc_metrics) + + +async def test_odbc_metrics_portability(service_client): + warnings = await service_client.metrics_portability() + assert not warnings + + +async def test_statement_metrics_generated_config_precedence_and_reset( + service_client, + monitor_client, + dynamic_config, +): + # Empty dynamic config falls back to static max_statement_metrics. + response = await service_client.get('/statement-metrics') + assert response.status == 200 + assert await _wait_for_statement_metrics(monitor_client) + + # __default__ overrides the static fallback and explicit zero clears it. + dynamic_config.set( + USERVER_ODBC_STATEMENT_METRICS_SETTINGS={ + '__default__': {'max_statement_metrics': 0}, + }, + ) + response = await service_client.get('/statement-metrics') + assert response.status == 200 + assert not await _statement_metric_lines(monitor_client) + + # Exact component name wins over __default__. + dynamic_config.set( + USERVER_ODBC_STATEMENT_METRICS_SETTINGS={ + '__default__': {'max_statement_metrics': 0}, + 'key-value-db': {'max_statement_metrics': 2}, + }, + ) + response = await service_client.get('/statement-metrics') + assert response.status == 200 + assert await _wait_for_statement_metrics(monitor_client) + + dynamic_config.set( + USERVER_ODBC_STATEMENT_METRICS_SETTINGS={ + '__default__': {'max_statement_metrics': 2}, + 'key-value-db': {'max_statement_metrics': 0}, + }, + ) + response = await service_client.get('/statement-metrics') + assert response.status == 200 + assert not await _statement_metric_lines(monitor_client) + + # Removing dynamic entries restores the static fallback. + dynamic_config.set(USERVER_ODBC_STATEMENT_METRICS_SETTINGS={}) + response = await service_client.get('/statement-metrics') + assert response.status == 200 + lines = await _wait_for_statement_metrics(monitor_client) + assert all('component=key-value-db' in line for line in lines) + assert all('odbc_pool=0' in line for line in lines) + assert all('SELECT 1' not in line for line in lines) + assert any(line.startswith('odbc.statement_timings') for line in lines) + assert any(line.startswith('odbc.statement_executed') for line in lines) + assert any(line.startswith('odbc.statement_errors') for line in lines) + + +async def test_prepared_cache_generated_config_precedence_metrics_and_reset( + service_client, + monitor_client, + dynamic_config, +): + async def execute_twice(value): + for _ in range(2): + response = await service_client.post( + f'/chaos?key=prepared-cache&value={value}', + ) + assert response.status == 201 + + # Empty dynamic config falls back to the static cache size. + await execute_twice('static') + lines = await _prepared_cache_metric_lines(monitor_client) + assert len(lines) == 4 + assert _prepared_cache_current(lines) > 0 + + # __default__ overrides static, and explicit zero clears before execution. + dynamic_config.set( + USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS={ + '__default__': {'max_prepared_cache_size': 0}, + }, + ) + await execute_twice('disabled-default') + lines = await _prepared_cache_metric_lines(monitor_client) + assert _prepared_cache_current(lines) == 0 + + # Exact component entry wins over __default__. + dynamic_config.set( + USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS={ + '__default__': {'max_prepared_cache_size': 0}, + 'key-value-db': {'max_prepared_cache_size': 2}, + }, + ) + await execute_twice('exact-enabled') + assert _prepared_cache_current( + await _prepared_cache_metric_lines(monitor_client), + ) > 0 + + dynamic_config.set( + USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS={ + '__default__': {'max_prepared_cache_size': 2}, + 'key-value-db': {'max_prepared_cache_size': 0}, + }, + ) + await execute_twice('exact-disabled') + assert _prepared_cache_current( + await _prepared_cache_metric_lines(monitor_client), + ) == 0 + + # Removing dynamic entries restores the static baseline. + dynamic_config.set(USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS={}) + await execute_twice('restored-static') + lines = await _prepared_cache_metric_lines(monitor_client) + assert len(lines) == 4 + assert _prepared_cache_current(lines) > 0 + assert all('component=key-value-db' in line for line in lines) + assert all('odbc_pool=0' in line for line in lines) + assert all('odbc_query=' not in line for line in lines) + assert all('SELECT ' not in line for line in lines) diff --git a/odbc/functional_tests/basic_chaos/tests/test_odbc.py b/odbc/functional_tests/basic_chaos/tests/test_odbc.py index b468455cc2bd..8babe7708f17 100644 --- a/odbc/functional_tests/basic_chaos/tests/test_odbc.py +++ b/odbc/functional_tests/basic_chaos/tests/test_odbc.py @@ -42,6 +42,35 @@ async def test_odbc_transaction_happy_path(service_client): await _check_crud(service_client, url=CHAOS_TRX_URL) +async def test_dynamic_command_control_generated_configs_and_reset( + service_client, + dynamic_config, +): + dynamic_config.set( + USERVER_ODBC_HANDLERS_COMMAND_CONTROL={ + '/command-control': { + 'GET': {'statement_timeout_ms': 100}, + }, + }, + ) + response = await service_client.get('/command-control') + assert response.status >= 500 + + dynamic_config.set( + USERVER_ODBC_HANDLERS_COMMAND_CONTROL={}, + USERVER_ODBC_QUERIES_COMMAND_CONTROL={ + 'odbc-functional-sleep': {'statement_timeout_ms': 100}, + }, + ) + response = await service_client.get('/command-control') + assert response.status >= 500 + + dynamic_config.set(USERVER_ODBC_QUERIES_COMMAND_CONTROL={}) + response = await service_client.get('/command-control') + assert response.status == 200 + assert response.text == 'ok' + + async def test_odbc_transaction_multi_statement(service_client): await _cleanup(service_client, 'multi_key1', 'multi_key2') diff --git a/odbc/functional_tests/secdist_update/CMakeLists.txt b/odbc/functional_tests/secdist_update/CMakeLists.txt new file mode 100644 index 000000000000..a16d090a637a --- /dev/null +++ b/odbc/functional_tests/secdist_update/CMakeLists.txt @@ -0,0 +1,9 @@ +project(userver-odbc-tests-secdist-update CXX) + +add_executable(${PROJECT_NAME} "odbc_service.cpp") +target_link_libraries(${PROJECT_NAME} userver::odbc) + +userver_chaos_testsuite_add( + ENV "TESTSUITE_PGSQL_SERVER_START_TIMEOUT=120.0" + RESOURCE_LOCKS userver_postgresql +) diff --git a/odbc/functional_tests/secdist_update/odbc_service.cpp b/odbc/functional_tests/secdist_update/odbc_service.cpp new file mode 100644 index 000000000000..dc7873e05690 --- /dev/null +++ b/odbc/functional_tests/secdist_update/odbc_service.cpp @@ -0,0 +1,51 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace odbc::secdist_update { + +class Handler final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName{"handler-odbc"}; + + Handler(const components::ComponentConfig& config, const components::ComponentContext& context) + : server::handlers::HttpHandlerBase{config, context}, + // Deliberately cache the stable Cluster identity. A secdist update + // must reconfigure this object in place. + cluster_{context.FindComponent("odbc-database").GetCluster()} + {} + + std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override { + const auto result = cluster_->Execute(storages::odbc::ClusterHostType::kMaster, "SELECT ?::integer", 42); + return std::to_string(result[0][0].GetInt32()); + } + +private: + const std::shared_ptr cluster_; +}; + +} // namespace odbc::secdist_update + +int main(int argc, char* argv[]) { + const auto component_list = + components::MinimalServerComponentList() + .AppendComponentList(clients::http::ComponentList()) + .Append() + .Append() + .Append() + .Append() + .Append() + .Append() + .Append("odbc-database"); + return utils::DaemonMain(argc, argv, component_list); +} diff --git a/odbc/functional_tests/secdist_update/schemas/postgresql/key_value.sql b/odbc/functional_tests/secdist_update/schemas/postgresql/key_value.sql new file mode 100644 index 000000000000..e0ac49d1ecfb --- /dev/null +++ b/odbc/functional_tests/secdist_update/schemas/postgresql/key_value.sql @@ -0,0 +1 @@ +SELECT 1; diff --git a/odbc/functional_tests/secdist_update/static_config.yaml b/odbc/functional_tests/secdist_update/static_config.yaml new file mode 100644 index 000000000000..ee56379feaf4 --- /dev/null +++ b/odbc/functional_tests/secdist_update/static_config.yaml @@ -0,0 +1,58 @@ +components_manager: + components: + handler-odbc: + path: /odbc + task_processor: main-task-processor + method: GET + + odbc-database: + blocking_task_processor: fs-task-processor + secdist_alias: odbc-test + min_pool_size: 0 + max_pool_size: 1 + dns_resolver: getaddrinfo + + testsuite-support: + + http-client: {} + http-client-core: + fs-task-processor: fs-task-processor + + tests-control: + method: POST + path: /tests/{action} + task_processor: main-task-processor + testpoint-url: $mockserver/testpoint + testpoint-timeout: 10s + skip-unregistered-testpoints: true + throttling_enabled: false + + secdist: + load-enabled: true + update-period: 1s + default-secdist-provider: + config: /etc/odbc_service/secdist.json + + server: + listener: + port: 8099 + task_processor: main-task-processor + + logging: + fs-task-processor: fs-task-processor + loggers: + default: + file_path: '@stderr' + level: info + overflow_behavior: discard + + dns-client: + fs-task-processor: fs-task-processor + + task_processors: + main-task-processor: + worker_threads: 4 + fs-task-processor: + worker_threads: 2 + + default_task_processor: main-task-processor diff --git a/odbc/functional_tests/secdist_update/tests/conftest.py b/odbc/functional_tests/secdist_update/tests/conftest.py new file mode 100644 index 000000000000..e9fb9e4e6de5 --- /dev/null +++ b/odbc/functional_tests/secdist_update/tests/conftest.py @@ -0,0 +1,45 @@ +import json + +import pytest +from testsuite.databases.pgsql import discover + +pytest_plugins = ['pytest_userver.plugins.postgresql'] +USERVER_CONFIG_HOOKS = ['userver_config_secdist'] + + +@pytest.fixture(name='pgsql_local', scope='session') +def _pgsql_local(service_source_dir, pgsql_local_create): + databases = discover.find_schemas( + 'pg', + [service_source_dir.joinpath('schemas/postgresql')], + ) + return pgsql_local_create(list(databases.values())) + + +@pytest.fixture(scope='session') +def secdist_path(service_tmpdir): + path = service_tmpdir / 'secdist.json' + path.write_text( + json.dumps({ + 'odbc_settings': { + 'databases': { + 'odbc-test': { + 'dsn': ( + 'Driver={PostgreSQL Unicode};Server=localhost;' + 'Port=1;Database=postgres;Uid=testsuite;Pwd=;' + ), + }, + }, + }, + }), + ) + return path + + +@pytest.fixture(scope='session') +def userver_config_secdist(secdist_path): + def _hook(config_yaml, _config_vars): + components = config_yaml['components_manager']['components'] + components['default-secdist-provider']['config'] = str(secdist_path) + + return _hook diff --git a/odbc/functional_tests/secdist_update/tests/test_secdist.py b/odbc/functional_tests/secdist_update/tests/test_secdist.py new file mode 100644 index 000000000000..21c23353e5e9 --- /dev/null +++ b/odbc/functional_tests/secdist_update/tests/test_secdist.py @@ -0,0 +1,52 @@ +import asyncio +import json +import os + + +def _valid_dsn(pgsql_local): + database = pgsql_local['key_value'] + return ( + f'Driver={{PostgreSQL Unicode}};Server={database.host};' + f'Port={database.port};Database={database.dbname};' + f'Uid={database.user or "testsuite"};Pwd={database.password or ""};' + ) + + +def _replace_secdist(path, dsn): + temporary = path.with_suffix('.tmp') + temporary.write_text( + json.dumps({ + 'odbc_settings': { + 'databases': {'odbc-test': {'dsn': dsn}}, + }, + }), + ) + os.replace(temporary, path) + + +async def test_secdist_hot_reload(service_client, secdist_path, pgsql_local, testpoint): + failed = await service_client.get('/odbc') + assert failed.status == 500 + + @testpoint('odbc-new-dsn-list') + def new_dsn_list(_data): + pass + + await service_client.update_server_state() + + # Periodic secdist notifications with unchanged data must not rebuild all + # pools or reset their metrics. + await asyncio.sleep(1.2) + assert new_dsn_list.times_called == 0 + + _replace_secdist(secdist_path, _valid_dsn(pgsql_local)) + await new_dsn_list.wait_call(timeout=10) + + for _ in range(20): + response = await service_client.get('/odbc') + if response.status == 200: + assert response.text == '42' + return + await asyncio.sleep(0.1) + + raise AssertionError('cached ODBC cluster did not recover after secdist update') diff --git a/odbc/include/userver/storages/odbc.hpp b/odbc/include/userver/storages/odbc.hpp index 5857f22d88d9..85bad39f0ed6 100644 --- a/odbc/include/userver/storages/odbc.hpp +++ b/odbc/include/userver/storages/odbc.hpp @@ -4,9 +4,16 @@ /// This file is mainly for documentation purposes and inclusion of all headers /// that are required for working with ODBC µserver component. +#include #include +#include #include +#include #include +#include +#include +#include +#include USERVER_NAMESPACE_BEGIN diff --git a/odbc/include/userver/storages/odbc/bulk.hpp b/odbc/include/userver/storages/odbc/bulk.hpp new file mode 100644 index 000000000000..40add3d2f53e --- /dev/null +++ b/odbc/include/userver/storages/odbc/bulk.hpp @@ -0,0 +1,134 @@ +#pragma once + +/// @file userver/storages/odbc/bulk.hpp +/// @brief Owning parameters and execution outcome for ODBC bulk DML. + +#include +#include +#include +#include + +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +inline constexpr std::size_t kDefaultBulkRows = 1000; + +/// Status of one input row in an ODBC bulk execution. +enum class BulkRowStatus { + /// Driver confirmed successful execution without diagnostics. + kSuccess, + /// Driver confirmed success and reported warning-class diagnostics. + kSuccessWithInfo, + /// Driver reported that this row failed. + kError, + /// Driver reported that this row was not used. + kUnused, + /// The row was processed, but row-specific diagnostics are unavailable. + kDiagnosticsUnavailable, + /// The driver did not provide a trustworthy per-row status. + kUnknown, +}; + +/// @brief Owning, ordered rows of parameters for ODBC bulk DML. +/// +/// The first row fixes the column count and every later row must match it. +/// Columns must also have one normalized type in every row. Use an empty +/// `std::optional` for SQL NULL; raw `nullptr` and `std::nullopt` are +/// untyped and are rejected by bulk preflight. +class BulkParameterStore final { +public: + BulkParameterStore() = default; + BulkParameterStore(const BulkParameterStore&) = delete; + BulkParameterStore(BulkParameterStore&&) noexcept = default; + BulkParameterStore& operator=(const BulkParameterStore&) = delete; + BulkParameterStore& operator=(BulkParameterStore&&) noexcept = default; + + /// Append a row copied from values accepted by ParameterStore. + template + requires((impl::kIsParameterArgument && ...)) + BulkParameterStore& PushBackRow(const Args&... args) { + AppendRow(impl::MakeParameterList(args...)); + return *this; + } + + /// Append a row, transferring its owned parameter values. + BulkParameterStore& PushBackRow(ParameterStore&& row); + + bool IsEmpty() const noexcept { return rows_.empty(); } + std::size_t RowsCount() const noexcept { return rows_.size(); } + std::size_t ColumnsCount() const noexcept { return columns_count_; } + +private: + friend class Cluster; + friend class Transaction; + + void AppendRow(impl::ParameterList row); + const impl::ParameterRows& GetRows() const noexcept { return rows_; } + + impl::ParameterRows rows_; + std::size_t columns_count_{0}; +}; + +/// @brief Outcome snapshot of an ODBC bulk execution. +/// +/// `Processed()` is absent when the driver did not provide a reliable count. +/// `RowsAffected()` is absent when any DML result reported an unknown count. +/// `Succeeded()` counts only kSuccess and kSuccessWithInfo rows; unknown and +/// diagnostics-unavailable rows are deliberately not assumed successful. +class BulkResult final { +public: + BulkResult() = default; + BulkResult( + std::size_t requested, + std::optional processed, + std::optional rows_affected, + std::vector statuses + ); + + /// Number of input rows. Always equals `Statuses().size()`. + std::size_t Requested() const noexcept { return requested_; } + /// Reliable processed-row count, if supplied by the driver. + std::optional Processed() const noexcept { return processed_; } + /// Number of rows with a confirmed successful status. + std::size_t Succeeded() const noexcept { return succeeded_; } + /// Checked aggregate DML row count, or null when any count is unknown. + std::optional RowsAffected() const noexcept { return rows_affected_; } + /// One status for every requested row, including unused tail rows. + const std::vector& Statuses() const noexcept { return statuses_; } + +private: + std::size_t requested_{0}; + std::optional processed_{0}; + std::size_t succeeded_{0}; + std::optional rows_affected_{0}; + std::vector statuses_; +}; + +/// Bulk DML failed after possibly executing a subset of the requested rows. +/// +/// The result snapshot is observational: execution is never retried after +/// `SQLExecute` starts. In direct autocommit mode, completed chunks or scalar +/// fallback rows may already be committed. Use `Transaction::ExecuteBulk` and +/// roll back on failure when atomicity is required. +class BulkExecutionError : public StatementError { +public: + BulkExecutionError( + std::string message, + std::vector diagnostics, + BulkResult result, + bool invalid_handle = false + ); + + const BulkResult& GetResult() const noexcept { return result_; } + +private: + BulkResult result_; +}; + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/cluster.hpp b/odbc/include/userver/storages/odbc/cluster.hpp index 92c87ceb6d45..0eefb6206f78 100644 --- a/odbc/include/userver/storages/odbc/cluster.hpp +++ b/odbc/include/userver/storages/odbc/cluster.hpp @@ -7,10 +7,15 @@ #include #include -#include +#include #include +#include #include +#include +#include +#include +#include #include #include #include @@ -20,11 +25,10 @@ USERVER_NAMESPACE_BEGIN namespace storages::odbc { -struct CommandControl; - namespace detail { class ClusterImpl; +struct BulkLayout; using ClusterImplPtr = std::unique_ptr; } // namespace detail @@ -33,22 +37,159 @@ using ClusterImplPtr = std::unique_ptr; class Cluster { public: Cluster(const settings::ODBCClusterSettings& settings, clients::dns::Resolver* resolver); + Cluster( + const settings::ODBCClusterSettings& settings, + clients::dns::Resolver* resolver, + engine::TaskProcessor& blocking_task_processor + ); ~Cluster(); - ResultSet Execute(ClusterHostTypeFlags flags, const Query& query); - - ResultSet Execute(engine::Deadline deadline, ClusterHostTypeFlags flags, const Query& query); + /// @brief Execute a statement, binding every argument to an ODBC `?` placeholder. + /// + /// @warning Never interpolate untrusted values into @p query. Passing them as + /// separate arguments ensures that they are sent to the ODBC driver as data. + template + requires((impl::kIsParameterArgument && ...)) + ResultSet Execute(ClusterHostTypeFlags flags, const Query& query, const Args&... args) { + return Execute(flags, std::nullopt, query, args...); + } + + /// @brief Execute a statement with per-operation timeout overrides. + template + requires((impl::kIsParameterArgument && ...)) + ResultSet Execute( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const Args&... args + ) { + return DoExecute(command_control, flags, query, impl::MakeParameterList(args...)); + } + + /// @brief Execute a statement with an owning dynamic parameter list. + ResultSet Execute(ClusterHostTypeFlags flags, const Query& query, const ParameterStore& store); + + /// @brief Execute a statement with a dynamic parameter list and timeout overrides. + ResultSet Execute( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const ParameterStore& store + ); + + /// @brief Execute a row-producing statement as an incremental cursor. + /// + /// @warning The cursor pins a pooled connection until it becomes terminal. + template + requires((impl::kIsParameterArgument && ...)) + Cursor ExecuteCursor(ClusterHostTypeFlags flags, const Query& query, const Args&... args) { + return ExecuteCursor(flags, std::nullopt, query, args...); + } + + /// @brief Execute an incremental cursor with per-operation timeout + /// overrides. The resolved durations are reused as a fresh budget for every + /// Fetch call. + template + requires((impl::kIsParameterArgument && ...)) + Cursor ExecuteCursor( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const Args&... args + ) { + return DoExecuteCursor(command_control, flags, query, impl::MakeParameterList(args...)); + } + + Cursor ExecuteCursor(ClusterHostTypeFlags flags, const Query& query, const ParameterStore& store); + + Cursor ExecuteCursor( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const ParameterStore& store + ); + + /// Execute rows as bounded chunks of DML that must not return result sets. + /// Earlier chunks may remain committed if a later row fails; no executed + /// chunk is retried. Use a Transaction when rollback atomicity is needed. + BulkResult ExecuteBulk( + ClusterHostTypeFlags flags, + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows = kDefaultBulkRows + ); + + /// Execute bulk DML with per-operation timeout overrides. + BulkResult ExecuteBulk( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows = kDefaultBulkRows + ); Transaction Begin(ClusterHostTypeFlags flags); - Transaction Begin(engine::Deadline deadline, ClusterHostTypeFlags flags); + Transaction Begin(ClusterHostTypeFlags flags, OptionalCommandControl command_control); + + /// Start a transaction with explicit ODBC isolation/access options. + Transaction Begin(ClusterHostTypeFlags flags, const TransactionOptions& options); + + /// Start a transaction with explicit options and timeout overrides. + Transaction Begin( + ClusterHostTypeFlags flags, + const TransactionOptions& options, + OptionalCommandControl command_control + ); void WriteStatistics(utils::statistics::Writer& writer) const; /// @brief Set default command control (timeouts) from dynamic config void SetDefaultCommandControl(const CommandControl& cc); + /// @brief Atomically replace command controls looked up by the current + /// task-inherited HTTP handler path and method. + /// + /// Each configured field overlays the lower-priority default independently. + /// Passing an empty map clears the complete handler layer. + void SetHandlersCommandControl(CommandControlByHandlerMap command_control); + + /// @brief Atomically replace command controls looked up by Query name. + /// + /// Each configured field overlays default and handler fields independently. + /// Unnamed queries skip this layer. Passing an empty map clears it. + void SetQueriesCommandControl(CommandControlByQueryMap command_control); + + /// @brief Set the per-pool bound for named query latency and error metrics. + /// + /// A zero bound disables accounting and clears all retained named query + /// names. Shrinking the bound evicts the least recently used names. Each + /// retained name exports three metric series. + void SetStatementMetricsSettings(const settings::StatementMetricsSettings& settings); + + /// @brief Set the per-connection prepared statement cache bound. + /// + /// A zero bound disables and clears the cache. Shrinking evicts the least + /// recently used statements; growing preserves existing entries. Existing + /// physical connections apply changes before their next operation. + void SetPreparedStatementCacheSettings(const settings::PreparedStatementCacheSettings& settings); + + /// @brief Atomically replace cluster pools for future operations. + /// Existing queries and transactions keep their old pools alive. + void UpdateSettings(const settings::ODBCClusterSettings& settings); + + /// @cond + void UpdateDsns(const std::vector& dsns); + void SetPoolSettingsOverride(std::optional settings); + void SetPreparedStatementCacheSettingsOverride(std::optional settings); + void ApplyDynamicCommandControls( + CommandControl default_command_control, + CommandControlByHandlerMap handlers_command_control, + CommandControlByQueryMap queries_command_control + ); + /// @endcond + /// @brief Get current default network timeout std::optional GetDefaultNetworkTimeout() const; @@ -56,6 +197,27 @@ class Cluster { std::optional GetDefaultStatementTimeout() const; private: + ResultSet DoExecute( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterList& parameters + ); + Cursor DoExecuteCursor( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterList& parameters + ); + BulkResult DoExecuteBulk( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterRows& rows, + const detail::BulkLayout& layout, + std::size_t chunk_rows + ); + detail::ClusterImplPtr impl_; }; diff --git a/odbc/include/userver/storages/odbc/command_control.hpp b/odbc/include/userver/storages/odbc/command_control.hpp new file mode 100644 index 000000000000..5444f609d660 --- /dev/null +++ b/odbc/include/userver/storages/odbc/command_control.hpp @@ -0,0 +1,43 @@ +#pragma once + +/// @file userver/storages/odbc/command_control.hpp +/// @brief Per-operation timeout settings for the ODBC driver. + +#include +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +struct CommandControl final { + /// Overall operation budget used for pool waiting and connection login and + /// as an upper bound for statement/transaction deadlines. Blocking ODBC + /// calls are ultimately subject to the timeout capabilities and whole- + /// second resolution of the selected ODBC driver. + std::optional network_timeout; + + /// Timeout for statement execution. ODBC drivers accept this timeout in + /// whole seconds, so the value is rounded up when passed to a driver. + std::optional statement_timeout; + + bool operator==(const CommandControl&) const = default; +}; + +using OptionalCommandControl = std::optional; + +/// Command controls keyed by an HTTP method. +using CommandControlByMethodMap = utils::impl::TransparentMap; + +/// Command controls keyed first by handler path, then by HTTP method. +using CommandControlByHandlerMap = utils::impl::TransparentMap; + +/// Command controls keyed by storages::odbc::Query name. +using CommandControlByQueryMap = utils::impl::TransparentMap; + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/component.hpp b/odbc/include/userver/storages/odbc/component.hpp index f86147a4508c..3dee88451d53 100644 --- a/odbc/include/userver/storages/odbc/component.hpp +++ b/odbc/include/userver/storages/odbc/component.hpp @@ -4,9 +4,14 @@ /// @brief @copybrief components::Odbc #include +#include +#include +#include #include #include +#include +#include USERVER_NAMESPACE_BEGIN @@ -16,7 +21,31 @@ class Cluster; namespace components { -/// @brief Component that owns a storages::odbc::Cluster +/// @ingroup userver_components +/// +/// @brief ODBC client component that owns a storages::odbc::Cluster. +/// +/// ## Dynamic options: +/// * @ref USERVER_ODBC_DEFAULT_COMMAND_CONTROL +/// * @ref USERVER_ODBC_HANDLERS_COMMAND_CONTROL +/// * @ref USERVER_ODBC_QUERIES_COMMAND_CONTROL +/// * @ref USERVER_ODBC_CONNECTION_POOL_SETTINGS +/// * @ref USERVER_ODBC_STATEMENT_METRICS_SETTINGS +/// * @ref USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS +/// +/// ## Static configuration example: +/// +/// @snippet odbc/functional_tests/basic_chaos/static_config.yaml ODBC component config +/// +/// Exactly one of `dsn`, `pools`, and `secdist_alias` must be specified. +/// With `secdist_alias`, connection data is loaded from components::Secdist and +/// is updated without changing the Cluster object returned by GetCluster(). +/// +/// ## Static options of components::Odbc: +/// @include{doc} scripts/docs/en/components_schema/odbc/src/storages/odbc/component.md +/// +/// Options inherited from @ref components::ComponentBase: +/// @include{doc} scripts/docs/en/components_schema/core/src/components/impl/component_base.md class Odbc final : public ComponentBase { public: static constexpr std::string_view kName = "odbc"; @@ -30,14 +59,23 @@ class Odbc final : public ComponentBase { private: void OnConfigUpdate(const dynamic_config::Snapshot& config); + void OnSecdistUpdate(const storages::secdist::SecdistConfig& secdist); std::string name_; + storages::odbc::settings::StatementMetricsSettings statement_metrics_settings_fallback_; + std::optional secdist_alias_; std::shared_ptr cluster_; dynamic_config::Source config_source_; + + // Subscriptions must be the last fields because callbacks use all fields above. concurrent::AsyncEventSubscriberScope config_subscription_; + concurrent::AsyncEventSubscriberScope secdist_subscription_; }; +template <> +inline constexpr bool kHasValidate = true; + } // namespace components USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/cursor.hpp b/odbc/include/userver/storages/odbc/cursor.hpp new file mode 100644 index 000000000000..a537a4b3ebe3 --- /dev/null +++ b/odbc/include/userver/storages/odbc/cursor.hpp @@ -0,0 +1,60 @@ +#pragma once + +/// @file userver/storages/odbc/cursor.hpp +/// @brief @copybrief storages::odbc::Cursor + +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +namespace detail { +class CursorImpl; +class ClusterImpl; +} // namespace detail + +/// @brief A move-only incremental ODBC result cursor. +/// +/// Each Fetch materializes at most the requested number of rows into an owning +/// ResultSet. The cursor pins its connection until EOF, destruction, or an +/// error. This bounds memory materialized by userver, but an ODBC driver may +/// still buffer rows internally and server-side streaming is not guaranteed. +/// Fetch calls on a cursor must be sequential. +class Cursor final { +public: + Cursor(const Cursor&) = delete; + Cursor& operator=(const Cursor&) = delete; + + Cursor(Cursor&&) noexcept; + Cursor& operator=(Cursor&&) noexcept; + ~Cursor() noexcept; + + /// @brief Fetch up to @p rows and return an owning result chunk. + /// @throws LogicError if rows is zero or the cursor is already terminal. + ResultSet Fetch(std::size_t rows); + + /// @brief Whether EOF, an error, invalidation, or a moved-from state has + /// made this cursor terminal. + bool Done() const noexcept; + + /// @brief Number of rows returned by successful Fetch calls. + std::size_t FetchedSoFar() const noexcept; + + explicit operator bool() const noexcept { return !Done(); } + +private: + friend class detail::ClusterImpl; + friend class Transaction; + + explicit Cursor(std::shared_ptr impl); + + std::shared_ptr impl_; +}; + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/exception.hpp b/odbc/include/userver/storages/odbc/exception.hpp index 20469b994b06..ff8bbaa93484 100644 --- a/odbc/include/userver/storages/odbc/exception.hpp +++ b/odbc/include/userver/storages/odbc/exception.hpp @@ -5,11 +5,21 @@ #include #include +#include +#include +#include USERVER_NAMESPACE_BEGIN namespace storages::odbc { +/// A single diagnostic record reported by the ODBC driver manager or driver. +struct DiagnosticRecord final { + std::string sql_state; + int native_error{0}; + std::string message; +}; + class Error : public std::runtime_error { using std::runtime_error::runtime_error; }; @@ -19,7 +29,23 @@ class LogicError : public Error { }; class RuntimeError : public Error { +public: using Error::Error; + + RuntimeError(std::string message, std::vector diagnostics, bool invalid_handle = false); + + /// Structured driver diagnostics, in the order returned by ODBC. + const std::vector& GetDiagnostics() const noexcept; + + /// Whether any diagnostic has the specified two-character SQLSTATE class. + bool HasSqlStateClass(std::string_view sql_state_class) const noexcept; + + /// Whether the failed ODBC call returned SQL_INVALID_HANDLE. + bool IsInvalidHandle() const noexcept; + +private: + std::vector diagnostics_; + bool invalid_handle_{false}; }; class ConnectionError : public RuntimeError { @@ -30,6 +56,11 @@ class StatementError : public RuntimeError { using RuntimeError::RuntimeError; }; +/// Thrown when an ODBC pool cannot provide a connection for a non-timeout reason. +class PoolError : public RuntimeError { + using RuntimeError::RuntimeError; +}; + /// Thrown when the operation is aborted because an @ref engine::Deadline has expired /// (including task-inherited request deadlines). class OperationInterrupted : public RuntimeError { diff --git a/odbc/include/userver/storages/odbc/field.hpp b/odbc/include/userver/storages/odbc/field.hpp index 1fc0999cac2e..8f2a4bda0ef2 100644 --- a/odbc/include/userver/storages/odbc/field.hpp +++ b/odbc/include/userver/storages/odbc/field.hpp @@ -3,14 +3,65 @@ /// @file userver/storages/odbc/field.hpp /// @brief @copybrief storages::odbc::Field +#include #include +#include +#include +#include +#include +#include +#include +#include +#include #include +#include USERVER_NAMESPACE_BEGIN namespace storages::odbc { +/// @cond +namespace impl { + +template +struct IsOptional : std::false_type {}; + +template +struct IsOptional> : std::true_type { + using ValueType = T; +}; + +template +inline constexpr bool kIsOptional = IsOptional>::value; + +template +inline constexpr bool kIsFieldScalar = + std::same_as, bool> || + (std::integral> && !std::same_as, bool> && + sizeof(std::remove_cv_t) <= sizeof(std::uint64_t)) || + std::same_as, float> || std::same_as, double> || + std::same_as, std::string> || std::same_as, Bytes> || + std::same_as, Date> || std::same_as, Time> || + std::same_as, Timestamp> || kIsDecimal>; + +template +struct IsFieldAsType + : std::bool_constant< + kIsFieldScalar || (!kIsFieldScalar && io::traits::kHasFromOdbc>)> {}; + +template +struct IsFieldAsType> + : std::bool_constant< + !io::traits::kHasMappingDeclaration> && + (kIsFieldScalar || (!kIsFieldScalar && io::traits::kHasFromOdbc))> {}; + +template +inline constexpr bool kIsFieldAsType = IsFieldAsType>::value; + +} // namespace impl +/// @endcond + /// @brief Single cell in an ODBC result set row class Field { public: @@ -27,6 +78,11 @@ class Field { double GetDouble() const; bool GetBool() const; + /// Converts the field with strict SQL category, NULL and range checks. + /// Use `As>()` to accept SQL NULL. + template + T As() const; + protected: friend class Row; @@ -39,11 +95,82 @@ class Field { {} private: + std::int64_t GetSignedIntegerForAs() const; + std::uint64_t GetUnsignedIntegerForAs() const; + double GetFloatingPointForAs() const; + std::string GetStringForAs() const; + bool GetBoolForAs() const; + Bytes GetBytesForAs() const; + Date GetDateForAs() const; + Time GetTimeForAs() const; + Timestamp GetTimestampForAs() const; + std::string GetDecimalForAs(std::size_t precision, std::size_t scale) const; + detail::ResultWrapperPtr res_; size_type row_index_{0}; size_type field_index_{0}; }; +template +T Field::As() const { + using Value = std::remove_cv_t; + static_assert( + !impl::kIsOptional || !io::traits::kHasMappingDeclaration, + "CppToOdbc> is not supported; map T and use std::optional for SQL NULL" + ); + static_assert(impl::kIsFieldAsType, "Unsupported ODBC Field::As() type"); + + if constexpr (impl::kIsOptional) { + using Inner = typename impl::IsOptional::ValueType; + if (IsNull()) { + return std::nullopt; + } + return Value{As()}; + } else if constexpr (!impl::kIsFieldScalar && io::traits::kHasFromOdbc) { + using Mapping = io::CppToOdbc; + using BoundType = typename Mapping::BoundType; + return Mapping::FromOdbc(As()); + } else if constexpr (std::same_as) { + return GetBoolForAs(); + } else if constexpr (std::signed_integral) { + const auto value = GetSignedIntegerForAs(); + if (value < static_cast(std::numeric_limits::lowest()) || + value > static_cast(std::numeric_limits::max())) + { + throw ResultSetError("ODBC integer field does not fit into the requested signed type"); + } + return static_cast(value); + } else if constexpr (std::unsigned_integral) { + const auto value = GetUnsignedIntegerForAs(); + if (value > static_cast(std::numeric_limits::max())) { + throw ResultSetError("ODBC integer field does not fit into the requested unsigned type"); + } + return static_cast(value); + } else if constexpr (std::same_as) { + const auto value = GetFloatingPointForAs(); + if (value < static_cast(std::numeric_limits::lowest()) || + value > static_cast(std::numeric_limits::max())) + { + throw ResultSetError("ODBC floating-point field does not fit into float"); + } + return static_cast(value); + } else if constexpr (std::same_as) { + return GetFloatingPointForAs(); + } else if constexpr (std::same_as) { + return GetStringForAs(); + } else if constexpr (std::same_as) { + return GetBytesForAs(); + } else if constexpr (std::same_as) { + return GetDateForAs(); + } else if constexpr (std::same_as) { + return GetTimeForAs(); + } else if constexpr (std::same_as) { + return GetTimestampForAs(); + } else if constexpr (impl::kIsDecimal) { + return Value{GetDecimalForAs(Value::kPrecision, Value::kScale)}; + } +} + } // namespace storages::odbc USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/impl/parameter.hpp b/odbc/include/userver/storages/odbc/impl/parameter.hpp new file mode 100644 index 000000000000..aaff1edf3356 --- /dev/null +++ b/odbc/include/userver/storages/odbc/impl/parameter.hpp @@ -0,0 +1,296 @@ +#pragma once + +/// @file userver/storages/odbc/impl/parameter.hpp +/// @brief Internal storage for ODBC query parameters. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc::impl { + +template +struct IsNativeParameterValue { +private: + using Value = std::remove_cvref_t; + using Pointee = std::remove_pointer_t; + using Element = std::remove_extent_t; + +public: + static constexpr bool value = + std::integral || std::floating_point || std::is_enum_v || + std::same_as || std::same_as || std::same_as || + std::same_as || std::same_as || std::same_as || kIsDecimal || + std::same_as || std::same_as || + (std::is_pointer_v && (std::same_as || std::same_as)) || + (std::is_array_v && std::same_as, char>); +}; + +template +inline constexpr bool kIsNativeParameterValue = IsNativeParameterValue::value; + +template +constexpr bool IsParameterScalarValue() { + using Value = std::remove_cvref_t; + if constexpr (io::traits::kIsOptional) { + if constexpr (io::traits::kHasMappingDeclaration) { + return false; + } else { + using Inner = typename io::traits::IsOptional::ValueType; + return !io::traits::kIsOptional && IsParameterScalarValue(); + } + } else if constexpr (kIsNativeParameterValue && !std::is_enum_v) { + // Native non-enum behavior cannot be shadowed by a user mapping. + return true; + } else if constexpr (io::traits::kHasMappingDeclaration) { + // A declared mapping suppresses enum and aggregate fallbacks even when + // the mapping is malformed or lacks the input direction. + return io::traits::kHasToOdbc; + } else { + return kIsNativeParameterValue; + } +} + +template +inline constexpr bool kIsParameterStoreValue = IsParameterScalarValue(); + +struct OdbcParameterMappingTag; + +template +constexpr bool AreParameterMembersMappable(std::index_sequence) { + return sizeof...(Index) != 0 && + ((!std::is_reference_v> && + kIsParameterStoreValue>) && + ...); +} + +template +constexpr bool DetectParameterAggregate() { + using Value = std::remove_cvref_t; + if constexpr (io::traits::kIsOptional || io::traits::kHasMappingDeclaration || + !std::is_class_v || !std::is_aggregate_v || !std::is_standard_layout_v || + std::is_union_v || !io::traits::kAggregateHasNoBaseClass || + !boost::pfr::is_implicitly_reflectable_v) + { + return false; + } else { + return AreParameterMembersMappable(std::make_index_sequence>{}); + } +} + +template +inline constexpr bool kIsParameterAggregate = DetectParameterAggregate(); + +template +inline constexpr bool kIsParameterArgument = kIsParameterStoreValue || kIsParameterAggregate; + +enum class ParameterType { + kBoolean, + kSignedInteger, + kUnsignedInteger, + kFloatingPoint, + kString, + kBytes, + kDate, + kTime, + kTimestamp, + kDecimal, + kUnknown, +}; + +struct DecimalParameter final { + std::string representation; + std::uint8_t precision; + std::uint8_t scale; +}; + +/// A type-erased, owning query parameter. Owning the value is important because +/// an ODBC driver is allowed to read bound buffers until SQLExecute returns. +class Parameter final { +public: + using Value = std::variant< + bool, + std::int64_t, + std::uint64_t, + double, + std::string, + Bytes, + Date, + Time, + Timestamp, + DecimalParameter>; + + Parameter(std::nullptr_t) + : type_{ParameterType::kUnknown}, + is_null_{true}, + value_{std::string{}} + {} + Parameter(std::nullopt_t) + : Parameter{nullptr} + {} + + Parameter(bool value) + : type_{ParameterType::kBoolean}, + value_{value} + {} + + template + requires(!std::same_as) + Parameter(T value) + : type_{ParameterType::kSignedInteger}, + value_{static_cast(value)} + {} + + template + requires(!std::same_as) + Parameter(T value) + : type_{ParameterType::kUnsignedInteger}, + value_{static_cast(value)} + {} + + template + Parameter(T value) + : type_{ParameterType::kFloatingPoint}, + value_{static_cast(value)} + {} + + template + requires(std::is_enum_v && !io::traits::kHasMappingDeclaration) + Parameter(T value) + : Parameter{static_cast>(value)} + {} + + template + requires(io::traits::kHasToOdbc && !io::traits::kIsDirectBoundType>) + Parameter(const T& value) + : Parameter{io::CppToOdbc>::ToOdbc(value)} + {} + + Parameter(const char* value) + : type_{ParameterType::kString}, + is_null_{value == nullptr}, + value_{value == nullptr ? std::string{} : std::string{value}} + {} + Parameter(std::string value) + : type_{ParameterType::kString}, + value_{std::move(value)} + {} + Parameter(std::string_view value) + : Parameter{std::string{value}} + {} + + Parameter(Bytes value) + : type_{ParameterType::kBytes}, + value_{std::move(value)} + {} + + Parameter(Date value) + : type_{ParameterType::kDate}, + value_{value} + {} + + Parameter(Time value) + : type_{ParameterType::kTime}, + value_{value} + {} + + Parameter(Timestamp value) + : type_{ParameterType::kTimestamp}, + value_{value} + {} + + template + Parameter(const Decimal& value) + : type_{ParameterType::kDecimal}, + value_{DecimalParameter{ + std::string{value.GetRepresentation()}, + static_cast(Precision), + static_cast(Scale), + }} + {} + + template + requires kIsParameterStoreValue> + Parameter(const std::optional& value) + : Parameter{value ? Parameter{*value} : NullOf()} + {} + + ParameterType GetType() const noexcept { return type_; } + bool IsNull() const noexcept { return is_null_; } + + template + const T& Get() const { + return std::get(value_); + } + +private: + template + static Parameter NullOf() { + using Value = std::remove_cv_t; + if constexpr (io::traits::kHasValidBoundType && + (!io::traits::kIsDirectBoundType || std::is_enum_v)) + { + return NullOf>(); + } + Parameter result{Value{}}; + result.is_null_ = true; + return result; + } + + ParameterType type_; + bool is_null_{false}; + Value value_; +}; + +using ParameterList = std::vector; +using ParameterRows = std::vector; + +template +requires kIsParameterArgument +constexpr std::size_t ParameterArgumentWidth() { + using Value = std::remove_cvref_t; + if constexpr (kIsParameterStoreValue) { + return 1; + } else { + return boost::pfr::tuple_size_v; + } +} + +template +requires kIsParameterArgument +void AppendParameterArgument(ParameterList& result, const T& argument) { + using Value = std::remove_cvref_t; + if constexpr (kIsParameterStoreValue) { + result.emplace_back(argument); + } else { + boost::pfr::for_each_field(argument, [&result](const auto& field) { result.emplace_back(field); }); + } +} + +template +requires((kIsParameterArgument && ...)) +ParameterList MakeParameterList(const Args&... args) { + ParameterList result; + result.reserve((ParameterArgumentWidth() + ... + std::size_t{0})); + (AppendParameterArgument(result, args), ...); + return result; +} + +} // namespace storages::odbc::impl + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/io/type_mapping.hpp b/odbc/include/userver/storages/odbc/io/type_mapping.hpp new file mode 100644 index 000000000000..7bd1e0829d4a --- /dev/null +++ b/odbc/include/userver/storages/odbc/io/type_mapping.hpp @@ -0,0 +1,141 @@ +#pragma once + +/// @file userver/storages/odbc/io/type_mapping.hpp +/// @brief Explicit C++ conversions to and from ODBC bound scalar types. + +#include +#include +#include +#include +#include +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc::io { + +/// @brief Customization point for mapping a user type to one ODBC scalar. +/// +/// Specializations declare a cv-unqualified, non-reference `BoundType` and +/// may independently provide `static BoundType ToOdbc(const T&)` for query +/// parameters and `static T FromOdbc(BoundType)` for result fields. +template +struct CppToOdbc; + +/// @cond +namespace traits { + +template +struct IsOptional : std::false_type {}; + +template +struct IsOptional> : std::true_type { + using ValueType = T; +}; + +template +inline constexpr bool kIsOptional = IsOptional>::value; + +template +inline constexpr bool kIsDirectBoundType = + std::same_as> && !kIsOptional && + (std::same_as || (std::integral && !std::same_as && sizeof(T) <= sizeof(std::uint64_t)) || + std::same_as || std::same_as || std::same_as || std::same_as || + std::same_as || std::same_as || std::same_as || + storages::odbc::impl::kIsDecimal); + +template +inline constexpr bool kHasMappingDeclaration = requires { sizeof(CppToOdbc>); }; + +template +constexpr bool HasValidBoundType() { + using Value = std::remove_cvref_t; + if constexpr (!kHasMappingDeclaration || kIsOptional) { + return false; + } else if constexpr (requires { typename CppToOdbc::BoundType; }) { + return kIsDirectBoundType::BoundType>; + } else { + return false; + } +} + +template +inline constexpr bool kHasValidBoundType = HasValidBoundType>(); + +template +constexpr bool HasToOdbc() { + using Value = std::remove_cvref_t; + if constexpr (!kHasValidBoundType) { + return false; + } else { + using Mapping = CppToOdbc; + using BoundType = typename Mapping::BoundType; + return requires { static_cast(&Mapping::ToOdbc); }; + } +} + +template +inline constexpr bool kHasToOdbc = HasToOdbc>(); + +template +constexpr bool HasFromOdbc() { + using Value = std::remove_cvref_t; + if constexpr (!kHasValidBoundType) { + return false; + } else { + using Mapping = CppToOdbc; + using BoundType = typename Mapping::BoundType; + return requires { static_cast(&Mapping::FromOdbc); }; + } +} + +template +inline constexpr bool kHasFromOdbc = HasFromOdbc>(); + +template +using BoundType = typename CppToOdbc>::BoundType; + +template +struct NonBaseInitializer final { + template + requires(!std::is_base_of_v, Derived>) + operator Type() const noexcept { // NOLINT(google-explicit-constructor) + std::abort(); + } +}; + +template +constexpr bool IsNonBaseAggregateInitializable(std::index_sequence) { + return requires { Value{(static_cast(Index), NonBaseInitializer{})...}; }; +} + +// Matches the maximum aggregate arity supported by Boost.PFR's generated core17 implementation. +inline constexpr std::size_t kPfrMaxFields = 200; + +template +constexpr bool HasNonBaseAggregateArity(std::index_sequence) { + return (IsNonBaseAggregateInitializable(std::make_index_sequence{}) || ...); +} + +template +constexpr bool AggregateHasNoBaseClass() { + using Value = std::remove_cvref_t; + if constexpr (!std::is_aggregate_v || std::is_empty_v) { + return false; + } else { + return HasNonBaseAggregateArity(std::make_index_sequence{}); + } +} + +template +inline constexpr bool kAggregateHasNoBaseClass = AggregateHasNoBaseClass(); + +} // namespace traits +/// @endcond + +} // namespace storages::odbc::io + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/odbc_fwd.hpp b/odbc/include/userver/storages/odbc/odbc_fwd.hpp index f283aadd878e..9af2803cc8f9 100644 --- a/odbc/include/userver/storages/odbc/odbc_fwd.hpp +++ b/odbc/include/userver/storages/odbc/odbc_fwd.hpp @@ -11,8 +11,10 @@ namespace storages::odbc { class ResultSet; class Row; +class Cursor; class Cluster; +class Transaction; /// @brief Smart pointer to the storages::odbc::Cluster using ClusterPtr = std::shared_ptr; diff --git a/odbc/include/userver/storages/odbc/parameter_store.hpp b/odbc/include/userver/storages/odbc/parameter_store.hpp new file mode 100644 index 000000000000..1c68421f89be --- /dev/null +++ b/odbc/include/userver/storages/odbc/parameter_store.hpp @@ -0,0 +1,79 @@ +#pragma once + +/// @file userver/storages/odbc/parameter_store.hpp +/// @brief @copybrief storages::odbc::ParameterStore + +#include +#include +#include +#include +#include +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +class Cluster; +class Transaction; +class BulkParameterStore; + +/// @ingroup userver_containers +/// +/// @brief Owning, ordered list of dynamically assembled ODBC parameters. +/// +/// Values are copied into the store and remain valid independently of the +/// source objects. Use an empty `std::optional` for SQL NULL: `T` determines +/// the parameter type used for ODBC binding. Raw `nullptr` and `std::nullopt` +/// remain untyped, just like in the variadic API, and should only be used when +/// the driver can infer the type from the statement. A null `const char*` is a +/// typed string NULL. +/// +/// @warning Parameters are always values for existing `?` placeholders. Never +/// interpolate them into the SQL query text. +class ParameterStore final { +public: + ParameterStore() = default; + ParameterStore(const ParameterStore&) = delete; + ParameterStore(ParameterStore&&) noexcept = default; + ParameterStore& operator=(const ParameterStore&) = delete; + ParameterStore& operator=(ParameterStore&&) noexcept = default; + + /// @brief Copies a scalar parameter or the declaration-order fields of a + /// supported aggregate to the end of the ordered list. + /// @returns `*this` for chained construction. + template + requires impl::kIsParameterArgument + ParameterStore& PushBack(const T& parameter) { + auto appended = impl::MakeParameterList(parameter); + static_assert(std::is_nothrow_move_constructible_v); + if (appended.size() > parameters_.max_size() - parameters_.size()) { + throw std::length_error("ODBC ParameterStore size exceeds its maximum"); + } + parameters_.reserve(parameters_.size() + appended.size()); + std::move(appended.begin(), appended.end(), std::back_inserter(parameters_)); + return *this; + } + + /// Returns whether the parameter list is empty. + bool IsEmpty() const noexcept { return parameters_.empty(); } + + /// Returns the number of stored parameters. + std::size_t Size() const noexcept { return parameters_.size(); } + +private: + friend class Cluster; + friend class Transaction; + friend class BulkParameterStore; + + const impl::ParameterList& GetParameters() const noexcept { return parameters_; } + + impl::ParameterList parameters_; +}; + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/result_set.hpp b/odbc/include/userver/storages/odbc/result_set.hpp index 3365d615b0b5..c8a70785aa71 100644 --- a/odbc/include/userver/storages/odbc/result_set.hpp +++ b/odbc/include/userver/storages/odbc/result_set.hpp @@ -3,9 +3,20 @@ /// @file userver/storages/odbc/result_set.hpp /// @brief @copybrief storages::odbc::ResultSet +#include +#include #include #include +#include +#include +#include +#include +#include +#include +#include + +#include #include #include @@ -13,6 +24,50 @@ USERVER_NAMESPACE_BEGIN namespace storages::odbc { +/// @cond +namespace impl { + +struct OdbcResultMappingTag; + +template +constexpr bool AreResultMembersMappable(std::index_sequence) { + return sizeof...(Index) != 0 && + ((!std::is_reference_v> && + kIsFieldAsType>>) && + ...); +} + +template +constexpr bool DetectResultAggregate() { + using Value = std::remove_cv_t; + if constexpr (std::is_class_v && std::is_aggregate_v && std::is_standard_layout_v && + !std::is_union_v && io::traits::kAggregateHasNoBaseClass && + boost::pfr::is_implicitly_reflectable_v && !kIsFieldAsType && + !io::traits::kHasMappingDeclaration) + { + return AreResultMembersMappable(std::make_index_sequence>{}); + } else { + return false; + } +} + +template +inline constexpr bool kIsResultAggregate = DetectResultAggregate(); + +template +inline constexpr bool kIsResultValue = kIsFieldAsType> || kIsResultAggregate; + +template +concept ResultContainer = + std::default_initializable && !std::same_as, std::string> && + requires { typename Container::value_type; } && kIsResultValue && + requires(Container& container, typename Container::value_type value) { + container.insert(container.end(), std::move(value)); + }; + +} // namespace impl +/// @endcond + /// @brief Result set for ODBC query execution class ResultSet final { public: @@ -36,15 +91,112 @@ class ResultSet final { size_type Size() const; + /// @brief Number of rows affected by a data-modifying statement. + /// Returns zero when the driver reports an unknown count. + size_type RowsAffected() const; + + /// @brief Get a result column name by zero-based index. + std::string_view GetFieldName(size_type index) const; + /// @brief Check if the result set is empty bool IsEmpty() const; reference operator[](size_type index) const&; + /// Materializes every row into the container's value type. Scalar values + /// require exactly one result column; aggregate values are initialized in + /// declaration order and require an exact column count. + template + requires impl::ResultContainer + Container AsContainer() const; + + /// Materializes the only result row, requiring exactly one row. + template + requires impl::kIsResultValue + T AsSingleRow() const; + + /// Returns no value for zero rows, materializes one row, and rejects more + /// than one row. For optional-valued T the outer optional represents row + /// presence and the inner optional represents SQL NULL. + template + requires impl::kIsResultValue + std::optional AsOptionalSingleRow() const; + private: + template + T MapAggregate(size_type row_index, std::index_sequence) const; + + template + T MapRow(size_type row_index) const; + std::shared_ptr pimpl_; }; +template +T ResultSet::MapAggregate(size_type row_index, std::index_sequence) const { + return T{ + operator[](row_index)[Index] + .template As(std::declval()))>>()... + }; +} + +template +T ResultSet::MapRow(size_type row_index) const { + using Value = std::remove_cv_t; + static_assert(impl::kIsResultValue, "Unsupported ODBC typed result value"); + + if constexpr (impl::kIsFieldAsType) { + if (FieldCount() != 1) { + throw ResultSetError("ODBC scalar result mapping requires exactly one column"); + } + return operator[](row_index)[0].template As(); + } else { + constexpr auto kFieldCount = boost::pfr::tuple_size_v; + if (FieldCount() != kFieldCount) { + throw ResultSetError("ODBC aggregate result mapping requires exactly one column per aggregate member"); + } + return MapAggregate(row_index, std::make_index_sequence{}); + } +} + +template +requires impl::ResultContainer +Container ResultSet::AsContainer() const { + using Value = typename Container::value_type; + static_assert(impl::kIsResultValue, "Unsupported ODBC typed result container value"); + + Container result; + if constexpr (requires { result.reserve(Size()); }) { + result.reserve(Size()); + } + auto output = std::inserter(result, result.end()); + for (size_type index = 0; index < Size(); ++index) { + *output++ = MapRow(index); + } + return result; +} + +template +requires impl::kIsResultValue +T ResultSet::AsSingleRow() const { + if (Size() != 1) { + throw ResultSetError("ODBC single-row result mapping requires exactly one row"); + } + return MapRow(0); +} + +template +requires impl::kIsResultValue +std::optional ResultSet::AsOptionalSingleRow() const { + if (Size() > 1) { + throw ResultSetError("ODBC optional single-row result mapping accepts at most one row"); + } + if (IsEmpty()) { + return std::nullopt; + } + return std::optional{MapRow(0)}; +} + } // namespace storages::odbc USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/settings.hpp b/odbc/include/userver/storages/odbc/settings.hpp index e408f6b830c9..fd08739c53f8 100644 --- a/odbc/include/userver/storages/odbc/settings.hpp +++ b/odbc/include/userver/storages/odbc/settings.hpp @@ -3,6 +3,7 @@ /// @file userver/storages/odbc/settings.hpp /// @brief ODBC cluster static configuration (DSN pools) +#include #include #include @@ -10,18 +11,43 @@ USERVER_NAMESPACE_BEGIN namespace storages::odbc::settings { +/// @brief Named ODBC query metrics options. +struct StatementMetricsSettings final { + /// Maximum number of named query-name entries retained by each ODBC pool. + /// Each entry exports three metric series. A value of 0 disables named + /// query metrics. + std::size_t max_statements{0}; + + bool operator==(const StatementMetricsSettings&) const = default; +}; + +/// @brief Per-connection prepared statement cache options. +struct PreparedStatementCacheSettings final { + /// Maximum number of parameterized SQL statements retained per physical + /// ODBC connection. A value of 0 disables the cache. + std::size_t max_size{0}; + + bool operator==(const PreparedStatementCacheSettings&) const = default; +}; + struct PoolSettings final { std::size_t min_size{5}; std::size_t max_size{10}; + + bool operator==(const PoolSettings&) const = default; }; struct HostSettings final { const std::string dsn; PoolSettings pool; + + bool operator==(const HostSettings&) const = default; }; struct ODBCClusterSettings final { std::vector pools; + + bool operator==(const ODBCClusterSettings&) const = default; }; } // namespace storages::odbc::settings diff --git a/odbc/include/userver/storages/odbc/transaction.hpp b/odbc/include/userver/storages/odbc/transaction.hpp index 4d7eec58bf4c..0f0855598ab5 100644 --- a/odbc/include/userver/storages/odbc/transaction.hpp +++ b/odbc/include/userver/storages/odbc/transaction.hpp @@ -10,14 +10,21 @@ #include #include +#include +#include +#include +#include +#include #include #include +#include USERVER_NAMESPACE_BEGIN namespace storages::odbc { namespace detail { +struct BulkLayout; class ConnectionPtr; class Pool; } // namespace detail @@ -29,12 +36,78 @@ class Pool; /// storages::odbc::Cluster class Transaction final { public: - explicit Transaction(detail::ConnectionPtr&& connection, detail::Pool& pool, engine::Deadline deadline); + explicit Transaction( + detail::ConnectionPtr&& connection, + detail::Pool& pool, + std::chrono::milliseconds network_timeout, + std::chrono::milliseconds statement_timeout + ); + explicit Transaction( + detail::ConnectionPtr&& connection, + detail::Pool& pool, + const TransactionOptions& options, + std::chrono::milliseconds network_timeout, + std::chrono::milliseconds statement_timeout + ); ~Transaction(); Transaction(const Transaction& other) = delete; Transaction(Transaction&& other) noexcept; - ResultSet Execute(const Query& query); + /// @brief Execute a statement, binding every argument to an ODBC `?` placeholder. + template + requires((impl::kIsParameterArgument && ...)) + ResultSet Execute(const Query& query, const Args&... args) { + return Execute(std::nullopt, query, args...); + } + + /// @brief Execute a statement with per-statement timeout overrides. + template + requires((impl::kIsParameterArgument && ...)) + ResultSet Execute(OptionalCommandControl command_control, const Query& query, const Args&... args) { + return DoExecute(command_control, query, impl::MakeParameterList(args...)); + } + + /// @brief Execute a statement with an owning dynamic parameter list. + ResultSet Execute(const Query& query, const ParameterStore& store); + + /// @brief Execute a statement with a dynamic parameter list and timeout overrides. + ResultSet Execute(OptionalCommandControl command_control, const Query& query, const ParameterStore& store); + + /// @brief Execute a row-producing statement as an incremental cursor. + /// + /// No other transaction operation is allowed until the cursor observes EOF + /// or is destroyed. + template + requires((impl::kIsParameterArgument && ...)) + Cursor ExecuteCursor(const Query& query, const Args&... args) { + return ExecuteCursor(std::nullopt, query, args...); + } + + template + requires((impl::kIsParameterArgument && ...)) + Cursor ExecuteCursor(OptionalCommandControl command_control, const Query& query, const Args&... args) { + return DoExecuteCursor(command_control, query, impl::MakeParameterList(args...)); + } + + Cursor ExecuteCursor(const Query& query, const ParameterStore& store); + + Cursor ExecuteCursor(OptionalCommandControl command_control, const Query& query, const ParameterStore& store); + + /// Execute rows as bounded chunks of DML that must not return result sets. + /// On failure, inspect BulkExecutionError and roll back the transaction. + BulkResult ExecuteBulk( + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows = kDefaultBulkRows + ); + + /// Execute bulk DML with per-operation timeout overrides. + BulkResult ExecuteBulk( + OptionalCommandControl command_control, + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows = kDefaultBulkRows + ); /// @brief Commit the transaction void Commit(); @@ -43,12 +116,30 @@ class Transaction final { void Rollback(); private: + ResultSet DoExecute( + OptionalCommandControl command_control, + const Query& query, + const impl::ParameterList& parameters + ); + Cursor DoExecuteCursor( + OptionalCommandControl command_control, + const Query& query, + const impl::ParameterList& parameters + ); + BulkResult DoExecuteBulk( + OptionalCommandControl command_control, + const Query& query, + const impl::ParameterRows& rows, + const detail::BulkLayout& layout, + std::size_t chunk_rows + ); void AssertValid() const; // shared_ptr(16) + unique_ptr(8) = 24 bytes, align 8 utils::FastPimpl connection_; detail::Pool* pool_; - engine::Deadline deadline_; + std::chrono::milliseconds network_timeout_; + std::chrono::milliseconds statement_timeout_; utils::datetime::SteadyCoarseClock::time_point start_time_; std::chrono::microseconds busy_time_{0}; tracing::Span span_; diff --git a/odbc/include/userver/storages/odbc/transaction_options.hpp b/odbc/include/userver/storages/odbc/transaction_options.hpp new file mode 100644 index 000000000000..5ae4c66cda78 --- /dev/null +++ b/odbc/include/userver/storages/odbc/transaction_options.hpp @@ -0,0 +1,63 @@ +#pragma once + +/// @file userver/storages/odbc/transaction_options.hpp +/// @brief ODBC transaction options + +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +/// Portable ODBC transaction isolation levels. +enum class IsolationLevel : std::uint8_t { + kReadUncommitted, + kReadCommitted, + kRepeatableRead, + kSerializable, +}; + +/// ODBC transaction access-mode hint. +/// +/// @warning `kReadOnly` requests `SQL_MODE_READ_ONLY` from the ODBC driver, but +/// ODBC defines this as an intent/optimization hint. It does not guarantee that +/// the database rejects write statements. +enum class AccessMode : std::uint8_t { + kReadOnly, + kReadWrite, +}; + +/// Options for starting an ODBC transaction. +/// +/// Empty optionals preserve the physical connection's current driver defaults; +/// the driver does not silently force READ COMMITTED or READ WRITE. +struct TransactionOptions final { + std::optional isolation_level; + std::optional access_mode; + + // Explicit keeps the legacy Cluster::Begin(flags, {}) call unambiguous: an + // empty braced argument continues to mean OptionalCommandControl. + constexpr explicit TransactionOptions() = default; + + constexpr explicit TransactionOptions(IsolationLevel isolation) + : isolation_level{isolation} + {} + + constexpr explicit TransactionOptions(AccessMode mode) + : access_mode{mode} + {} + + constexpr TransactionOptions(IsolationLevel isolation, AccessMode mode) + : isolation_level{isolation}, + access_mode{mode} + {} +}; + +constexpr bool operator==(const TransactionOptions& lhs, const TransactionOptions& rhs) noexcept { + return lhs.isolation_level == rhs.isolation_level && lhs.access_mode == rhs.access_mode; +} + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/include/userver/storages/odbc/types.hpp b/odbc/include/userver/storages/odbc/types.hpp new file mode 100644 index 000000000000..3831a06768d0 --- /dev/null +++ b/odbc/include/userver/storages/odbc/types.hpp @@ -0,0 +1,251 @@ +#pragma once + +/// @file userver/storages/odbc/types.hpp +/// @brief Portable value types for standard ODBC SQL types. + +#include +#include +#include +#include +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +/// Owning byte sequence for SQL BINARY, VARBINARY and LONGVARBINARY. +class Bytes final { +public: + using ValueType = std::uint8_t; + using Container = std::vector; + + Bytes() = default; + explicit Bytes(Container bytes); + Bytes(std::initializer_list bytes); + + const Container& GetBytes() const noexcept; + std::size_t Size() const noexcept; + bool IsEmpty() const noexcept; + + bool operator==(const Bytes&) const noexcept = default; + +private: + Container bytes_; +}; + +/// Timezone-independent Gregorian calendar date in the portable 1..9999 range. +class Date final { +public: + Date() noexcept; + Date(std::uint32_t year, std::uint32_t month, std::uint32_t day); + + std::uint32_t GetYear() const noexcept; + std::uint32_t GetMonth() const noexcept; + std::uint32_t GetDay() const noexcept; + std::string ToString() const; + + bool operator==(const Date&) const noexcept = default; + +private: + std::uint16_t year_{1970}; + std::uint8_t month_{1}; + std::uint8_t day_{1}; +}; + +/// Timezone-independent time of day with the portable `SQL_TIME_STRUCT` +/// resolution of one second. +class Time final { +public: + Time() noexcept = default; + Time(std::uint32_t hour, std::uint32_t minute, std::uint32_t second); + + std::uint32_t GetHour() const noexcept; + std::uint32_t GetMinute() const noexcept; + std::uint32_t GetSecond() const noexcept; + std::string ToString() const; + + bool operator==(const Time&) const noexcept = default; + +private: + std::uint8_t hour_{0}; + std::uint8_t minute_{0}; + std::uint8_t second_{0}; +}; + +/// Timezone-independent timestamp with nanosecond fraction storage. +/// +/// No implicit conversion to or from `std::chrono::system_clock::time_point` +/// is provided because an ODBC TIMESTAMP has no timezone. +class Timestamp final { +public: + Timestamp() noexcept = default; + Timestamp(Date date, Time time, std::uint32_t fraction_nanoseconds = 0); + Timestamp( + std::uint32_t year, + std::uint32_t month, + std::uint32_t day, + std::uint32_t hour, + std::uint32_t minute, + std::uint32_t second, + std::uint32_t fraction_nanoseconds = 0 + ); + + const Date& GetDate() const noexcept; + const Time& GetTime() const noexcept; + std::uint32_t GetFractionNanoseconds() const noexcept; + std::string ToString() const; + + bool operator==(const Timestamp&) const noexcept = default; + +private: + Date date_; + Time time_; + std::uint32_t fraction_nanoseconds_{0}; +}; + +/// Exact fixed-point SQL DECIMAL/NUMERIC value. +/// +/// Accepted syntax is `[-+]digits` for Scale=0 and +/// `[-+]digits.Scale-digits` otherwise. Exponents, whitespace, NaN and +/// infinities are rejected. Values are canonicalized by removing a leading +/// plus and redundant integer zeroes; negative zero is normalized to positive +/// zero. Exactly Scale fractional digits, including trailing zeroes, are +/// retained. ODBC SQL_NUMERIC_STRUCT limits portable precision to 38 digits. +template +class Decimal final { + static_assert(Precision >= 1 && Precision <= 38, "ODBC Decimal precision must be in the range 1..38"); + static_assert(Scale <= Precision, "ODBC Decimal scale must not exceed precision"); + +public: + static constexpr std::size_t kPrecision = Precision; + static constexpr std::size_t kScale = Scale; + + Decimal() + : representation_{MakeZero()} + {} + + explicit Decimal(std::string_view representation) + : representation_{Validate(representation)} + {} + + std::string_view GetRepresentation() const noexcept; + static constexpr std::size_t GetPrecision() noexcept { return Precision; } + static constexpr std::size_t GetScale() noexcept { return Scale; } + + bool operator==(const Decimal&) const noexcept = default; + +private: + static std::string MakeZero(); + static std::string Validate(std::string_view representation); + + std::string representation_; +}; + +/// @cond +namespace impl { + +template +struct IsDecimal : std::false_type {}; + +template +struct IsDecimal> : std::true_type {}; + +template +inline constexpr bool kIsDecimal = IsDecimal>::value; + +} // namespace impl +/// @endcond + +template +std::string_view Decimal::GetRepresentation() const noexcept { + return representation_; +} + +template +std::string Decimal::MakeZero() { + if constexpr (Scale == 0) { + return "0"; + } else { + return std::string{"0."} + std::string(Scale, '0'); + } +} + +template +std::string Decimal::Validate(std::string_view representation) { + if (representation.empty()) { + throw std::invalid_argument("ODBC Decimal representation must not be empty"); + } + + std::size_t index = representation.front() == '-' || representation.front() == '+' ? 1 : 0; + const auto integer_begin = index; + while (index < representation.size() && representation[index] >= '0' && representation[index] <= '9') { + ++index; + } + if (index == integer_begin) { + throw std::invalid_argument("ODBC Decimal requires at least one integer digit"); + } + const auto integer_end = index; + + if constexpr (Scale == 0) { + if (index != representation.size()) { + throw std::invalid_argument("ODBC Decimal with scale 0 must not contain a fractional part"); + } + } else { + if (index == representation.size() || representation[index] != '.') { + throw std::invalid_argument("ODBC Decimal representation does not contain its declared scale"); + } + ++index; + const auto fractional_begin = index; + while (index < representation.size() && representation[index] >= '0' && representation[index] <= '9') { + ++index; + } + if (index != representation.size() || index - fractional_begin != Scale) { + throw std::invalid_argument("ODBC Decimal fractional digits do not match its declared scale"); + } + } + + auto first_significant = integer_begin; + while (first_significant < integer_end && representation[first_significant] == '0') { + ++first_significant; + } + const auto significant_integer_digits = first_significant == integer_end ? 0 : integer_end - first_significant; + if (significant_integer_digits > Precision - Scale) { + throw std::out_of_range("ODBC Decimal magnitude exceeds its declared precision and scale"); + } + + const bool fractional_is_zero = [&] { + if constexpr (Scale == 0) { + return true; + } else { + for (std::size_t position = integer_end + 1; position < representation.size(); ++position) { + if (representation[position] != '0') { + return false; + } + } + return true; + } + }(); + const bool is_zero = significant_integer_digits == 0 && fractional_is_zero; + + std::string result; + if (!is_zero && representation.front() == '-') { + result.push_back('-'); + } + if (first_significant == integer_end) { + result.push_back('0'); + } else { + result.append(representation.substr(first_significant, significant_integer_digits)); + } + if constexpr (Scale != 0) { + result.push_back('.'); + result.append(representation.substr(integer_end + 1, Scale)); + } + return result; +} + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/library.yaml b/odbc/library.yaml index f522bb78badd..99f502f9fcf7 100644 --- a/odbc/library.yaml +++ b/odbc/library.yaml @@ -12,4 +12,8 @@ libraries: configs: names: - USERVER_ODBC_DEFAULT_COMMAND_CONTROL + - USERVER_ODBC_HANDLERS_COMMAND_CONTROL + - USERVER_ODBC_QUERIES_COMMAND_CONTROL - USERVER_ODBC_CONNECTION_POOL_SETTINGS + - USERVER_ODBC_STATEMENT_METRICS_SETTINGS + - USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS diff --git a/odbc/src/storages/odbc/bulk.cpp b/odbc/src/storages/odbc/bulk.cpp new file mode 100644 index 000000000000..71951f3edca1 --- /dev/null +++ b/odbc/src/storages/odbc/bulk.cpp @@ -0,0 +1,68 @@ +#include + +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +BulkParameterStore& BulkParameterStore::PushBackRow(ParameterStore&& row) { + AppendRow(std::move(row.parameters_)); + return *this; +} + +void BulkParameterStore::AppendRow(impl::ParameterList row) { + if (row.empty()) { + throw LogicError("ODBC bulk parameter rows must contain at least one column"); + } + if (!rows_.empty() && row.size() != columns_count_) { + throw LogicError(fmt::format("ODBC bulk parameter row has {} columns, expected {}", row.size(), columns_count_) + ); + } + if (rows_.empty()) { + columns_count_ = row.size(); + } + rows_.push_back(std::move(row)); +} + +BulkResult::BulkResult( + std::size_t requested, + std::optional processed, + std::optional rows_affected, + std::vector statuses +) + : requested_{requested}, + processed_{processed}, + rows_affected_{rows_affected}, + statuses_{std::move(statuses)} +{ + if (statuses_.size() != requested_) { + throw LogicError( + fmt::format("ODBC bulk result contains {} statuses for {} requested rows", statuses_.size(), requested_) + ); + } + if (processed_ && *processed_ > requested_) { + throw LogicError( + fmt::format("ODBC bulk result reports {} processed rows for {} requested rows", *processed_, requested_) + ); + } + succeeded_ = static_cast(std::count_if(statuses_.begin(), statuses_.end(), [](BulkRowStatus status) { + return status == BulkRowStatus::kSuccess || status == BulkRowStatus::kSuccessWithInfo; + })); +} + +BulkExecutionError::BulkExecutionError( + std::string message, + std::vector diagnostics, + BulkResult result, + bool invalid_handle +) + : StatementError{std::move(message), std::move(diagnostics), invalid_handle}, + result_{std::move(result)} +{} + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/src/storages/odbc/cluster.cpp b/odbc/src/storages/odbc/cluster.cpp index cf39a92042e5..818189c1c1e9 100644 --- a/odbc/src/storages/odbc/cluster.cpp +++ b/odbc/src/storages/odbc/cluster.cpp @@ -1,8 +1,10 @@ #include +#include #include -#include +#include +#include #include USERVER_NAMESPACE_BEGIN @@ -10,29 +12,165 @@ USERVER_NAMESPACE_BEGIN namespace storages::odbc { Cluster::Cluster(const settings::ODBCClusterSettings& settings, clients::dns::Resolver* resolver) - : impl_(std::make_unique(settings, resolver)) + : Cluster{settings, resolver, engine::current_task::GetBlockingTaskProcessor()} +{} + +Cluster::Cluster( + const settings::ODBCClusterSettings& settings, + clients::dns::Resolver* resolver, + engine::TaskProcessor& blocking_task_processor +) + : impl_(std::make_unique(settings, resolver, blocking_task_processor)) { UASSERT(!settings.pools.empty()); } Cluster::~Cluster() = default; -ResultSet Cluster::Execute(ClusterHostTypeFlags flags, const Query& query) { return impl_->Execute(flags, query); } +ResultSet Cluster::DoExecute( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterList& parameters +) { + return impl_->Execute(flags, command_control, query, parameters); +} + +ResultSet Cluster::Execute(ClusterHostTypeFlags flags, const Query& query, const ParameterStore& store) { + return Execute(flags, std::nullopt, query, store); +} + +ResultSet Cluster::Execute( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const ParameterStore& store +) { + return DoExecute(command_control, flags, query, store.GetParameters()); +} + +Cursor Cluster::DoExecuteCursor( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterList& parameters +) { + return impl_->ExecuteCursor(flags, command_control, query, parameters); +} + +Cursor Cluster::ExecuteCursor(ClusterHostTypeFlags flags, const Query& query, const ParameterStore& store) { + return ExecuteCursor(flags, std::nullopt, query, store); +} -ResultSet Cluster::Execute(engine::Deadline deadline, ClusterHostTypeFlags flags, const Query& query) { - return impl_->Execute(deadline, flags, query); +Cursor Cluster::ExecuteCursor( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const ParameterStore& store +) { + return DoExecuteCursor(command_control, flags, query, store.GetParameters()); +} + +BulkResult Cluster::ExecuteBulk( + ClusterHostTypeFlags flags, + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows +) { + return ExecuteBulk(flags, std::nullopt, query, rows, chunk_rows); +} + +BulkResult Cluster::ExecuteBulk( + ClusterHostTypeFlags flags, + OptionalCommandControl command_control, + const Query& query, + const BulkParameterStore& rows, + std::size_t chunk_rows +) { + if (chunk_rows == 0) { + throw LogicError("ODBC bulk chunk size must be greater than zero"); + } + if (rows.IsEmpty()) { + return {}; + } + const auto layout = detail::ValidateBulkRows(rows.GetRows()); + return DoExecuteBulk(command_control, flags, query, rows.GetRows(), layout, chunk_rows); +} + +BulkResult Cluster::DoExecuteBulk( + OptionalCommandControl command_control, + ClusterHostTypeFlags flags, + const Query& query, + const impl::ParameterRows& rows, + const detail::BulkLayout& layout, + std::size_t chunk_rows +) { + return impl_->ExecuteBulk(flags, command_control, query, rows, layout, chunk_rows); } Transaction Cluster::Begin(ClusterHostTypeFlags flags) { return impl_->Begin(flags); } -Transaction Cluster::Begin(engine::Deadline deadline, ClusterHostTypeFlags flags) { - return impl_->Begin(deadline, flags); +Transaction Cluster::Begin(ClusterHostTypeFlags flags, OptionalCommandControl command_control) { + return impl_->Begin(flags, command_control); +} + +Transaction Cluster::Begin(ClusterHostTypeFlags flags, const TransactionOptions& options) { + return impl_->Begin(flags, options); +} + +Transaction Cluster::Begin( + ClusterHostTypeFlags flags, + const TransactionOptions& options, + OptionalCommandControl command_control +) { + return impl_->Begin(flags, options, command_control); } void Cluster::WriteStatistics(utils::statistics::Writer& writer) const { impl_->WriteStatistics(writer); } void Cluster::SetDefaultCommandControl(const CommandControl& cc) { impl_->SetDefaultCommandControl(cc); } +void Cluster::SetHandlersCommandControl(CommandControlByHandlerMap command_control) { + impl_->SetHandlersCommandControl(std::move(command_control)); +} + +void Cluster::SetQueriesCommandControl(CommandControlByQueryMap command_control) { + impl_->SetQueriesCommandControl(std::move(command_control)); +} + +void Cluster::SetStatementMetricsSettings(const settings::StatementMetricsSettings& settings) { + impl_->SetStatementMetricsSettings(settings); +} + +void Cluster::SetPreparedStatementCacheSettings(const settings::PreparedStatementCacheSettings& settings) { + impl_->SetPreparedStatementCacheSettings(settings); +} + +void Cluster::SetPreparedStatementCacheSettingsOverride(std::optional settings +) { + impl_->SetPreparedStatementCacheSettingsOverride(settings); +} + +void Cluster::UpdateSettings(const settings::ODBCClusterSettings& settings) { impl_->UpdateSettings(settings); } + +void Cluster::UpdateDsns(const std::vector& dsns) { impl_->UpdateDsns(dsns); } + +void Cluster::SetPoolSettingsOverride(std::optional settings) { + impl_->SetPoolSettingsOverride(settings); +} + +void Cluster::ApplyDynamicCommandControls( + CommandControl default_command_control, + CommandControlByHandlerMap handlers_command_control, + CommandControlByQueryMap queries_command_control +) { + impl_->ApplyDynamicCommandControls( + std::move(default_command_control), + std::move(handlers_command_control), + std::move(queries_command_control) + ); +} + std::optional Cluster::GetDefaultNetworkTimeout() const { return impl_->GetDefaultNetworkTimeout(); } diff --git a/odbc/src/storages/odbc/component.cpp b/odbc/src/storages/odbc/component.cpp index 2dff078310d9..ea936d4f1913 100644 --- a/odbc/src/storages/odbc/component.cpp +++ b/odbc/src/storages/odbc/component.cpp @@ -1,12 +1,16 @@ #include #include +#include +#include +#include #include #include #include #include #include +#include #include #include #include @@ -16,11 +20,16 @@ #include #include -#include "odbc_config.hpp" #include "odbc_secdist.hpp" +#include + #include #include +#include +#include +#include +#include #ifndef ARCADIA_ROOT #include "generated/src/storages/odbc/component.yaml.hpp" // Y_IGNORE @@ -32,16 +41,37 @@ namespace components { namespace { +void ValidatePoolSettings(const storages::odbc::settings::PoolSettings& settings) { + UINVARIANT(settings.max_size > 0, "ODBC max_pool_size must be positive"); + UINVARIANT(settings.min_size <= settings.max_size, "ODBC min_pool_size must not exceed max_pool_size"); +} + +void ValidateNonEmpty(std::string_view value, std::string_view option) { + if (value.empty()) { + throw std::runtime_error("ODBC component option '" + std::string{option} + "' must not be empty"); + } +} + +engine::TaskProcessor& GetBlockingTaskProcessor( + const components::ComponentConfig& config, + const components::ComponentContext& context +) { + const auto name = config["blocking_task_processor"].As>(); + return name ? context.GetTaskProcessor(*name) : engine::current_task::GetBlockingTaskProcessor(); +} + storages::odbc::settings::ODBCClusterSettings MakeClusterSettingsFromConfig(const components::ComponentConfig& config) { using storages::odbc::settings::HostSettings; using storages::odbc::settings::ODBCClusterSettings; using storages::odbc::settings::PoolSettings; if (const auto dsn_opt = config["dsn"].As>(); dsn_opt.has_value()) { + ValidateNonEmpty(*dsn_opt, "dsn"); const auto min_size = config["min_pool_size"].As(PoolSettings{}.min_size); const auto max_size = config["max_pool_size"].As(PoolSettings{}.max_size); + ValidatePoolSettings(PoolSettings{.min_size = min_size, .max_size = max_size}); return ODBCClusterSettings{std::vector{ - HostSettings{*dsn_opt, PoolSettings{min_size, max_size}}, + HostSettings{.dsn = *dsn_opt, .pool = {.min_size = min_size, .max_size = max_size}}, }}; } @@ -52,9 +82,14 @@ storages::odbc::settings::ODBCClusterSettings MakeClusterSettingsFromConfig(cons for (std::size_t i = 0; i < pools_cfg.GetSize(); ++i) { const auto pool = pools_cfg[i]; const auto dsn = pool["dsn"].As(); + ValidateNonEmpty(dsn, "pools[" + std::to_string(i) + "].dsn"); const auto min_size = pool["min_pool_size"].As(PoolSettings{}.min_size); const auto max_size = pool["max_pool_size"].As(PoolSettings{}.max_size); - pools.emplace_back(HostSettings{dsn, PoolSettings{min_size, max_size}}); + ValidatePoolSettings(PoolSettings{.min_size = min_size, .max_size = max_size}); + pools.emplace_back(HostSettings{ + .dsn = dsn, + .pool = {.min_size = min_size, .max_size = max_size}, + }); } return ODBCClusterSettings{std::move(pools)}; } @@ -75,11 +110,15 @@ storages::odbc::settings::ODBCClusterSettings MakeClusterSettingsFromSecdist( const auto min_size = config["min_pool_size"].As(PoolSettings{}.min_size); const auto max_size = config["max_pool_size"].As(PoolSettings{}.max_size); + ValidatePoolSettings(PoolSettings{.min_size = min_size, .max_size = max_size}); std::vector pools; pools.reserve(connection_infos.size()); for (const auto& info : connection_infos) { - pools.emplace_back(HostSettings{info.dsn, PoolSettings{min_size, max_size}}); + pools.emplace_back(HostSettings{ + .dsn = info.dsn, + .pool = {.min_size = min_size, .max_size = max_size}, + }); } return ODBCClusterSettings{std::move(pools)}; @@ -90,16 +129,59 @@ storages::odbc::settings::ODBCClusterSettings MakeClusterSettings( const components::ComponentContext& context ) { const auto secdist_alias = config["secdist_alias"].As>(); + const auto dsn = config["dsn"].As>(); + const auto pools = config["pools"]; + const auto has_pools = !pools.IsMissing() && pools.GetSize() > 0; + + const auto connection_sources = + static_cast(secdist_alias.has_value()) + static_cast(dsn.has_value()) + + static_cast(has_pools); + UINVARIANT( + connection_sources == 1, + "Exactly one ODBC connection source must be configured: 'dsn', 'pools', or 'secdist_alias'" + ); if (secdist_alias.has_value()) { + ValidateNonEmpty(*secdist_alias, "secdist_alias"); const auto& secdist = context.FindComponent(); const auto& odbc_settings = secdist.Get().Get(); return MakeClusterSettingsFromSecdist(odbc_settings, *secdist_alias, config); } - auto settings = MakeClusterSettingsFromConfig(config); - UINVARIANT(!settings.pools.empty(), "Either 'dsn', 'pools', or 'secdist_alias' must be set"); - return settings; + return MakeClusterSettingsFromConfig(config); +} + +template +storages::odbc::CommandControl ConvertCommandControl(const ConfigCommandControl& command_control) { + return { + .network_timeout = command_control.network_timeout_ms, + .statement_timeout = command_control.statement_timeout_ms, + }; +} + +template +storages::odbc::CommandControlByQueryMap ConvertQueriesCommandControl(const ConfigMap& config) { + storages::odbc::CommandControlByQueryMap result; + result.reserve(config.extra.size()); + for (const auto& [name, command_control] : config.extra) { + result.emplace(name, ConvertCommandControl(command_control)); + } + return result; +} + +template +storages::odbc::CommandControlByHandlerMap ConvertHandlersCommandControl(const ConfigMap& config) { + storages::odbc::CommandControlByHandlerMap result; + result.reserve(config.extra.size()); + for (const auto& [path, config_by_method] : config.extra) { + storages::odbc::CommandControlByMethodMap by_method; + by_method.reserve(config_by_method.extra.size()); + for (const auto& [method, command_control] : config_by_method.extra) { + by_method.emplace(method, ConvertCommandControl(command_control)); + } + result.emplace(path, std::move(by_method)); + } + return result; } } // namespace @@ -107,11 +189,22 @@ storages::odbc::settings::ODBCClusterSettings MakeClusterSettings( Odbc::Odbc(const ComponentConfig& config, const ComponentContext& context) : ComponentBase{config, context}, name_{config.Name()}, - cluster_{std::make_shared< - storages::odbc::Cluster>(MakeClusterSettings(config, context), clients::dns::GetResolverPtr(config, context)) + statement_metrics_settings_fallback_{ + .max_statements = config["max_statement_metrics"].As(0), }, + secdist_alias_{config["secdist_alias"].As>()}, + cluster_{std::make_shared( + MakeClusterSettings(config, context), + clients::dns::GetResolverPtr(config, context), + GetBlockingTaskProcessor(config, context) + )}, config_source_{context.FindComponent().GetSource()} { + cluster_->SetStatementMetricsSettings(statement_metrics_settings_fallback_); + cluster_->SetPreparedStatementCacheSettings({ + .max_size = config["max_prepared_cache_size"].As(0), + }); + utils::statistics::RegisterWriterScope( context, "odbc", @@ -125,29 +218,76 @@ Odbc::Odbc(const ComponentConfig& config, const ComponentContext& context) "odbc", &Odbc::OnConfigUpdate, ::dynamic_config::USERVER_ODBC_CONNECTION_POOL_SETTINGS, - ::dynamic_config::USERVER_ODBC_DEFAULT_COMMAND_CONTROL + ::dynamic_config::USERVER_ODBC_DEFAULT_COMMAND_CONTROL, + ::dynamic_config::USERVER_ODBC_HANDLERS_COMMAND_CONTROL, + ::dynamic_config::USERVER_ODBC_QUERIES_COMMAND_CONTROL, + ::dynamic_config::USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS, + ::dynamic_config::USERVER_ODBC_STATEMENT_METRICS_SETTINGS ); + + if (secdist_alias_) { + auto& secdist = context.FindComponent(); + secdist_subscription_ = secdist.GetStorage().UpdateAndListen(this, name_, &Odbc::OnSecdistUpdate); + } } -Odbc::~Odbc() { config_subscription_.Unsubscribe(); } +Odbc::~Odbc() { + config_subscription_.Unsubscribe(); + secdist_subscription_.Unsubscribe(); +} void Odbc::OnConfigUpdate(const dynamic_config::Snapshot& config) { const auto& pool_settings = config[::dynamic_config::USERVER_ODBC_CONNECTION_POOL_SETTINGS]; // Apply default command control from dynamic config const auto pool_settings_opt = pool_settings.GetOptional(name_); + std::optional updated; if (pool_settings_opt.has_value()) { - // Note: Pool size changes require restart as ConnectionPoolBase - // doesn't support dynamic resizing. Log a warning if settings differ. - // In future versions, this could be enhanced to support dynamic resizing. + updated = storages::odbc::settings::PoolSettings{ + .min_size = pool_settings_opt->min_pool_size, + .max_size = pool_settings_opt->max_pool_size, + }; + ValidatePoolSettings(*updated); } + cluster_->SetPoolSettingsOverride(updated); - // Apply command control (timeouts) - const auto cc = config[::dynamic_config::USERVER_ODBC_DEFAULT_COMMAND_CONTROL]; - cluster_->SetDefaultCommandControl(storages::odbc::CommandControl{ - .network_timeout = cc.network_timeout_ms, - .statement_timeout = cc.statement_timeout_ms, - }); + const auto& statement_metrics = config[::dynamic_config::USERVER_ODBC_STATEMENT_METRICS_SETTINGS]; + auto statement_metrics_settings = statement_metrics_settings_fallback_; + if (const auto dynamic_settings = statement_metrics.GetOptional(name_)) { + statement_metrics_settings.max_statements = dynamic_settings->max_statement_metrics; + } + cluster_->SetStatementMetricsSettings(statement_metrics_settings); + + const auto& prepared_statement_cache = config[::dynamic_config::USERVER_ODBC_PREPARED_STATEMENT_CACHE_SETTINGS]; + std::optional prepared_statement_cache_override; + if (const auto dynamic_settings = prepared_statement_cache.GetOptional(name_)) { + prepared_statement_cache_override = { + .max_size = dynamic_settings->max_prepared_cache_size, + }; + } + cluster_->SetPreparedStatementCacheSettingsOverride(prepared_statement_cache_override); + + const auto& default_command_control = config[::dynamic_config::USERVER_ODBC_DEFAULT_COMMAND_CONTROL]; + const auto& handlers_command_control = config[::dynamic_config::USERVER_ODBC_HANDLERS_COMMAND_CONTROL]; + const auto& queries_command_control = config[::dynamic_config::USERVER_ODBC_QUERIES_COMMAND_CONTROL]; + cluster_->ApplyDynamicCommandControls( + ConvertCommandControl(default_command_control), + ConvertHandlersCommandControl(handlers_command_control), + ConvertQueriesCommandControl(queries_command_control) + ); +} + +void Odbc::OnSecdistUpdate(const storages::secdist::SecdistConfig& secdist) { + UASSERT(secdist_alias_); + const auto& odbc_settings = secdist.Get(); + const auto connection_infos = odbc_settings.GetConnectionInfos(*secdist_alias_); + + std::vector dsns; + dsns.reserve(connection_infos.size()); + for (const auto& info : connection_infos) { + dsns.push_back(info.dsn); + } + cluster_->UpdateDsns(dsns); } std::shared_ptr Odbc::GetCluster() const { return cluster_; } diff --git a/odbc/src/storages/odbc/component.yaml b/odbc/src/storages/odbc/component.yaml index 4aa03947941a..1d81700db8fc 100644 --- a/odbc/src/storages/odbc/component.yaml +++ b/odbc/src/storages/odbc/component.yaml @@ -2,26 +2,52 @@ type: object description: ODBC client component additionalProperties: false properties: + blocking_task_processor: + type: string + description: | + Task processor used for synchronous ODBC driver-manager and driver calls. + Defaults to the global blocking task processor. secdist_alias: type: string description: | name of the database in secdist config. + Must not be empty. If specified, DSN will be read from secdist instead of static config. + Mutually exclusive with `dsn` and `pools`. dsn: type: string - description: connection DSN string (used for single-pool configuration) + description: | + connection DSN string (used for single-pool configuration). + Must not be empty. + Mutually exclusive with `secdist_alias` and `pools`. min_pool_size: type: integer + minimum: 0 description: | number of connections created initially by this component instance. Connections are kept even without requests default: 1 max_pool_size: type: integer + minimum: 1 description: | maximum number of connections that can be created by this component instance. Should not be less than `min_pool_size` default: 10 + max_statement_metrics: + type: integer + minimum: 0 + description: | + maximum number of named query-name entries retained per ODBC pool. + Each entry exports three metric series. Zero disables named query metrics + default: 0 + max_prepared_cache_size: + type: integer + minimum: 0 + description: | + maximum number of prepared parameterized SQL statements retained by each + physical ODBC connection. Zero disables the prepared statement cache + default: 0 dns_resolver: type: string description: server hostname resolver type (getaddrinfo or async) @@ -31,7 +57,10 @@ properties: - async pools: type: array - description: list of connection pools (used for multi-pool configuration) + minItems: 1 + description: | + list of connection pools (used for multi-pool configuration). + Mutually exclusive with `secdist_alias` and `dsn`. items: type: object description: connection pool configuration @@ -39,12 +68,16 @@ properties: properties: dsn: type: string - description: connection DSN string for this pool + description: non-empty connection DSN string for this pool min_pool_size: type: integer + minimum: 0 description: minimum number of connections in this pool default: 1 max_pool_size: type: integer + minimum: 1 description: maximum number of connections in this pool default: 10 + required: + - dsn diff --git a/odbc/src/storages/odbc/cursor.cpp b/odbc/src/storages/odbc/cursor.cpp new file mode 100644 index 000000000000..f1cffd65bb94 --- /dev/null +++ b/odbc/src/storages/odbc/cursor.cpp @@ -0,0 +1,33 @@ +#include + +#include + +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc { + +Cursor::Cursor(std::shared_ptr impl) + : impl_{std::move(impl)} +{} + +Cursor::Cursor(Cursor&&) noexcept = default; +Cursor& Cursor::operator=(Cursor&&) noexcept = default; +Cursor::~Cursor() noexcept = default; + +ResultSet Cursor::Fetch(std::size_t rows) { + if (!impl_) { + throw LogicError("ODBC cursor is moved-from"); + } + return impl_->Fetch(rows); +} + +bool Cursor::Done() const noexcept { return !impl_ || impl_->Done(); } + +std::size_t Cursor::FetchedSoFar() const noexcept { return impl_ ? impl_->FetchedSoFar() : 0; } + +} // namespace storages::odbc + +USERVER_NAMESPACE_END diff --git a/odbc/src/storages/odbc/detail/broken_guard.cpp b/odbc/src/storages/odbc/detail/broken_guard.cpp index 32d637561fd5..a53f04e28d2a 100644 --- a/odbc/src/storages/odbc/detail/broken_guard.cpp +++ b/odbc/src/storages/odbc/detail/broken_guard.cpp @@ -10,7 +10,7 @@ BrokenGuard::BrokenGuard(Connection& connection) : connection_{connection}, exceptions_on_enter_{std::uncaught_exceptions()} { - if (connection_.IsBroken()) { + if (connection_.IsMarkedBroken()) { throw ConnectionError("Connection is broken."); } } diff --git a/odbc/src/storages/odbc/detail/bulk.cpp b/odbc/src/storages/odbc/detail/bulk.cpp new file mode 100644 index 000000000000..7cf8fcf8ad24 --- /dev/null +++ b/odbc/src/storages/odbc/detail/bulk.cpp @@ -0,0 +1,605 @@ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace storages::odbc::detail { + +namespace { + +BulkColumnType NormalizeType(impl::ParameterType type, std::size_t row, std::size_t column) { + using impl::ParameterType; + switch (type) { + case ParameterType::kBoolean: + return BulkColumnType::kBoolean; + case ParameterType::kSignedInteger: + case ParameterType::kUnsignedInteger: + return BulkColumnType::kInteger; + case ParameterType::kFloatingPoint: + return BulkColumnType::kFloatingPoint; + case ParameterType::kString: + return BulkColumnType::kString; + case ParameterType::kBytes: + return BulkColumnType::kBytes; + case ParameterType::kDate: + return BulkColumnType::kDate; + case ParameterType::kTime: + return BulkColumnType::kTime; + case ParameterType::kTimestamp: + return BulkColumnType::kTimestamp; + case ParameterType::kDecimal: + return BulkColumnType::kDecimal; + case ParameterType::kUnknown: + throw LogicError(fmt::format( + "ODBC bulk parameter at row {}, column {} is an untyped NULL; use a typed std::optional", + row, + column + )); + } + throw LogicError("Unknown ODBC bulk parameter type"); +} + +std::size_t PayloadSize(const impl::Parameter& parameter) { + switch (parameter.GetType()) { + case impl::ParameterType::kString: + return parameter.Get().size(); + case impl::ParameterType::kBytes: + return parameter.Get().GetBytes().size(); + default: + return 0; + } +} + +void ValidateRepresentablePayload(std::size_t size, std::size_t row, std::size_t column) { + if (size > static_cast(std::numeric_limits::max())) { + throw LogicError(fmt::format( + "ODBC bulk parameter at row {}, column {} has {} bytes, which does not fit SQLLEN", + row, + column, + size + )); + } +} + +std::size_t CheckedMultiply(std::size_t left, std::size_t right, std::string_view what) { + if (left != 0 && right > std::numeric_limits::max() / left) { + throw LogicError(fmt::format("ODBC bulk {} size overflows size_t", what)); + } + return left * right; +} + +std::size_t CheckedAdd(std::size_t left, std::size_t right, std::string_view what) { + if (right > std::numeric_limits::max() - left) { + throw LogicError(fmt::format("ODBC bulk {} size overflows size_t", what)); + } + return left + right; +} + +template +void ValidateVectorCount(std::size_t count, std::string_view what) { + if (count > std::vector{}.max_size()) { + throw LogicError(fmt::format("ODBC bulk {} cannot be represented by a vector", what)); + } +} + +SQL_NUMERIC_STRUCT MakeNumericStruct(const impl::DecimalParameter& parameter) { + SQL_NUMERIC_STRUCT result{}; + result.precision = parameter.precision; + result.scale = static_cast(parameter.scale); + result.sign = parameter.representation.front() == '-' ? 0 : 1; + + for (const char ch : parameter.representation) { + if (ch == '-' || ch == '+' || ch == '.') { + continue; + } + unsigned carry = static_cast(ch - '0'); + for (auto& byte : result.val) { + const auto value = static_cast(byte) * 10U + carry; + byte = static_cast(value & 0xffU); + carry = value >> 8U; + } + if (carry != 0) { + throw LogicError("ODBC bulk Decimal magnitude exceeds SQL_NUMERIC_STRUCT capacity"); + } + } + return result; +} + +std::size_t FixedValueSize(BulkColumnType type) { + switch (type) { + case BulkColumnType::kBoolean: + return sizeof(SQLCHAR); + case BulkColumnType::kInteger: + return sizeof(SQLBIGINT); + case BulkColumnType::kFloatingPoint: + return sizeof(SQLDOUBLE); + case BulkColumnType::kDate: + return sizeof(SQL_DATE_STRUCT); + case BulkColumnType::kTime: + return sizeof(SQL_TIME_STRUCT); + case BulkColumnType::kTimestamp: + return sizeof(SQL_TIMESTAMP_STRUCT); + case BulkColumnType::kDecimal: + return sizeof(SQL_NUMERIC_STRUCT); + case BulkColumnType::kString: + case BulkColumnType::kBytes: + return 0; + } + throw LogicError("Unknown ODBC bulk column type"); +} + +std::size_t ChunkPayloadStride( + const impl::ParameterRows& rows, + std::size_t begin, + std::size_t count, + std::size_t column +) { + std::size_t stride = 1; + for (std::size_t row = begin; row < begin + count; ++row) { + stride = std::max(stride, PayloadSize(rows[row][column])); + } + return stride; +} + +BulkColumnBinding MakeBinding(BulkColumnDescription description, std::size_t count, std::size_t stride) { + BulkColumnBinding binding; + binding.indicators.resize(count); + switch (description.type) { + case BulkColumnType::kBoolean: + binding.c_type = SQL_C_BIT; + binding.sql_type = SQL_BIT; + binding.column_size = 1; + binding.buffer_size = sizeof(SQLCHAR); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kInteger: + binding.c_type = SQL_C_SBIGINT; + binding.sql_type = SQL_BIGINT; + binding.column_size = 19; + binding.buffer_size = sizeof(SQLBIGINT); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kFloatingPoint: + binding.c_type = SQL_C_DOUBLE; + binding.sql_type = SQL_DOUBLE; + binding.column_size = 15; + binding.buffer_size = sizeof(SQLDOUBLE); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kString: + binding.c_type = SQL_C_CHAR; + binding.sql_type = SQL_VARCHAR; + binding.column_size = static_cast(stride); + binding.buffer_size = static_cast(stride); + binding.values = + BulkInlineValues{std::vector(CheckedMultiply(count, stride, "string buffer")), stride}; + break; + case BulkColumnType::kBytes: + binding.c_type = SQL_C_BINARY; + binding.sql_type = SQL_LONGVARBINARY; + binding.column_size = static_cast(stride); + binding.buffer_size = static_cast(stride); + binding + .values = BulkInlineValues{std::vector(CheckedMultiply(count, stride, "byte buffer")), stride}; + break; + case BulkColumnType::kDate: + binding.c_type = SQL_C_TYPE_DATE; + binding.sql_type = SQL_TYPE_DATE; + binding.column_size = 10; + binding.buffer_size = sizeof(SQL_DATE_STRUCT); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kTime: + binding.c_type = SQL_C_TYPE_TIME; + binding.sql_type = SQL_TYPE_TIME; + binding.column_size = 8; + binding.buffer_size = sizeof(SQL_TIME_STRUCT); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kTimestamp: + binding.c_type = SQL_C_TYPE_TIMESTAMP; + binding.sql_type = SQL_TYPE_TIMESTAMP; + binding.column_size = description.timestamp_has_fraction ? 29 : 19; + binding.decimal_digits = description.timestamp_has_fraction ? 9 : 0; + binding.buffer_size = sizeof(SQL_TIMESTAMP_STRUCT); + binding.values = BulkValues{std::vector(count)}; + break; + case BulkColumnType::kDecimal: + binding.c_type = SQL_C_NUMERIC; + binding.sql_type = SQL_DECIMAL; + binding.column_size = description.decimal_precision; + binding.decimal_digits = description.decimal_scale; + binding.buffer_size = 0; + binding.values = BulkValues{std::vector(count)}; + break; + } + return binding; +} + +void StoreValue(BulkColumnBinding& binding, const impl::Parameter& parameter, std::size_t row) { + if (parameter.IsNull()) { + binding.indicators[row] = SQL_NULL_DATA; + return; + } + + using impl::ParameterType; + switch (parameter.GetType()) { + case ParameterType::kBoolean: + std::get>(binding.values).values[row] = parameter.Get() ? 1 : 0; + break; + case ParameterType::kSignedInteger: + std::get>(binding.values).values[row] = parameter.Get(); + break; + case ParameterType::kUnsignedInteger: + std::get>(binding.values) + .values[row] = static_cast(parameter.Get()); + break; + case ParameterType::kFloatingPoint: + std::get>(binding.values).values[row] = parameter.Get(); + break; + case ParameterType::kString: { + const auto& value = parameter.Get(); + auto& target = std::get(binding.values); + std::memcpy(target.values.data() + row * target.stride, value.data(), value.size()); + binding.indicators[row] = static_cast(value.size()); + return; + } + case ParameterType::kBytes: { + const auto& value = parameter.Get().GetBytes(); + auto& target = std::get(binding.values); + std::memcpy(target.values.data() + row * target.stride, value.data(), value.size()); + binding.indicators[row] = static_cast(value.size()); + return; + } + case ParameterType::kDate: { + const auto& value = parameter.Get(); + std::get>(binding.values).values[row] = SQL_DATE_STRUCT{ + static_cast(value.GetYear()), + static_cast(value.GetMonth()), + static_cast(value.GetDay()), + }; + break; + } + case ParameterType::kTime: { + const auto& value = parameter.Get