diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d726ad..2e3796b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/rfc/src/include/sap_rfc.hpp b/rfc/src/include/sap_rfc.hpp index 858fbfd..b4e7b01 100644 --- a/rfc/src/include/sap_rfc.hpp +++ b/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 diff --git a/rfc/src/sap_rfc.cpp b/rfc/src/sap_rfc.cpp index f3702e2..947f4f1 100644 --- a/rfc/src/sap_rfc.cpp +++ b/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::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::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; } } diff --git a/rfc/src/scanner_read_table.cpp b/rfc/src/scanner_read_table.cpp index ae53e9c..d12314d 100644 --- a/rfc/src/scanner_read_table.cpp +++ b/rfc/src/scanner_read_table.cpp @@ -1,3 +1,4 @@ +#include #include #ifdef __GLIBC__ #include @@ -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::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 diff --git a/rfc/test/cpp/test_row_window_scheduler.cpp b/rfc/test/cpp/test_row_window_scheduler.cpp index 732aaee..9646d35 100644 --- a/rfc/test/cpp/test_row_window_scheduler.cpp +++ b/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,7 +48,7 @@ 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); } } @@ -56,13 +56,13 @@ TEST_CASE("every offset is a multiple of the batch size", "[erpl_rfc][partition] 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 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::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 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); +}