perf(scalar): build one decode plan per index scan, not one per batch

rust/lance-index-core/src/scalar.rs

@@ -223,6 +223,45 @@ pub trait IndexReader: Send + Sync {
             futures::stream::once(async move { Ok(batch) }),
         )))
     }
+    /// Stream the entire file as `batch_size`-row batches under a **single** decode plan,
+    /// or `None` if this reader has no such fast path (the caller then falls back to
+    /// per-batch reads).
+    ///
+    /// This exists because reading a file as N independent `read_record_batch` calls
+    /// builds N decode plans, and each plan issues its own small I/O: one read for the
+    /// page metadata it touches and one for the page data, neither of which coalesces with
+    /// the neighbouring batch's. A full scan therefore costs O(batches) round trips. A
+    /// single plan over the whole file schedules every page up front, so the scheduler can
+    /// coalesce those reads into a handful of large requests.
+    ///
+    /// A large index runs to thousands of pages, so this dominates the cost of a full
+    /// scan. Callers that read a file end to end should prefer this method.
+    ///
+    /// [`Self::read_range_stream`] is the same mechanism over a sub-range, but it uses a
+    /// fixed batch size; callers whose batches must line up with the file's own pages
+    /// (a btree page is one batch) need this method.
+    ///
+    /// # Memory profile
+    ///
+    /// Unlike a per-batch read, whose outstanding data is capped by its own readahead,
+    /// a whole-file plan is scheduled eagerly and its peak resident encoded bytes are
+    /// bounded only by the store's scheduler byte budget, which for [`IndexStore`]
+    /// implementations backed by `SchedulerConfig::max_bandwidth` is 32 MiB per I/O
+    /// thread (2 GiB at the cloud default of 64, 256 MiB at the local default of 8).
+    /// `batch_readahead` does not bound this; it bounds decode buffering only. None of
+    /// it is visible to a DataFusion memory pool.
+    ///
+    /// That budget is per store, so a caller that fans out over several stores at once
+    /// (see `BTreeIndex::merge_segments`, which opens one store per source segment)
+    /// multiplies it. Such callers should rescope each store first with
+    /// [`IndexStore::with_io_buffer_size`].
+    async fn whole_file_stream(
+        &self,
+        _batch_size: u32,
+        _batch_readahead: u32,
+    ) -> Result<Option<Pin<Box<dyn RecordBatchStream>>>> {
+        Ok(None)
+    }
     /// Return the number of batches in the file
     async fn num_batches(&self, batch_size: u64) -> u32;
     /// Return the number of rows in the file
@@ -272,6 +311,19 @@ pub trait IndexStore: std::fmt::Debug + Send + Sync + DeepSizeOf {
     /// Suggested I/O parallelism for the store
     fn io_parallelism(&self) -> usize;
 
+    /// Return an equivalent store whose scheduler holds at most `bytes` of outstanding
+    /// prefetched data.
+    ///
+    /// This is a hint, for callers that drive several stores concurrently and want the
+    /// aggregate prefetch to stay near what one store would have used on its own. Stores
+    /// that do not own a scheduler return themselves unchanged. Note that the underlying
+    /// budget is soft: a request is admitted regardless of remaining bytes when nothing
+    /// of lower priority is in flight, which is what guarantees forward progress at any
+    /// budget.
+    fn with_io_buffer_size(&self, _bytes: u64) -> Arc<dyn IndexStore> {
+        self.clone_arc()
+    }
+
     /// Create a new file and return a writer to store data in the file
     async fn new_index_file(&self, name: &str, schema: Arc<Schema>)
     -> Result<Box<dyn IndexWriter>>;