fix(rfc): refuse the ROWSKIPS ceiling instead of truncating, and bound the fallback retry

CHANGELOG.md

@@ -44,6 +44,23 @@ LOAD erpl;
   a different code path that did not, so every worker thread's glibc arena kept its
   high-water mark — which is what `getrusage` reports.
 
+- **[rfc]** **A partitioned scan past the `ROWSKIPS` ceiling returned a truncated result
+  instead of refusing.** The row-window scheduler signalled the ABAP `INT4` limit by
+  returning "no more windows", which is the same answer it gives at the end of a table —
+  and a worker retires on an empty chunk, which DuckDB reads as end-of-scan. A scan beyond
+  2,147,483,647 rows therefore produced a silently short answer, while `API_REFERENCE`
+  states that erpl refuses rather than wrapping. It now raises, naming the limit.
+
+  The same guard was off by one and refused the **last legal window** (start
+  2,147,450,880 at a 32768-row window). Both are covered by new boundary tests.
+
+- **[rfc]** **The `RFC_READ_TABLE` fallback could retry without limit.** After the runtime
+  fallback had switched functions, the selection call returns a cached success, and the
+  retry branch neither counted the attempt nor slept — so a second `TABLE_WITHOUT_DATA`
+  from the fallback function became a tight loop against the SAP system, once per
+  partition worker. It now retries only when the selected function actually changed, and
+  otherwise reports the original error.
+
 ## v2026.09.02 — a narrow extract can finally use more than one connection
 
 `sap_read_table` parallelised across *columns*, so reading a few columns out of a very

rfc/src/include/sap_rfc.hpp

@@ -76,12 +76,18 @@ namespace duckdb
 			// 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,
+			// Claims the next window.  `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);
+			//
+			// EXHAUSTED and ADDRESS_LIMIT must NOT be collapsed into one "false".  A
+			// worker retires on an empty chunk, and DuckDB reads an empty chunk as
+			// end-of-scan -- so returning the same answer for both would turn the
+			// ROWSKIPS ceiling into a silently truncated result, which is exactly what
+			// the ceiling exists to prevent.
+			enum class ClaimResult { CLAIMED, EXHAUSTED, ADDRESS_LIMIT };
+			ClaimResult 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

rfc/src/sap_rfc.cpp

@@ -1145,10 +1145,10 @@ namespace duckdb
         window_size = batches * batch_size;
     }
 
