feat: object store decides scheduler type (#6373)

## Summary

- Add `ObjectStore::prefers_lite_scheduler()` that returns `true` for
`file+uring://` stores, so the lite scheduler is used automatically
without needing the env var
- Change `SchedulerConfig::use_lite_scheduler` from `bool` to
`Option<bool>` — `Some(true/false)` overrides, `None` defers to the
object store's preference
- `LANCE_USE_LITE_SCHEDULER` env var still works as an override when
explicitly set

## Test plan

- [x] `cargo check -p lance-io --tests --benches` compiles cleanly
- [x] `cargo test -p lance-io` — all 148 tests pass
- [x] `cargo clippy -p lance-io --tests --benches -- -D warnings` — no
warnings

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

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

Author: westonpace

rust/lance-io/benches/scheduler.rs

@@ -22,14 +22,14 @@ use pprof::criterion::{Output, PProfProfiler};
 struct FullReadParams {
     io_parallelism: u32,
     page_size: u64,
-    use_lite_scheduler: bool,
+    use_lite_scheduler: Option<bool>,
 }
 
 impl Display for FullReadParams {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "full_read,parallel={},read_size={},use_lite_scheduler={}",
+            "full_read,parallel={},read_size={},use_lite_scheduler={:?}",
             self.io_parallelism, self.page_size, self.use_lite_scheduler
         )
     }
