perf: reduce Blob read buffering and expose read window controls (#4979)

Remote Blob reads retain and concatenate intermediate IPC buffers across
batches and HTTP requests, increasing peak memory. Decode fragmented IPC
bodies using their declared length, assemble Blob values once across
batches/requests, and collect large-object Range responses into
known-size buffers. Impossible IPC/Range allocations return errors
instead of panicking. Closing a remote Blob file releases cached bytes
immediately, including while a cross-window read is waiting for the
network.

Rust and Python expose configurable Blob read buffers. Python sync/async
`fetch_blob_files` and `open_blob_file(s)` accept `buffer_size`; the
default is 512 KiB and zero disables prefetch. Explicit ranges bypass
prefetch. New PE batch-endpoint adoption is separate.

Malformed-length and retained-cache regressions fail on the previous
implementation and pass after the correction. Real HTTP tests cover
buffer propagation, close during blocked input, batch/request
aggregation, and large-object fallback. The final extension was rebuilt
before Python validation. Non-Blob Arrow arrays may still require
alignment copies; a contiguous multi-batch result still requires its
final allocation.

## Benchmark

Peak client RSS increment in KiB, using
`python/benchmarks/bench_fetch_blobs_rss.py`, independent clients and
real loopback HTTP. Both extensions were rebuilt from clean, unchanged
source with the same toolchain/configuration; loaded artifact hashes
were checked. Baseline `42f11d13a2755e44b63876b5a5edd93901291150`, after
`0a8a76de5286949d80368f95a5c529adf422c187`.

| Equivalent fetch workload | Payload | Before | After |
| --- | ---: | ---: | ---: |
| Single response, one batch | 48 MiB | 63,812 | 63,896 |
| Multiple responses, one batch per response | 144 MiB | 423,104 |
195,628 |
| Single response, 12 rows per batch | 48 MiB | 130,520 | 71,348 |
| Multiple responses, 12 rows per batch | 144 MiB | 363,144 | 172,748 |
| One oversized row, Range fallback | 70 MiB | 182,816 | 85,548 |

These are one observation per workload and build, not a statistical
distribution or memory upper bound. The single-batch workload shows no
improvement. Cloud/Sophon deployment, credentialed S3 and long-running
stress have not been exercised.

python/src/table.rs

@@ -1243,6 +1243,15 @@ impl PyBlobFile {
             .map_err(|e| blob_file_error("tell", e))
     }
 
+    fn set_buffer_size(self_: PyRef<'_, Self>, size: usize) -> PyResult<()> {
+        self_.ensure_open()?;
+        let inner = self_.inner.clone();
+        self_
+            .py()
+            .detach(move || block_on(async move { inner.set_buffer_size(size).await }))
+            .map_err(|e| blob_file_error("set_buffer_size", e))
+    }
+
     fn size(self_: PyRef<'_, Self>) -> u64 {
         self_.inner.size()
     }
@@ -2019,11 +2028,12 @@ impl Table {
     }
 
     /// Open lazy blob handles for `row_ids` from blob v2 column `column`.
-    #[pyo3(signature = (column, row_ids))]
+    #[pyo3(signature = (column, row_ids, buffer_size=None))]
     pub fn fetch_blob_files(
         self_: PyRef<'_, Self>,
         column: String,
         row_ids: Vec<u64>,
+        buffer_size: Option<usize>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {
@@ -2031,6 +2041,13 @@ impl Table {
                 .fetch_blob_files(column.as_str(), &row_ids)
                 .await
                 .infer_error()?;
+            if let Some(size) = buffer_size {
+                for file in handles.iter().flatten() {
+                    file.set_buffer_size(size)
+                        .await
+                        .map_err(|e| blob_file_error("set_buffer_size", e))?;
+                }
+            }
             Ok(handles
                 .into_iter()
                 .map(|handle| {
@@ -2043,16 +2060,22 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (descriptor))]
+    #[pyo3(signature = (descriptor, buffer_size=None))]
     pub fn open_blob_file<'py>(
         self_: PyRef<'py, Self>,
         descriptor: &str,
+        buffer_size: Option<usize>,
     ) -> PyResult<Bound<'py, PyAny>> {
         let descriptor: BlobDescriptor = serde_json::from_str(descriptor)
             .map_err(|error| PyValueError::new_err(format!("invalid blob descriptor: {error}")))?;
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {
             let file = inner.open_blob_file(&descriptor).await.infer_error()?;
+            if let Some(size) = buffer_size {
+                file.set_buffer_size(size)
+                    .await
+                    .map_err(|e| blob_file_error("set_buffer_size", e))?;
+            }
             Ok(PyBlobFile {
                 inner: Arc::new(file),
                 closed: AtomicBool::new(false),