Merge remote-tracking branch 'origin/master' into claude/fix-partition-address-limit

# Conflicts:
#	CHANGELOG.md

Author: jrosskopf

CHANGELOG.md

@@ -26,6 +26,24 @@ LOAD erpl;
 
 ### Fixed
 
+- **[rfc]** **`fetch_size` and `partitions` did nothing to the batch size of a
+  single-column scan.** `MaxBatchSizeForColumnCount` returned the full 32768-row cap
+  whenever a scan projected one column, discarding the budget that
+  `ResolveEffectiveMaxBatchSize` had just divided across the partition workers — so every
+  worker asked SAP for `ROWCOUNT=32768` however `erpl_rfc_fetch_size` and `partitions`
+  were set. The exemption was deliberate, on the reasoning that a narrow scan is already
+  under any sane budget; that reasoning ignored the division, and a narrow scan is
+  precisely what partitioning exists for.
+
+  Measured on a 300,000-row single-column read at `partitions = 8`: peak RSS **405 MB →
+  307 MB (−24%)** with no loss of throughput (2.17s → 2.07s). The budget now binds for
+  every column count; only a zero budget or a scan with no projected column bypasses it.
+
+- **[rfc]** A partitioned scan never returned its allocator arenas to the OS. The serial
+  path calls `malloc_trim(0)` when the scan finishes, but each partition worker retires on
+  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 —

rfc/src/sap_rfc.cpp

@@ -1246,8 +1246,8 @@ namespace duckdb
         // so without dividing here the budget would be per worker and peak memory would
         // scale with the partition count -- the opposite of what a budget is for.
         //
-        // Narrow scans, which is what partitioning is for, are unaffected: the batch is
-        // already capped at MAX_BATCH_SIZE long before the divided budget binds.
+        // Narrow scans are the point of partitioning, so they must feel this division
+        // too -- see MaxBatchSizeForColumnCount, which no longer exempts a single column.
         auto budget = EffectiveFetchSize();
         auto workers = GetPartitionCount();
         if (workers > 1 && budget > 0) {
@@ -1592,7 +1592,15 @@ namespace duckdb
         // concurrent-row budget, floored to a power of two in
         // [STANDARD_VECTOR_SIZE, MAX_BATCH_SIZE] so the doubling warm-up still
         // reaches the cap cleanly and ROWSKIPS % ROWCOUNT stays valid.
-        if (num_columns <= 1 || concurrent_row_budget == 0) {
+        // A zero budget disables the cap, and a scan with no projected column has
+        // nothing to bound. Note that num_columns == 1 is NOT exempt: it used to be,
+        // on the reasoning that a narrow scan is already under any sane budget -- but
+        // ResolveEffectiveMaxBatchSize divides the budget by the partition count, and
+        // a narrow scan is precisely what partitioning is for. Exempting it meant
+        // erpl_rfc_fetch_size and partitions silently did nothing to the batch for the
+        // one scan shape that most needed them (heaptrack showed 32768-row batches on
+        // every worker at partitions=8).
+        if (num_columns == 0 || concurrent_row_budget == 0) {
             return MAX_BATCH_SIZE;
         }
         unsigned int per_col = concurrent_row_budget / num_columns;

rfc/src/scanner_read_table.cpp

@@ -172,6 +172,13 @@ namespace duckdb
                 }
                 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
+                    // handles are released, and each worker thread has its own glibc
+                    // arena. Without this the arenas keep their high-water mark and
+                    // getrusage still reports the peak long after the rows are gone.
+                    malloc_trim(0);
+#endif
                     return;
                 }
                 for (auto &sm : lstate.machines) {

rfc/test/cpp/test_read_table_batching.cpp

@@ -78,13 +78,39 @@ TEST_CASE("NextDesiredBatchSize doubles geometrically while keeping ROWSKIPS val
 	REQUIRE(size == SM::MAX_BATCH_SIZE);
 }
 
+TEST_CASE("A single-column scan honours a small budget instead of ignoring it",
+          "[erpl_rfc][batching]") {
+	using SM = RfcReadColumnStateMachine;
+
+	// ResolveEffectiveMaxBatchSize divides the budget across partition workers so that
+	// peak memory does not scale with the worker count. A single-column scan used to
+	// short-circuit to MAX_BATCH_SIZE before the divided budget was ever consulted, so
+	// every worker asked SAP for ROWCOUNT=32768 no matter what erpl_rfc_fetch_size or
+	// partitions said -- confirmed on a4h by heaptrack, which showed 32768-row batches
+	// under partitions=8. Narrow scans are exactly what partitioning is for, so this is
+	// the case where the budget matters most.
+	REQUIRE(SM::MaxBatchSizeForColumnCount(1, 2048) == (unsigned int)STANDARD_VECTOR_SIZE);
+	REQUIRE(SM::MaxBatchSizeForColumnCount(1, 8192) == 8192u);
+
+	// The default fetch size (16384 concurrent rows) split across 8 workers.
+	REQUIRE(SM::MaxBatchSizeForColumnCount(1, 16384u / 8u) == (unsigned int)STANDARD_VECTOR_SIZE);
+
+	// A budget large enough for the full batch still yields it -- the cap binds, it does
+	// not shrink for its own sake.
+	REQUIRE(SM::MaxBatchSizeForColumnCount(1, 256u * 1024u) == SM::MAX_BATCH_SIZE);
+
+	// Zero columns still means "no projection to bound"; a zero budget still disables it.
+	REQUIRE(SM::MaxBatchSizeForColumnCount(0, 2048) == SM::MAX_BATCH_SIZE);
+	REQUIRE(SM::MaxBatchSizeForColumnCount(1, 0) == SM::MAX_BATCH_SIZE);
+}
+
 TEST_CASE("MaxBatchSizeForColumnCount bounds the SDK buffer on wide scans",
           "[erpl_rfc][batching]") {
 	using SM = RfcReadColumnStateMachine;
 	// Use an explicit budget so the test is independent of the runtime default.
 	constexpr unsigned int BUDGET = 256u * 1024u;
 
-	// Narrow scans keep the full batch for throughput.
+	// Narrow scans keep the full batch when the budget is large enough to allow it.
 	REQUIRE(SM::MaxBatchSizeForColumnCount(0, BUDGET) == SM::MAX_BATCH_SIZE);
 	REQUIRE(SM::MaxBatchSizeForColumnCount(1, BUDGET) == SM::MAX_BATCH_SIZE);
 	REQUIRE(SM::MaxBatchSizeForColumnCount(2, BUDGET) == SM::MAX_BATCH_SIZE);