@@ -74,7 +74,7 @@ fn bench_full_read(c: &mut Criterion) {
     let runtime = Runtime::new().unwrap();
     let (obj_store, tmp_file) = runtime.block_on(create_data(DATA_SIZE));
 
-    for use_lite_scheduler in [false, true] {
+    for use_lite_scheduler in [Some(false), Some(true)] {
         for io_parallelism in [1, 16] {
             for page_size in [4096, 1024 * 1024] {
                 let params = FullReadParams {
@@ -101,10 +101,10 @@ fn bench_full_read(c: &mut Criterion) {
                             unsafe {
                                 std::env::set_var("IO_THREADS", io_parallelism.to_string());
                             }
-                            let mut config = SchedulerConfig::default_for_testing();
-                            if use_lite_scheduler {
-                                config = config.with_lite_scheduler();
-                            }
+                            let config = SchedulerConfig {
+                                use_lite_scheduler,
+                                ..SchedulerConfig::default_for_testing()
+                            };
                             runtime.block_on(async {
                                 let scheduler = ScanScheduler::new(obj_store, config);
                                 let file_scheduler = scheduler
@@ -142,15 +142,15 @@ struct RandomReadParams {
     io_parallelism: u32,
     item_size: u32,
     indices: Arc<Vec<u32>>,
-    use_lite_scheduler: bool,
+    use_lite_scheduler: Option<bool>,
     noisy_runtime: bool,
 }
 
 impl Display for RandomReadParams {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "random_read,parallel={},item_size={},use_lite_scheduler={},noisy={}",
+            "random_read,parallel={},item_size={},use_lite_scheduler={:?},noisy={}",
             self.io_parallelism, self.item_size, self.use_lite_scheduler, self.noisy_runtime
         )
     }
@@ -187,7 +187,7 @@ fn bench_random_read(c: &mut Criterion) {
     ));
 
     for noisy_runtime in [false, true] {
-        for use_lite_scheduler in [false, true] {
+        for use_lite_scheduler in [Some(false), Some(true)] {
             for io_parallelism in [1, 16] {
                 for item_size in [4096, 32 * 1024] {
                     let runtime = Runtime::new().unwrap();
@@ -240,10 +240,10 @@ fn bench_random_read(c: &mut Criterion) {
                                         }
                                     }
 
-                                    let mut config = SchedulerConfig::default_for_testing();
-                                    if use_lite_scheduler {
-                                        config = config.with_lite_scheduler();
-                                    }
+                                    let config = SchedulerConfig {
+                                        use_lite_scheduler,
+                                        ..SchedulerConfig::default_for_testing()
+                                    };
                                     let scheduler = ScanScheduler::new(obj_store, config);
                                     let file_scheduler = scheduler
                                         .open_file(&tmp_file, &CachedFileSize::unknown())

rust/lance-io/src/object_store.rs

@@ -534,11 +534,19 @@ impl ObjectStore {
 
     /// Returns true if the object store pointed to a local file system.
     pub fn is_local(&self) -> bool {
-        self.scheme == "file"
+        self.scheme == "file" || self.scheme == "file+uring"
     }
 
     pub fn is_cloud(&self) -> bool {
-        self.scheme != "file" && self.scheme != "memory"
+        !self.is_local() && self.scheme != "memory"
+    }
+
+    /// Whether this object store prefers the lite scheduler.
+    ///
+    /// The lite scheduler is designed for backends like io_uring where
+    /// tasks should only be polled when the consumer polls them.
+    pub fn prefers_lite_scheduler(&self) -> bool {
+        self.scheme == "file+uring"
     }
 
     pub fn scheme(&self) -> &str {

rust/lance-io/src/scheduler.rs

@@ -15,6 +15,7 @@ use std::sync::{Arc, Mutex};
 use std::time::Instant;
 use tokio::sync::Notify;
 
+use lance_core::utils::parse::str_is_truthy;
 use lance_core::{Error, Result};
 
 use crate::object_store::ObjectStore;
@@ -505,23 +506,29 @@ pub struct SchedulerConfig {
     /// This controls back pressure.  If data is not processed quickly enough then this
     /// buffer will fill up and the I/O loop will pause until the buffer is drained.
     pub io_buffer_size_bytes: u64,
-    /// Whether to use the new lite scheduler
-    pub use_lite_scheduler: bool,
+    /// Whether to use the lite scheduler.
+    ///
+    /// - `Some(true)` forces the lite scheduler (e.g. from env var or programmatic).
+    /// - `Some(false)` forces the standard scheduler.
+    /// - `None` defers to the object store's preference (see [`ObjectStore::prefers_lite_scheduler`]).
+    pub use_lite_scheduler: Option<bool>,
 }
 
 impl SchedulerConfig {
     pub fn new(io_buffer_size_bytes: u64) -> Self {
         Self {
             io_buffer_size_bytes,
-            use_lite_scheduler: std::env::var("LANCE_USE_LITE_SCHEDULER").is_ok(),
+            use_lite_scheduler: std::env::var("LANCE_USE_LITE_SCHEDULER")
+                .ok()
+                .map(|v| str_is_truthy(v.trim())),
         }
     }
 
     /// Big enough for unit testing
     pub fn default_for_testing() -> Self {
         Self {
             io_buffer_size_bytes: 256 * 1024 * 1024,
-            use_lite_scheduler: false,
+            use_lite_scheduler: None,
         }
     }
 
@@ -533,7 +540,7 @@ impl SchedulerConfig {
 
     pub fn with_lite_scheduler(self) -> Self {
         Self {
-            use_lite_scheduler: true,
+            use_lite_scheduler: Some(true),
             ..self
         }
     }
@@ -548,7 +555,10 @@ impl ScanScheduler {
     /// * config - configuration settings for the scheduler
     pub fn new(object_store: Arc<ObjectStore>, config: SchedulerConfig) -> Arc<Self> {
         let io_capacity = object_store.io_parallelism();
-        let io_queue = if config.use_lite_scheduler {
+        let use_lite = config
+            .use_lite_scheduler
+            .unwrap_or_else(|| object_store.prefers_lite_scheduler());
+        let io_queue = if use_lite {
             let io_queue = Arc::new(lite::IoQueue::new(
                 io_capacity as u64,
                 config.io_buffer_size_bytes,
@@ -743,6 +753,11 @@ impl ScanScheduler {
     pub fn stats(&self) -> ScanStats {
         ScanStats::new(self.stats.as_ref())
     }
+
+    #[cfg(test)]
+    fn uses_lite_scheduler(&self) -> bool {
+        matches!(self.io_queue, IoQueueType::Lite(_))
+    }
 }
 
 impl Drop for ScanScheduler {
@@ -1120,7 +1135,7 @@ mod tests {
 
         let config = SchedulerConfig {
             io_buffer_size_bytes: 1024 * 1024,
-            use_lite_scheduler: false,
+            use_lite_scheduler: None,
         };
 
         let scan_scheduler = ScanScheduler::new(obj_store, config);
@@ -1211,7 +1226,7 @@ mod tests {
 
         let config = SchedulerConfig {
             io_buffer_size_bytes: 10,
-            use_lite_scheduler: false,
+            use_lite_scheduler: None,
         };
 
         let scan_scheduler = ScanScheduler::new(obj_store.clone(), config);
@@ -1286,7 +1301,7 @@ mod tests {
         // Ensure deadlock prevention timeout can be disabled
         let config = SchedulerConfig {
             io_buffer_size_bytes: 10,
-            use_lite_scheduler: false,
+            use_lite_scheduler: None,
         };
 
         let scan_scheduler = ScanScheduler::new(obj_store, config);
@@ -1374,6 +1389,55 @@ mod tests {
         assert_eq!(fut3.await.unwrap()[0].len(), 100);
     }
 
+    #[tokio::test]
+    async fn test_object_store_selects_scheduler() {
+        // A memory:// store should use the standard scheduler when config is None
+        let memory_store = Arc::new(ObjectStore::memory());
+        assert!(!memory_store.prefers_lite_scheduler());
+        let config = SchedulerConfig {
+            io_buffer_size_bytes: 256 * 1024 * 1024,
+            use_lite_scheduler: None,
+        };
+        let scheduler = ScanScheduler::new(memory_store.clone(), config);
+        assert!(!scheduler.uses_lite_scheduler());
+
+        // A file+uring:// store should use the lite scheduler when config is None
+        let uring_store = Arc::new(ObjectStore::new(
+            Arc::new(InMemory::new()),
+            Url::parse("file+uring:///tmp").unwrap(),
+            None,
+            None,
+            false,
+            false,
+            8,
+            DEFAULT_DOWNLOAD_RETRY_COUNT,
+            None,
+        ));
+        assert!(uring_store.prefers_lite_scheduler());
+        let config = SchedulerConfig {
+            io_buffer_size_bytes: 256 * 1024 * 1024,
+            use_lite_scheduler: None,
+        };
+        let scheduler = ScanScheduler::new(uring_store.clone(), config);
+        assert!(scheduler.uses_lite_scheduler());
+
+        // Explicit Some(false) overrides a file+uring:// store's preference
+        let config = SchedulerConfig {
+            io_buffer_size_bytes: 256 * 1024 * 1024,
+            use_lite_scheduler: Some(false),
+        };
+        let scheduler = ScanScheduler::new(uring_store, config);
+        assert!(!scheduler.uses_lite_scheduler());
+
+        // Explicit Some(true) overrides a memory:// store's preference
+        let config = SchedulerConfig {
+            io_buffer_size_bytes: 256 * 1024 * 1024,
+            use_lite_scheduler: Some(true),
+        };
+        let scheduler = ScanScheduler::new(memory_store, config);
+        assert!(scheduler.uses_lite_scheduler());
+    }
+
     #[test_log::test(tokio::test(flavor = "multi_thread"))]
     async fn stress_backpressure() {
         // This test ensures that the backpressure mechanism works correctly with
@@ -1389,7 +1453,7 @@ mod tests {
         // Only one request will be allowed in
         let config = SchedulerConfig {
             io_buffer_size_bytes: 1,
-            use_lite_scheduler: false,
+            use_lite_scheduler: None,
         };
         let scan_scheduler = ScanScheduler::new(obj_store.clone(), config);
         let file_scheduler = scan_scheduler