feat(rfc): row-range partitioning and a fetch-size knob for sap_read_table

API_REFERENCE.md

@@ -92,6 +92,7 @@ Read data from an SAP table or CDS view. Supports projection pushdown, filter pu
 | `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 |
+| `partitions` | UINTEGER | `erpl_rfc_partitions` | Read this many row ranges in parallel. Same rows, unspecified order |
 | `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 |
@@ -1283,6 +1284,8 @@ Notes:
 | `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_partitions` | UBIGINT | 0 | Split a `sap_read_table` scan into this many row ranges read in parallel. `0` reads in one pass, parallelising across columns instead. See [Narrow tables](#narrow-tables-use-partitions-not-threads) |
+| `erpl_rfc_partition_window_rows` | UBIGINT | 0 | Rows a partition worker claims at a time; `0` uses one RFC batch per window |
 | `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` |
@@ -1442,6 +1445,54 @@ running a large parallel extract against a production system.
 `erpl_rfc_fetch_size`. Both remain supported and write the same value — they are two
 names for one knob, not two knobs.
 
+#### Narrow tables: use `partitions`, not `threads`
+
+`threads` parallelises across **columns** — one concurrent `RFC_READ_TABLE` call per
+projected column. That suits a wide extract and does nothing for a narrow one: a
+single-column scan issues one call, and `threads` has nothing to spread.
+
+`partitions` splits the **rows** instead. Each worker claims a window of the table and
+reads it with `ROWSKIPS`/`ROWCOUNT`:
+
+```sql
+SELECT * FROM sap_read_table('DD02L', partitions = 8);
+
+-- or as a session default
+SET erpl_rfc_partitions = 8;
+```
+
+**What to expect.** Measured on a 164,664-row single-column extract, release build:
+
+| `partitions` | 1 | 2 | 4 | 8 | 16 |
+|---|---|---|---|---|---|
+| wall | 2.23s | 1.45s | 1.11s | 0.83s | 0.92s |
+
+About **2.7x at eight workers**, with the knee clearly visible — past it, more workers
+cost more than they return. Where the knee falls is a property of the SAP system's
+capacity, not of erpl, so find it by raising the value and watching throughput.
+
+**Rows come back unordered.** An unpartitioned scan calls `RFC_READ_TABLE` with
+`GET_SORTED='X'` and returns rows in that order; partitioned workers finish in whatever
+order they finish. That is why this is opt-in rather than the default. Add an
+`ORDER BY` if you need one.
+
+`erpl_rfc_partition_window_rows` sets how many rows a worker claims at a time. The
+default is one RFC batch per window, which shares the table evenly across workers;
+larger windows mean fewer, coarser hand-offs.
+
+Two caveats worth knowing:
+
+- `ROWSKIPS` is a 32-bit integer, so a partitioned scan cannot start a window beyond
+  row 2,147,483,647. erpl refuses rather than wrapping, because a wrap would silently
+  re-read an earlier range and duplicate rows. Narrow the read with a `WHERE` clause.
+- Windows are offsets into a server-side sort, not a snapshot. If the table is written
+  while you read it, rows can shift between windows. That is true of an unpartitioned
+  scan too — it also pages by offset — but partitioning makes it easier to observe.
+
+For a genuinely large extract, running several *processes* against SAP still scales
+better than any in-process approach, because SAP-side concurrency limits apply per
+client program. Check with your Basis team first.
+
 ### SSH Tunnel + SAP Connection
 
 Complete workflow for connecting through an SSH jump host:

CHANGELOG.md

@@ -24,6 +24,19 @@ LOAD erpl;
 
 ## Unreleased
 
+### Added
+
+- **[rfc]** **`sap_read_table` can now split a scan by rows.** Parallelism was per
+  *column* — one concurrent `RFC_READ_TABLE` call per projected column — so a narrow
+  extract got none at all, and raising `threads` did nothing. `partitions` hands each
+  worker a row window read with `ROWSKIPS`/`ROWCOUNT`. Measured on a 164,664-row
+  single-column extract: 2.23s unpartitioned, 0.83s at eight partitions (**2.7x**),
+  with the knee visible at eight. Opt-in, because partitioned workers finish in
+  whatever order they finish while an unpartitioned scan returns rows sorted.
+
+- **[rfc]** `erpl_rfc_partitions` and `erpl_rfc_partition_window_rows`, plus a
+  `partitions` named parameter.
+
 ### Fixed
 
 - **[rfc]** **`sap_rfc_invoke` failed on any module whose result parameters are all
@@ -37,6 +50,17 @@ LOAD erpl;
   genuinely pivoted (path-selected) data from a bare invoke. Selecting a table through
   `path :=` still pivots to the row's fields, as before.
 
+- **[rfc]** `MAX_ROWS` combined with `partitions` hung. The unpartitioned path trims
+  `ROWCOUNT` to land exactly on the limit, which a partitioned window cannot do because
+  `ROWCOUNT` must stay batch-aligned; it over-fetches the final batch, and the clip then
+  left the read spinning with rows it would never emit.
+
+- **[rfc]** The runtime `RFC_READ_TABLE` fallback reassigned a `std::string` on the
+  bind data from execute time. Column-parallel reads already made that reachable from
+  two tasks at once; it is now serialised.
+
+---
+
 ## v2026.08.31 — every predicate applied, and most of them pushed to SAP
 
 Two of the entries below are **silent wrong-results bugs in released erpl**, not

rfc/src/erpl_rfc_extension.cpp

@@ -165,6 +165,14 @@ namespace duckdb {
         SetRfcMaxThreads(parameter.GetValue<unsigned int>());
     }
 
+    static void OnRfcPartitions(ClientContext &, SetScope, Value &parameter) {
+        SetRfcPartitions((idx_t)parameter.GetValue<uint64_t>());
+    }
+
+    static void OnRfcPartitionWindowRows(ClientContext &, SetScope, Value &parameter) {
+        SetRfcPartitionWindowRows((idx_t)parameter.GetValue<uint64_t>());
+    }
+
     static void OnRfcBackend(ClientContext &, SetScope, Value &parameter) {
         SetRfcBackend(parameter.GetValue<string>());
     }
@@ -297,6 +305,29 @@ namespace duckdb {
             Value::UINTEGER(0),
             OnRfcMaxThreads);
 
+        config.AddExtensionOption(
+            "erpl_rfc_partitions",
+            "Split a sap_read_table scan into this many row ranges read in parallel. "
+            "0 (the default) reads the table in one pass, parallelising across COLUMNS "
+            "instead -- which is what helps a wide extract and does nothing for a narrow "
+            "one. Set it above 1 when you are reading few columns from a large table. "
+            "Rows then come back in worker-completion order rather than the sorted order "
+            "an unpartitioned scan produces, so this is opt-in. Override per query with "
+            "the partitions named parameter.",
+            LogicalType::UBIGINT,
+            Value::UBIGINT(0),
+            OnRfcPartitions);
+
+        config.AddExtensionOption(
+            "erpl_rfc_partition_window_rows",
+            "Rows per window handed to one worker of a partitioned scan (default "
+            "131072). Larger windows mean fewer, bigger RFC calls and coarser load "
+            "balancing across workers; smaller ones the reverse. Only used when "
+            "erpl_rfc_partitions is above 1.",
+            LogicalType::UBIGINT,
+            Value::UBIGINT(0),
+            OnRfcPartitionWindowRows);
+
         config.AddExtensionOption(
             "erpl_rfc_pushdown_filters",
             "Translate SQL WHERE predicates into RFC_READ_TABLE's OPTIONS table so SAP "

rfc/src/include/sap_rfc.hpp

@@ -53,6 +53,53 @@ namespace duckdb
 	void SetRfcMaxThreads(unsigned int n);
 	unsigned int GetRfcMaxThreads();
 
+	void SetRfcPartitions(idx_t n);
+	idx_t GetRfcPartitions();
+	void SetRfcPartitionWindowRows(idx_t n);
+	idx_t GetRfcPartitionWindowRows();
+
+	// Hands out disjoint row windows to the workers of a partitioned sap_read_table
+	// scan.  Kept free of any RFC dependency on purpose: this is the code that loses
+	// or duplicates rows if it is wrong, and the equivalent logic in erpl_odp could
+	// only be exercised through a live connection -- which is why an observed
+	// under-count there went unexplained.
+	//
+	// Two SAP-side rules shape it, both verified against the trial:
+	//   * RFC_READ_TABLE rejects ROWSKIPS % ROWCOUNT != 0 outright, so every offset
+	//     handed out is a multiple of the batch size.
+	//   * A read at or past the end returns zero rows rather than erroring, so a
+	//     short read is how a worker discovers the table is exhausted.
+	class RfcRowWindowScheduler
+	{
+		public:
+			// window_size is rounded up to a multiple of batch_size so that every
+			// offset stays batch-aligned.  max_rows == 0 means "no limit".
+			RfcRowWindowScheduler(idx_t window_size, idx_t batch_size, idx_t max_rows);
+
+			// Claims the next window.  Returns false once the table is exhausted or
+			// MAX_ROWS is reached.  `count` is the LOGICAL row count for the window,
+			// which may be smaller than the window size when MAX_ROWS clips it; the
+			// caller still asks SAP for whole batches and clips locally, because
+			// ROWCOUNT must stay batch-aligned.
+			bool Claim(idx_t &offset, idx_t &count);
+
+			// Called by a worker that read fewer rows than it asked for.  Prevents
+			// NEW claims only -- a worker already holding a window must always be
+			// allowed to finish it, which is exactly the invariant erpl_odp broke.
+			void ReportExhausted();
+
+			bool IsExhausted() const;
+			idx_t BatchSize() const { return batch_size; }
+			idx_t WindowSize() const { return window_size; }
+
+		private:
+			idx_t window_size;
+			idx_t batch_size;
+			idx_t max_rows;
+			std::atomic<idx_t> next_offset{0};
+			std::atomic<bool> exhausted{false};
+	};
+
 	class RfcReadTableBindData : public TableFunctionData
     {
 		public: 
@@ -101,6 +148,8 @@ namespace duckdb
 			bool ReadTableSupportsEtData();
 			bool TrySelectFallbackReadTableFunction(std::shared_ptr<RfcConnection> connection);
 			void ResolveReadTableResultPath(std::shared_ptr<RfcConnection> connection);
+			// Guarded read of the result path; the runtime fallback can reassign it.
+			std::string GetReadTableResultPath();
 			void ResolveReadTableImportParams(std::shared_ptr<RfcConnection> connection);
 			bool ReadTableHasParam(const std::string &param_name);
 
@@ -157,6 +206,16 @@ namespace duckdb
 			// column index.  Copied because the TableFilterSet belongs to the plan.
 			std::vector<std::pair<idx_t, duckdb::unique_ptr<TableFilter>>> residual_filters;
 			unsigned int fetch_size_override = 0;
+			// projected column index -> RFC column index, built once by
+			// ActivateColumns() and read-only afterwards.  Keeps the projection
+			// reachable without the state machines.
+			std::vector<idx_t> projected_to_rfc_column;
+			idx_t partition_count = 0;
+			// Guards the one post-bind write to the bind data: the runtime
+			// RFC_READ_TABLE fallback, which reassigns read_table_function.
+			std::mutex fallback_selection_lock;
+			bool fallback_selection_done = false;
+			bool fallback_selection_succeeded = false;
 			// 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.
@@ -189,11 +248,36 @@ namespace duckdb
 			// optional: filter_pushdown = true makes DuckDB drop the filter from the
 			// plan, so anything the scan does not apply is simply not applied.
 			void ApplyResidualFilters(DataChunk &output);
+
+			// A fresh set of state machines carrying the current projection, for one
+			// worker of a partitioned scan.  Built from the projection recorded by
+			// ActivateColumns() so a worker's set matches the bind-owned one.
+			std::vector<RfcReadColumnStateMachine> CreateWindowStateMachines();
+
+			// Read one step of `machines` into `output`.  Columns are read one after
+			// another rather than fanned out: a partitioned scan already has one RFC
+			// call in flight per worker, and fanning out per column inside each worker
+			// would multiply concurrent calls by the column count with nothing bounding
+			// the product.
+			void StepWindow(ClientContext &context, DataChunk &output,
+			                std::vector<RfcReadColumnStateMachine> &machines);
 			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; }
+
+			// Row-range partitioning (opt-in).  0 or 1 means the scan runs the
+			// column-parallel path unchanged.
+			void SetPartitionCount(idx_t n) { partition_count = n; }
+			idx_t GetPartitionCount() const {
+				return partition_count != 0 ? partition_count : GetRfcPartitions();
+			}
+			idx_t GetPartitionWindowRows() const;
+			// Compute effective_max_batch_size from the active column count once, before
+			// any worker runs.  Step() does the same for the unpartitioned path.
+			void ResolveEffectiveMaxBatchSize();
+			unsigned int GetLimit() const { return limit; }
 			unsigned int EffectiveFetchSize() const {
 				return fetch_size_override != 0 ? fetch_size_override : GetRfcReadTableBatchBudget();
 			}
@@ -228,6 +312,7 @@ namespace duckdb
 			bool IsRowIdColumnId();
 			void SetRowIdColumnId();
 			unsigned int GetCardinality();
+			unsigned int GetTotalRows() const { return total_rows; }
 			unsigned int GetBatchCount();
 			unsigned int GetDesiredBatchSize();
 			std::string ToString();
@@ -252,6 +337,10 @@ namespace duckdb
 			// won a persistent slot from the bind-data budget.  Used by
 			// ExecuteNextTableReadForColumn to decide whether to close the
 			// connection at the end of a successful batch.
+			// Point this state machine at one window of a partitioned scan.  The batch
+			// size is pinned for the window's lifetime; see CreateFunctionArguments.
+			void SetWindow(idx_t offset, unsigned int batch_size, unsigned int rows_in_window);
+
 			bool HasApprovedPersistentSlot() const {
 				return persistent_decision == PersistentDecision::APPROVED;
 			}
@@ -306,6 +395,12 @@ namespace duckdb
 			bool active = true;
 			bool row_id_column_id = false;
 			unsigned int desired_batch_size = STANDARD_VECTOR_SIZE;
+			// Partitioned scans only.  window_offset is the first row of the window
+			// this state machine is reading; fixed_batch_size pins desired_batch_size
+			// so it never doubles, which is what keeps every ROWSKIPS a multiple of
+			// ROWCOUNT once an offset is involved.
+			idx_t window_offset = 0;
+			bool fixed_batch_size = false;
 			unsigned int pending_records = 0;
 			unsigned int cardinality = 0;
 			unsigned int batch_count = 0;
@@ -354,6 +449,32 @@ namespace duckdb
 			std::mutex thread_lock;
 	};
 
+	// Per-scan state for a partitioned read.  Holds the window scheduler; a null
+	// scheduler means the scan runs the unpartitioned, column-parallel path exactly as
+	// before.
+	class RfcReadTableGlobalState : public duckdb::GlobalTableFunctionState
+	{
+		public:
+			RfcReadTableGlobalState(idx_t max_threads_p,
+			                        duckdb::unique_ptr<RfcRowWindowScheduler> scheduler_p)
+				: max_threads(max_threads_p), scheduler(std::move(scheduler_p)) { }
+
+			idx_t MaxThreads() const override { return max_threads; }
+			bool IsPartitioned() const { return scheduler != nullptr; }
+
+			idx_t max_threads;
+			duckdb::unique_ptr<RfcRowWindowScheduler> scheduler;
+	};
+
+	// One worker of a partitioned scan: its own state machines, reading its own window.
+	class RfcReadTableWindowState : public duckdb::LocalTableFunctionState
+	{
+		public:
+			std::vector<RfcReadColumnStateMachine> machines;
+			bool holds_window = false;
+			idx_t window_rows = 0;
+	};
+
 	class RfcReadColumnTask : public duckdb::BaseExecutorTask 
 	{
 		public: