diff --git a/API_REFERENCE.md b/API_REFERENCE.md index 70b660b..2045e92 100644 --- a/API_REFERENCE.md +++ b/API_REFERENCE.md @@ -91,6 +91,7 @@ Read data from an SAP table or CDS view. Supports projection pushdown, filter pu | `THREADS` | UINTEGER | 0 | Number of parallel read threads | | `COLUMNS` | LIST(VARCHAR) | all | Columns to retrieve | | `FILTER` | VARCHAR | — | SAP WHERE clause filter | +| `fetch_size` | UINTEGER | `erpl_rfc_fetch_size` | Concurrent result rows per round-trip; transport only, never changes the rows returned | | `MAX_ROWS` | UINTEGER | 0 (all) | Maximum rows to return | | `READ_TABLE_FUNCTION` | VARCHAR | `'RFC_READ_TABLE'` | RFC function to use (see note) | | `READ_TABLE_DELIMITER` | VARCHAR | — | Delimiter for TABLE2 variants | @@ -1280,6 +1281,8 @@ Notes: | `erpl_rfc_persistent_connections` | BOOLEAN | `true` | Cache one RFC connection + function descriptor per column for a `sap_read_table` scan instead of reopening per batch | | `erpl_rfc_max_persistent_connections` | UINTEGER | 16 | Upper bound on RFC connections a scan caches concurrently (issue #67); columns past the cap use per-batch open/close | | `erpl_rfc_read_table_batch_budget` | UINTEGER | 1310720 | Target max concurrent result rows (projected columns × per-column batch) for `sap_read_table`; bounds peak memory on wide tables (issue #69). Lower = less memory but more RFC round-trips; `0` disables the cap | +| `erpl_rfc_fetch_size` | UINTEGER | 1310720 | How much `sap_read_table` asks SAP for per round-trip, as concurrent result rows. Alias of `erpl_rfc_read_table_batch_budget`. See [Tuning large reads](#tuning-large-reads) | +| `erpl_rfc_max_threads` | UINTEGER | 0 | Default for the `threads` named parameter; `0` lets erpl choose | | `erpl_rfc_pushdown_filters` | BOOLEAN | `true` | Translate SQL `WHERE` predicates into `RFC_READ_TABLE`'s `OPTIONS` table so SAP filters the rows instead of sending them all. Turning it off never changes which rows come back, only how many cross the wire. See [Filter Pushdown](#filter-pushdown) | | `erpl_rfc_backend` | VARCHAR | `'nwrfc'` | Which implementation serves RFC calls: `'nwrfc'` (SAP's NetWeaver RFC SDK) or `'proto'` (the pure-Rust erpl-proto implementation). Must be set **before the first SAP call**; frozen for the life of the process once resolved. Environment override: `ERPL_RFC_BACKEND` | | `erpl_rfc_backend_path` | VARCHAR | `''` | Explicit path to the RFC backend shared library, overriding the search. Empty means: next to the extension, then the loader's library path. Environment override: `ERPL_RFC_BACKEND_PATH` | @@ -1403,14 +1406,42 @@ releases that reject the generated syntax, and as a way to check that a filter i not the cause of a discrepancy: the same query must return the same rows with it on and off. -### Parallel Reads +### Tuning large reads -Use the `THREADS` parameter on `sap_read_table` and `sap_odp_read_full` for large tables: +Paging and parallelism use the **same two names everywhere**, so what you learn on +one extension applies to the others: + +| | Named parameter | Session setting | Meaning | +|---|---|---|---| +| Parallelism | `threads` | `erpl__max_threads` | How many SAP calls run at once. `0` = let erpl decide | +| Fetch granule | `fetch_size` | `erpl__fetch_size` | How much is asked for per round-trip, in the protocol's own unit | ```sql -SELECT * FROM sap_read_table('LARGE_TABLE', THREADS=8); +-- per query +SELECT * FROM sap_read_table('LARGE_TABLE', threads = 8, fetch_size = 262144); + +-- or as a session default +SET erpl_rfc_fetch_size = 262144; +SET erpl_rfc_max_threads = 8; ``` +The unit of `fetch_size` follows the protocol: for `sap_read_table` it is +**concurrent result rows** (projected columns x per-column batch), which is what +bounds the SAP SDK's own buffer on wide tables. + +**Both are transport settings only.** Every value returns exactly the same rows; +they trade memory against round-trips, nothing else. Lower `fetch_size` to cap +memory harder on a wide table, raise it for fewer round-trips on a narrow one. + +How much `threads` helps depends on the SAP system's capacity — work processes, +application servers, database sessions — not on erpl. Raise it while watching +throughput rather than setting it blindly, and check with your Basis team before +running a large parallel extract against a production system. + +`erpl_rfc_read_table_batch_budget` is the original spelling of +`erpl_rfc_fetch_size`. Both remain supported and write the same value — they are two +names for one knob, not two knobs. + ### SSH Tunnel + SAP Connection Complete workflow for connecting through an SSH jump host: diff --git a/rfc/src/erpl_rfc_extension.cpp b/rfc/src/erpl_rfc_extension.cpp index ad8eb25..b235e04 100644 --- a/rfc/src/erpl_rfc_extension.cpp +++ b/rfc/src/erpl_rfc_extension.cpp @@ -161,6 +161,10 @@ namespace duckdb { SetRfcPushdownFilters(parameter.GetValue()); } + static void OnRfcMaxThreads(ClientContext &, SetScope, Value ¶meter) { + SetRfcMaxThreads(parameter.GetValue()); + } + static void OnRfcBackend(ClientContext &, SetScope, Value ¶meter) { SetRfcBackend(parameter.GetValue()); } @@ -266,6 +270,33 @@ namespace duckdb { Value::UINTEGER(RfcReadColumnStateMachine::DEFAULT_READ_TABLE_BATCH_BUDGET), OnReadTableBatchBudget); + // The shared tuning vocabulary: `erpl__fetch_size` and + // `erpl__max_threads` mean the same thing in every extension, so what you + // learn on one applies to the others. erpl_rfc_read_table_batch_budget above is + // the original name and keeps working -- both write the same value, so setting + // either takes effect; they are two spellings of one knob, not two knobs. + config.AddExtensionOption( + "erpl_rfc_fetch_size", + "How much sap_read_table asks SAP for at a time, as a target upper bound on " + "concurrent result rows (projected columns x per-column batch). Lower it to " + "cap memory harder at the cost of more RFC round-trips; raise it for fewer " + "round-trips on narrow tables. 0 disables the cap. Override per query with " + "the fetch_size named parameter. Alias of erpl_rfc_read_table_batch_budget.", + LogicalType::UINTEGER, + Value::UINTEGER(RfcReadColumnStateMachine::DEFAULT_READ_TABLE_BATCH_BUDGET), + OnReadTableBatchBudget); + + config.AddExtensionOption( + "erpl_rfc_max_threads", + "Default number of concurrent RFC calls a sap_read_table scan may make. 0 " + "(the default) lets erpl choose. Override per query with the threads named " + "parameter. How much parallelism helps depends on the SAP system's own " + "capacity -- work processes, application servers and database sessions -- so " + "raise it while watching throughput rather than setting it blindly.", + LogicalType::UINTEGER, + Value::UINTEGER(0), + OnRfcMaxThreads); + config.AddExtensionOption( "erpl_rfc_pushdown_filters", "Translate SQL WHERE predicates into RFC_READ_TABLE's OPTIONS table so SAP " diff --git a/rfc/src/include/sap_rfc.hpp b/rfc/src/include/sap_rfc.hpp index 3b467fc..0a04baa 100644 --- a/rfc/src/include/sap_rfc.hpp +++ b/rfc/src/include/sap_rfc.hpp @@ -50,6 +50,9 @@ namespace duckdb void SetRfcPushdownFilters(bool enabled); bool GetRfcPushdownFilters(); + void SetRfcMaxThreads(unsigned int n); + unsigned int GetRfcMaxThreads(); + class RfcReadTableBindData : public TableFunctionData { public: @@ -153,6 +156,7 @@ namespace duckdb // Filters that could not be translated into OPTIONS, keyed by projected // column index. Copied because the TableFilterSet belongs to the plan. std::vector>> residual_filters; + unsigned int fetch_size_override = 0; // Defaults to RfcReadColumnStateMachine::MAX_BATCH_SIZE (that class // is defined later in this header, so we can't name the constant // here); Step() overwrites this with the column-count-aware cap. @@ -186,6 +190,13 @@ namespace duckdb // plan, so anything the scan does not apply is simply not applied. void ApplyResidualFilters(DataChunk &output); bool HasResidualFilters() const { return ! residual_filters.empty(); } + + // Per-scan override for the concurrent-row budget (the `fetch_size` named + // parameter). 0 means "use the erpl_rfc_fetch_size session setting". + void SetFetchSize(unsigned int n) { fetch_size_override = n; } + unsigned int EffectiveFetchSize() const { + return fetch_size_override != 0 ? fetch_size_override : GetRfcReadTableBatchBudget(); + } }; enum class ReadTableStates { diff --git a/rfc/src/sap_rfc.cpp b/rfc/src/sap_rfc.cpp index 477edea..1173e6a 100644 --- a/rfc/src/sap_rfc.cpp +++ b/rfc/src/sap_rfc.cpp @@ -61,6 +61,12 @@ namespace duckdb void SetRfcPushdownFilters(bool enabled) { g_rfc_pushdown_filters.store(enabled, std::memory_order_relaxed); } bool GetRfcPushdownFilters() { return g_rfc_pushdown_filters.load(std::memory_order_relaxed); } + // Session default for the `threads` named parameter. 0 means "erpl decides", + // which today is one RFC call per projected column. + static std::atomic g_rfc_max_threads{0}; + void SetRfcMaxThreads(unsigned int n) { g_rfc_max_threads.store(n, std::memory_order_relaxed); } + unsigned int GetRfcMaxThreads() { return g_rfc_max_threads.load(std::memory_order_relaxed); } + // Concurrent-row budget that bounds the SAP SDK result buffer on wide // sap_read_table scans (issue #69). See MaxBatchSizeForColumnCount. static std::atomic g_rfc_read_table_batch_budget{RfcReadColumnStateMachine::DEFAULT_READ_TABLE_BATCH_BUDGET}; @@ -1074,7 +1080,7 @@ namespace duckdb // outside any per-state-machine lock — and read locklessly by the // tasks scheduled below; the active set is fixed for the whole scan. effective_max_batch_size = RfcReadColumnStateMachine::MaxBatchSizeForColumnCount( - (unsigned int)active.size(), GetRfcReadTableBatchBudget()); + (unsigned int)active.size(), EffectiveFetchSize()); // When max_threads > 0, the user wants at most that many concurrent // RFC calls. Enforce by scheduling tasks in batches. Each batch is diff --git a/rfc/src/scanner_read_table.cpp b/rfc/src/scanner_read_table.cpp index fdb2459..4480034 100644 --- a/rfc/src/scanner_read_table.cpp +++ b/rfc/src/scanner_read_table.cpp @@ -29,9 +29,11 @@ namespace duckdb auto table_name = input.inputs[0].ToString(); auto &named_params = input.named_parameters; + // Per-query `threads` wins; otherwise the erpl_rfc_max_threads session default; + // otherwise 0, which means erpl decides (one RFC call per projected column). auto max_read_threads = named_params.find("THREADS") != named_params.end() ? named_params["THREADS"].GetValue() - : 0; + : GetRfcMaxThreads(); auto limit = named_params.find("MAX_ROWS") != named_params.end() ? named_params["MAX_ROWS"].GetValue() : 0; @@ -54,12 +56,20 @@ namespace duckdb ? named_params["SECRET"].ToString() : ""; + // `fetch_size` is the shared name across sap_read_table, sap_odp_read_* and the + // BICS scanners. Each protocol keeps its own natural unit -- here it is + // concurrent result rows, which is what bounds the SAP SDK's own buffer. + auto fetch_size = named_params.find("FETCH_SIZE") != named_params.end() + ? named_params["FETCH_SIZE"].GetValue() + : 0; + auto bind_data = make_uniq(table_name, max_read_threads, limit, read_table_function, read_table_delimiter, read_table_function_user_set, &DefaultRfcConnectionFactory, context); if (!secret_name.empty()) { bind_data->SetSecretName(secret_name); } + bind_data->SetFetchSize(fetch_size); bind_data->InitOptionsFromWhereClause(where_clause); try { bind_data->InitAndVerifyFields(fields); @@ -149,6 +159,7 @@ namespace duckdb RfcReadTableBind, RfcReadTableInitGlobalState); fun.named_parameters["THREADS"] = LogicalType::UINTEGER; + fun.named_parameters["FETCH_SIZE"] = LogicalType::UINTEGER; fun.named_parameters["COLUMNS"] = LogicalType::LIST(LogicalType::VARCHAR); fun.named_parameters["FILTER"] = LogicalType::VARCHAR; fun.named_parameters["MAX_ROWS"] = LogicalType::UINTEGER; diff --git a/rfc/test/sql/sap_read_table_tuning_api.test b/rfc/test/sql/sap_read_table_tuning_api.test new file mode 100644 index 0000000..f9e364b --- /dev/null +++ b/rfc/test/sql/sap_read_table_tuning_api.test @@ -0,0 +1,131 @@ +# name: test/sql/sap_read_table_tuning_api.test +# description: The shared tuning vocabulary -- threads / fetch_size and their +# erpl_rfc_* session defaults -- must change only how the data is +# fetched, never which rows come back. +# group: [rfc] + +# `threads` and `fetch_size` mean the same thing in every erpl extension, so what a +# user learns on sap_read_table applies to sap_odp_read_* too. The invariant that +# makes them safe to tune is that they are pure transport settings: identical result +# sets at every value. +# +# erpl_rfc_read_table_batch_budget is the original spelling of erpl_rfc_fetch_size. +# They are two names for one knob, not two knobs -- both write the same value. + +require erpl_rfc + +require-env ERPL_SAP_ASHOST + +require-env ERPL_SAP_SYSNR + +require-env ERPL_SAP_USER + +require-env ERPL_SAP_PASSWORD + +require-env ERPL_SAP_CLIENT + +require-env ERPL_SAP_LANG + +statement ok +CREATE SECRET abap_trial ( + TYPE sap_rfc, + ASHOST '${ERPL_SAP_ASHOST}', + SYSNR '${ERPL_SAP_SYSNR}', + CLIENT '${ERPL_SAP_CLIENT}', + USER '${ERPL_SAP_USER}', + PASSWD '${ERPL_SAP_PASSWORD}', + LANG '${ERPL_SAP_LANG}' +); + +# --------------------------------------------------------------------- +# Both spellings of the fetch-size knob exist and are settable. +# --------------------------------------------------------------------- +statement ok +SET erpl_rfc_fetch_size = 32768; + +query I +SELECT current_setting('erpl_rfc_fetch_size'); +---- +32768 + +statement ok +SET erpl_rfc_read_table_batch_budget = 65536; + +query I +SELECT current_setting('erpl_rfc_read_table_batch_budget'); +---- +65536 + +statement ok +SET erpl_rfc_max_threads = 4; + +query I +SELECT current_setting('erpl_rfc_max_threads'); +---- +4 + +statement ok +SET erpl_rfc_max_threads = 0; + +# --------------------------------------------------------------------- +# The named parameters are accepted and do not change the result. +# +# fetch_size is deliberately pushed far below the row count so the scan is +# forced through many more round-trips than the default would use -- if batch +# boundaries could drop or duplicate a row, this is where it would show. +# --------------------------------------------------------------------- +statement ok +SET erpl_rfc_fetch_size = 1310720; + +statement ok +CREATE TABLE tune_ref AS +SELECT TABNAME, TABCLASS FROM sap_read_table('DD02L'); + +statement ok +CREATE TABLE tune_small AS +SELECT TABNAME, TABCLASS FROM sap_read_table('DD02L', fetch_size = 4096); + +statement ok +CREATE TABLE tune_threads AS +SELECT TABNAME, TABCLASS FROM sap_read_table('DD02L', fetch_size = 8192, threads = 2); + +query I +SELECT count(*) > 50000 FROM tune_ref; +---- +true + +query I +SELECT count(*) FROM ( + (SELECT * FROM tune_ref EXCEPT ALL SELECT * FROM tune_small) + UNION ALL + (SELECT * FROM tune_small EXCEPT ALL SELECT * FROM tune_ref) +); +---- +0 + +query I +SELECT count(*) FROM ( + (SELECT * FROM tune_ref EXCEPT ALL SELECT * FROM tune_threads) + UNION ALL + (SELECT * FROM tune_threads EXCEPT ALL SELECT * FROM tune_ref) +); +---- +0 + +# The session default applies when no named parameter is given, and is likewise +# transport-only. +statement ok +SET erpl_rfc_fetch_size = 4096; + +statement ok +CREATE TABLE tune_session AS +SELECT TABNAME, TABCLASS FROM sap_read_table('DD02L'); + +query I +SELECT count(*) FROM ( + (SELECT * FROM tune_ref EXCEPT ALL SELECT * FROM tune_session) + UNION ALL + (SELECT * FROM tune_session EXCEPT ALL SELECT * FROM tune_ref) +); +---- +0