-    bool RfcRowWindowScheduler::Claim(idx_t &offset, idx_t &count)
+    RfcRowWindowScheduler::ClaimResult RfcRowWindowScheduler::Claim(idx_t &offset, idx_t &count)
     {
         if (exhausted.load(std::memory_order_acquire)) {
-            return false;
+            return ClaimResult::EXHAUSTED;
         }
 
         auto claimed = next_offset.fetch_add(window_size, std::memory_order_relaxed);
@@ -1157,14 +1157,26 @@ namespace duckdb
         // INT32_MAX.  Stopping here also keeps next_offset from ever wrapping, which
         // would start handing out offsets that have already been read -- duplicated
         // rows rather than an error.
-        if (claimed > (idx_t)std::numeric_limits<int32_t>::max() - window_size) {
+        //
+        // This is reported apart from EXHAUSTED and never as an empty chunk: DuckDB
+        // reads an empty chunk as end-of-scan, so collapsing the two would answer a
+        // 2.1-billion-row scan with a silently truncated prefix -- trading the
+        // duplicate-rows failure this guard prevents for a missing-rows one, which is
+        // no better and is what API_REFERENCE promises we do not do.
+        //
+        // The bound is `claimed > INT32_MAX`, not `> INT32_MAX - window_size`:
+        // next_offset is 64-bit so subtracting the window buys no wrap protection, and
+        // every individual ROWSKIPS inside the last window is already range-checked
+        // loudly in CreateFunctionArguments.  The old bound refused the last legal
+        // window start (2,147,450,880 at a 32768 window).
+        if (claimed > (idx_t)std::numeric_limits<int32_t>::max()) {
             exhausted.store(true, std::memory_order_release);
-            return false;
+            return ClaimResult::ADDRESS_LIMIT;
         }
 
         if (max_rows > 0) {
             if (claimed >= max_rows) {
-                return false;
+                return ClaimResult::EXHAUSTED;
             }
             // The last window under MAX_ROWS is short.  Only the LOGICAL count
             // shrinks; the caller still requests whole batches from SAP and clips,
@@ -1176,7 +1188,7 @@ namespace duckdb
         }
 
         offset = claimed;
-        return true;
+        return ClaimResult::CLAIMED;
     }
 
     void RfcRowWindowScheduler::ReportExhausted()
@@ -1911,11 +1923,19 @@ namespace duckdb
                 std::string err_msg(e.what());
                 if (!IsRetryableRfcError(err_msg)) {
                     if (rfc_type.IsStringType() && err_msg.find("TABLE_WITHOUT_DATA") != std::string::npos) {
+                        // Retry only if this attempt actually CHANGED the function we
+                        // call. TrySelectFallbackReadTableFunction returns a cached
+                        // success once the switch has happened, and this branch does not
+                        // increment `attempt` and does not sleep -- so retrying on a
+                        // cached success is an unbounded tight loop hammering SAP, once
+                        // per partition worker. Switching is a one-way move, so a second
+                        // TABLE_WITHOUT_DATA from the fallback is a real failure.
+                        auto before = bind_data->GetReadTableFunctionName();
                         auto fallback_connection = bind_data->OpenNewConnection();
-                        auto fallback_selected = bind_data->TrySelectFallbackReadTableFunction(fallback_connection);
+                        bind_data->TrySelectFallbackReadTableFunction(fallback_connection);
                         fallback_connection->Close();
-                        if (fallback_selected) {
-                            // retry immediately with the fallback function
+                        if (bind_data->GetReadTableFunctionName() != before) {
+                            // retry immediately with the newly selected function
                             continue;
                         }
                     }

rfc/src/scanner_read_table.cpp

@@ -1,3 +1,4 @@
+#include <limits>
 #include <regex>
 #ifdef __GLIBC__
 #include <malloc.h>
@@ -158,7 +159,18 @@ namespace duckdb
         while (true) {
             if (! lstate.holds_window) {
                 idx_t offset = 0, rows = 0;
-                if (! gstate.scheduler->Claim(offset, rows)) {
+                auto claim = gstate.scheduler->Claim(offset, rows);
+                if (claim == RfcRowWindowScheduler::ClaimResult::ADDRESS_LIMIT) {
+                    // Refuse loudly. Retiring the worker here would return a truncated
+                    // prefix and call it success, because DuckDB reads the empty chunk
+                    // that retires a worker as end-of-scan.
+                    throw InvalidInputException(
+                        "sap_read_table: a partitioned scan reached the ROWSKIPS limit of %d rows "
+                        "(ABAP INT4). Restrict the scan with a WHERE clause on an indexed column, "
+                        "or read it in ranges; erpl will not silently return a partial result.",
+                        std::numeric_limits<int32_t>::max());
+                }
+                if (claim != RfcRowWindowScheduler::ClaimResult::CLAIMED) {
                     // Nothing left to claim; an empty chunk retires this worker.
 #ifdef __GLIBC__
                     // Mirror the serial path: this worker is done, its per-column SDK

rfc/test/cpp/test_row_window_scheduler.cpp

@@ -30,15 +30,15 @@ TEST_CASE("windows are handed out in order and do not overlap", "[erpl_rfc][part
 	RfcRowWindowScheduler sched(/*window_size=*/4096, /*batch_size=*/2048, /*max_rows=*/0);
 
 	idx_t off = 0, count = 0;
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 0);
 	REQUIRE(count == 4096);
 
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 4096);
 	REQUIRE(count == 4096);
 
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 8192);
 }
 
@@ -48,21 +48,21 @@ TEST_CASE("every offset is a multiple of the batch size", "[erpl_rfc][partition]
 	RfcRowWindowScheduler sched(/*window_size=*/6144, /*batch_size=*/2048, /*max_rows=*/0);
 	idx_t off = 0, count = 0;
 	for (int i = 0; i < 10; i++) {
-		REQUIRE(sched.Claim(off, count));
+		REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 		REQUIRE(off % 2048 == 0);
 	}
 }
 
 TEST_CASE("a short read stops further claims", "[erpl_rfc][partition]") {
 	RfcRowWindowScheduler sched(/*window_size=*/4096, /*batch_size=*/2048, /*max_rows=*/0);
 	idx_t off = 0, count = 0;
-	REQUIRE(sched.Claim(off, count));
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 
 	// The second worker found the table ends inside its window.
 	sched.ReportExhausted();
 
-	REQUIRE_FALSE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) != RfcRowWindowScheduler::ClaimResult::CLAIMED);
 }
 
 TEST_CASE("exhaustion does not invalidate a window already claimed", "[erpl_rfc][partition]") {
@@ -73,15 +73,15 @@ TEST_CASE("exhaustion does not invalidate a window already claimed", "[erpl_rfc]
 	RfcRowWindowScheduler sched(/*window_size=*/4096, /*batch_size=*/2048, /*max_rows=*/0);
 	idx_t a_off = 0, a_count = 0;
 	idx_t b_off = 0, b_count = 0;
-	REQUIRE(sched.Claim(a_off, a_count));
-	REQUIRE(sched.Claim(b_off, b_count));
+	REQUIRE(sched.Claim(a_off, a_count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
+	REQUIRE(sched.Claim(b_off, b_count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 
 	sched.ReportExhausted();
 
 	// Both claims stay valid and stay disjoint; only NEW claims are refused.
 	REQUIRE(a_off + a_count == b_off);
 	idx_t off = 0, count = 0;
-	REQUIRE_FALSE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) != RfcRowWindowScheduler::ClaimResult::CLAIMED);
 }
 
 // ---------------------------------------------------------------------------
@@ -93,31 +93,31 @@ TEST_CASE("MAX_ROWS is a scan-wide limit, not a per-worker one", "[erpl_rfc][par
 	RfcRowWindowScheduler sched(/*window_size=*/4096, /*batch_size=*/2048, /*max_rows=*/5000);
 
 	idx_t off = 0, count = 0;
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 0);
 	REQUIRE(count == 4096);
 
 	// Only 904 rows remain under the limit.
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 4096);
 	REQUIRE(count == 904);
 
-	REQUIRE_FALSE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) != RfcRowWindowScheduler::ClaimResult::CLAIMED);
 }
 
 TEST_CASE("a MAX_ROWS below one window still yields exactly that many", "[erpl_rfc][partition]") {
 	RfcRowWindowScheduler sched(/*window_size=*/4096, /*batch_size=*/2048, /*max_rows=*/10);
 	idx_t off = 0, count = 0;
-	REQUIRE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(off == 0);
 	REQUIRE(count == 10);
-	REQUIRE_FALSE(sched.Claim(off, count));
+	REQUIRE(sched.Claim(off, count) != RfcRowWindowScheduler::ClaimResult::CLAIMED);
 }
 
 TEST_CASE("the total handed out never exceeds MAX_ROWS", "[erpl_rfc][partition]") {
 	RfcRowWindowScheduler sched(/*window_size=*/1024, /*batch_size=*/512, /*max_rows=*/3000);
 	idx_t off = 0, count = 0, total = 0;
-	while (sched.Claim(off, count)) {
+	while (sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED) {
 		total += count;
 	}
 	REQUIRE(total == 3000);
@@ -138,7 +138,7 @@ TEST_CASE("concurrent claims never overlap and never skip a row", "[erpl_rfc][pa
 	for (int t = 0; t < 8; t++) {
 		threads.emplace_back([&]() {
 			idx_t off = 0, count = 0;
-			while (sched.Claim(off, count)) {
+			while (sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED) {
 				std::lock_guard<std::mutex> g(m);
 				claims.emplace_back(off, count);
 			}
@@ -177,16 +177,27 @@ TEST_CASE("an absurd window size cannot wrap to zero", "[erpl_rfc][partition]")
 	REQUIRE(sched.WindowSize() > 0);
 	REQUIRE(sched.WindowSize() % 2048 == 0);
 
-	// A window this size covers everything ROWSKIPS can address, so exactly one
-	// claim is possible and the next is refused rather than advancing past the
-	// ABAP INT4 ceiling. What must never happen is a second claim at the SAME
-	// offset, which is what wrapping would produce.
-	idx_t a_off = 1, a_count = 0, b_off = 1, b_count = 0;
-	REQUIRE(sched.Claim(a_off, a_count));
+	// The constructor clamps the window to INT32_MAX before rounding down to whole
+	// batches, so the window is 2,147,481,600 rows. Two starts are therefore legal --
+	// 0 and 2,147,481,600, the latter still below the ROWSKIPS ceiling -- and only the
+	// third exceeds it. (The previous bound, `claimed > INT32_MAX - window_size`,
+	// refused the second and made the last legal window unreachable.) Any ROWSKIPS
+	// inside that final window that does cross the ceiling is refused loudly by
+	// CreateFunctionArguments, not here.
+	//
+	// What must never happen is a second claim at the SAME offset, which is what a
+	// wrapped window_size would produce.
+	idx_t a_off = 1, a_count = 0, b_off = 1, c_off = 1, b_count = 0, c_count = 0;
+	REQUIRE(sched.Claim(a_off, a_count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(a_off == 0);
 	REQUIRE(a_count > 0);
 
-	REQUIRE_FALSE(sched.Claim(b_off, b_count));
+	REQUIRE(sched.Claim(b_off, b_count) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
+	REQUIRE(b_off != a_off);
+	REQUIRE(b_off <= (idx_t)std::numeric_limits<int32_t>::max());
+
+	// Past the ceiling: refused loudly, and distinguishable from a normal end-of-scan.
+	REQUIRE(sched.Claim(c_off, c_count) == RfcRowWindowScheduler::ClaimResult::ADDRESS_LIMIT);
 }
 
 TEST_CASE("claims never repeat an offset even at extreme window sizes",
@@ -199,7 +210,7 @@ TEST_CASE("claims never repeat an offset even at extreme window sizes",
 
 		std::set<idx_t> seen;
 		idx_t off = 0, count = 0;
-		for (int i = 0; i < 5 && sched.Claim(off, count); i++) {
+		for (int i = 0; i < 5 && sched.Claim(off, count) == RfcRowWindowScheduler::ClaimResult::CLAIMED; i++) {
 			REQUIRE(seen.insert(off).second);   // never the same offset twice
 			REQUIRE(count > 0);
 		}
@@ -219,7 +230,62 @@ TEST_CASE("an absurd batch size cannot break alignment either", "[erpl_rfc][part
 	REQUIRE(sched.WindowSize() % sched.BatchSize() == 0);
 
 	idx_t a = 0, c = 0;
-	REQUIRE(sched.Claim(a, c));
+	REQUIRE(sched.Claim(a, c) == RfcRowWindowScheduler::ClaimResult::CLAIMED);
 	REQUIRE(a == 0);
 	REQUIRE(c > 0);
 }
+
+TEST_CASE("The ROWSKIPS ceiling is refused loudly, not reported as end-of-scan",
+          "[erpl_rfc][partition]") {
+	using Sched = RfcRowWindowScheduler;
+	constexpr idx_t INT32_MAX_ROWS = 2147483647;
+	constexpr idx_t BATCH = 32768;
+
+	// A worker retires on an empty chunk and DuckDB reads that as end-of-scan, so
+	// EXHAUSTED and ADDRESS_LIMIT must stay distinguishable: collapsing them turns a
+	// 2.1-billion-row scan into a silently truncated prefix.
+	Sched sched(BATCH, BATCH, 0);
+
+	// The last legal window start is the greatest batch-aligned offset <= INT32_MAX.
+	// The previous guard (claimed > INT32_MAX - window_size) refused it.
+	const idx_t last_legal = (INT32_MAX_ROWS / BATCH) * BATCH;
+
+	idx_t offset = 0, count = 0;
+	idx_t seen_last_legal = 0;
+	for (idx_t i = 0;; i++) {
+		auto r = sched.Claim(offset, count);
+		if (r == Sched::ClaimResult::CLAIMED) {
+			if (offset == last_legal) {
+				seen_last_legal++;
+			}
+			REQUIRE(offset <= INT32_MAX_ROWS);
+			continue;
+		}
+		// Unbounded scan: the only way out is the address limit, never EXHAUSTED.
+		REQUIRE(r == Sched::ClaimResult::ADDRESS_LIMIT);
+		break;
+	}
+	REQUIRE(seen_last_legal == 1);
+
+	// Once the limit is hit the scheduler stays stopped, and a later caller must not be
+	// told "exhausted" -- it would draw the same wrong conclusion.
+	auto again = sched.Claim(offset, count);
+	REQUIRE(again == Sched::ClaimResult::EXHAUSTED);
+}
+
+TEST_CASE("A bounded scan reports EXHAUSTED, never the address limit",
+          "[erpl_rfc][partition]") {
+	using Sched = RfcRowWindowScheduler;
+	Sched sched(2048, 2048, 10000);
+
+	idx_t offset = 0, count = 0, total = 0;
+	while (true) {
+		auto r = sched.Claim(offset, count);
+		if (r != Sched::ClaimResult::CLAIMED) {
+			REQUIRE(r == Sched::ClaimResult::EXHAUSTED);
+			break;
+		}
+		total += count;
+	}
+	REQUIRE(total == 10000);
+}