feat(io): bypass backpressure for io_buffer_size=0 and 2.0 indirect I/O (#6627)

## Summary
- When `SchedulerConfig` is created with `io_buffer_size_bytes = 0`, the
byte-based backpressure system is disabled entirely — all I/O proceeds
without waiting for budget. Previously a
zero buffer size would deadlock (every read exceeds a zero budget).
- New env var `LANCE_BYPASS_INDIRECT_IO_BACKPRESSURE`: when set,
indirect I/O in the **2.0 list scheduler** (`indirect_schedule_task` —
items fetched after decoding offsets) bypasses the
backpressure budget. This is surfaced via a new
`EncodingsIo::with_bypass_backpressure()` method implemented by
`LanceEncodingsIo`.
Both the standard and lite schedulers are updated consistently. The IOPS
concurrency limit is unaffected; only the byte-budget gate is bypassed.
## Changed files
| File | Change |
|------|--------|
| `rust/lance-io/src/scheduler.rs` | `no_backpressure` on
`IoQueueState`; `bypass_backpressure` on `IoTask`/`MutableBatch`;
`FileScheduler::with_bypass_backpressure()` |
| `rust/lance-io/src/scheduler/lite.rs` | `no_backpressure` on
`SimpleBackpressureThrottle`; `force_acquire` on `BackpressureThrottle`
trait; `bypass_backpressure` on lite `IoTask` |
| `rust/lance-encoding/src/lib.rs` |
`EncodingsIo::with_bypass_backpressure()` default method |
| `rust/lance-file/src/io.rs` | `LanceEncodingsIo` override of
`with_bypass_backpressure()` |
| `rust/lance-encoding/src/previous/encodings/logical/list.rs` |
`LANCE_BYPASS_INDIRECT_IO_BACKPRESSURE` env var + bypass io wrapping in
indirect schedule |
## Test plan
- [ ] `cargo test -p lance-io --lib` — 155 tests pass (io_uring tests
excluded, require kernel support)
- [ ] `cargo test -p lance-encoding --lib` — 369 tests pass
- [ ] `cargo check --workspace --tests` — clean

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: westonpace

rust/lance-encoding/src/lib.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
-use std::ops::Range;
+use std::{ops::Range, sync::Arc};
 
 use bytes::Bytes;
 use futures::{FutureExt, TryFutureExt, future::BoxFuture};
@@ -75,6 +75,17 @@ pub trait EncodingsIo: std::fmt::Debug + Send + Sync {
             .map_ok(|mut v| v.pop().unwrap())
             .boxed()
     }
+
+    /// Returns a version of this I/O service that bypasses backpressure for all requests.
+    ///
+    /// This is intended for indirect I/O (e.g. fetching items after decoding offsets) where
+    /// blocking on backpressure could cause deadlocks or excessive latency.
+    ///
+    /// Returns `None` if this implementation does not support bypass (e.g. in-memory or test
+    /// schedulers), in which case the caller should fall back to using self.
+    fn with_bypass_backpressure(&self) -> Option<Arc<dyn EncodingsIo>> {
+        None
+    }
 }
 
 /// An implementation of EncodingsIo that serves data from an in-memory buffer

rust/lance-encoding/src/previous/encodings/logical/list.rs

@@ -1,7 +1,11 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
-use std::{collections::VecDeque, ops::Range, sync::Arc};
+use std::{
+    collections::VecDeque,
+    ops::Range,
+    sync::{Arc, OnceLock},
+};
 
 use arrow_array::{
     Array, ArrayRef, BooleanArray, Int32Array, Int64Array, LargeListArray, ListArray, UInt64Array,
@@ -12,7 +16,7 @@ use arrow_array::{
 use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, Buffer, NullBuffer, OffsetBuffer};
 use arrow_schema::{DataType, Field, Fields};
 use futures::{FutureExt, future::BoxFuture};
-use lance_core::{Error, Result, cache::LanceCache};
+use lance_core::{Error, Result, cache::LanceCache, utils::parse::str_is_truthy};
 use log::trace;
 use tokio::task::JoinHandle;
 
@@ -35,6 +39,20 @@ use crate::{
     utils::accumulation::AccumulationQueue,
 };
 
+/// When set, indirect I/O in the 2.0 list scheduler bypasses the backpressure system.
+///
+/// This can be a blunt instrument to avoid deadlocks in 2.0 scenarios
+/// Set LANCE_BYPASS_INDIRECT_IO_BACKPRESSURE=1 to enable.
+static BYPASS_INDIRECT_IO_BACKPRESSURE: OnceLock<bool> = OnceLock::new();
+
+fn bypass_indirect_io_backpressure() -> bool {
+    *BYPASS_INDIRECT_IO_BACKPRESSURE.get_or_init(|| {
+        std::env::var("LANCE_BYPASS_INDIRECT_IO_BACKPRESSURE")
+            .map(|val| str_is_truthy(&val))
+            .unwrap_or(false)
+    })
+}
+
 // Scheduling lists is tricky.  Imagine the following scenario:
 //
 // * There are 2000 offsets per offsets page
@@ -454,7 +472,12 @@ impl SchedulingJob for ListFieldSchedulingJob<'_> {
 
         let items_scheduler = self.scheduler.items_scheduler.clone();
         let items_type = self.scheduler.items_field.data_type().clone();
-        let io = context.io().clone();
+        let base_io = context.io().clone();
+        let io = if bypass_indirect_io_backpressure() {
+            base_io.with_bypass_backpressure().unwrap_or(base_io)
+        } else {
+            base_io
+        };
         let cache = context.cache().clone();
 
         // Immediately spawn the indirect scheduling

rust/lance-file/src/io.rs

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
+use std::sync::Arc;
+
 use futures::{FutureExt, future::BoxFuture};
 use lance_encoding::EncodingsIo;
 use lance_io::scheduler::FileScheduler;
@@ -29,6 +31,13 @@ impl LanceEncodingsIo {
 }
 
 impl EncodingsIo for LanceEncodingsIo {
+    fn with_bypass_backpressure(&self) -> Option<Arc<dyn EncodingsIo>> {
+        Some(Arc::new(Self {
+            scheduler: self.scheduler.with_bypass_backpressure(),
+            read_chunk_size: self.read_chunk_size,
+        }))
+    }
+
     fn submit_request(
         &self,
         ranges: Vec<std::ops::Range<u64>>,