fix: list columns fail in blocking decode path

  Summary

  Reading List, LargeList, Binary, or LargeBinary columns via read_stream_projected_blocking (used by the Java JNI LanceFileReader) fails with:

  Caused by: java.io.IOException: LanceError(Internal):
    drain was called on primitive field decoder for data type Int32 on column 2
    but the decoder was never awaited,
    /app/rust/lance-encoding/src/previous/encodings/logical/primitive.rs:348:27

  Full Rust-side error:

  thread 'tokio-runtime-worker' panicked at
    rust/lance-encoding/src/previous/encodings/logical/primitive.rs:348:27:

  Internal error: drain was called on primitive field decoder for data type Int32
  on column 4 but the decoder was never awaited

  Stack:
    lance_encoding::previous::encodings::logical::primitive::PrimitiveFieldDecoder::drain
    lance_encoding::previous::encodings::logical::list::ListPageDecoder::drain
    lance_encoding::previous::encodings::logical::struct::SimpleStructDecoder::drain
    lance_encoding::decoder::BatchDecodeIterator::next_batch_task
    lance_file::reader::FileReader::read_stream_projected_blocking
    lance_jni::file_reader::BlockingFileReader::open_stream
    Java_org_lance_file_LanceFileReader_readAllNative

  Schema:
  col_a: int64
  col_b: list<int32>

  This is caused by two independent bugs:

  Bug 1: tokio::spawn in list scheduler is incompatible with the blocking decode path.

  ListFieldSchedulingJob::schedule_next uses tokio::spawn to run indirect_schedule_task concurrently. This works in the async path where an active tokio runtime drives the spawned task.
  However, in the blocking path (schedule_and_decode_blocking), the scheduling runs outside any tokio runtime context, so the spawned task is never executed, and the JoinHandle cannot be
   awaited from a different runtime.

  Bug 2: BatchDecodeIterator skips wait when all rows are already scheduled.

  For list columns, all rows are marked as "scheduled" after the first schedule_ranges call (because the ListPageDecoder message is sent synchronously), even though the item data has not
   been loaded yet. Starting from the second batch, scheduled_need == 0, so wait_for_io() is skipped entirely. The subsequent drain then encounters PrimitiveFieldDecoder instances whose
  physical decoders were never awaited.

  Changes

  - list.rs: Replace tokio::spawn(indirect_schedule_task(...)) with an inline BoxFuture via .boxed(). This removes the dependency on a tokio runtime during scheduling and makes the
  indirect scheduling work in both async and blocking contexts. The ListPageDecoder.unloaded field type changes from JoinHandle to BoxFuture, and the JoinError handling in
  wait_for_loaded is simplified accordingly.
  - decoder.rs: Add an else branch in BatchDecodeIterator::next_batch_task to call self.root_decoder.wait(loaded_need, ...) even when scheduled_need == 0. This ensures that list item
  data is fully loaded before draining, regardless of the scheduling status.
  - reader.rs: Add test_project_list_int32 regression test covering projected reads of list<int32> columns on both V2_0 and V2_1 file formats.

Author: frankliee

rust/lance-encoding/src/decoder.rs

@@ -1654,6 +1654,13 @@ impl<T: RootDecoderType> BatchDecodeIterator<T> {
                 let under_scheduled = desired_scheduled - actually_scheduled;
                 to_take -= under_scheduled;
             }
+        } else {
+            // All rows are scheduled but we still need to wait for data to be loaded.
+            // This is important for types with indirect I/O (e.g. List, Binary) where
+            // scheduling completes but item data may not yet be loaded.
+            let loaded_need = self.rows_drained + to_take.min(self.rows_per_batch as u64) - 1;
+            self.root_decoder
+                .wait(loaded_need, &self.wait_for_io_runtime)?;
         }
 
         if to_take == 0 {

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

@@ -14,7 +14,6 @@ use arrow_schema::{DataType, Field, Fields};
 use futures::{FutureExt, future::BoxFuture};
 use lance_core::{Error, Result, cache::LanceCache};
 use log::trace;
-use tokio::task::JoinHandle;
 
 use crate::{
     EncodingsIo,
@@ -457,8 +456,9 @@ impl SchedulingJob for ListFieldSchedulingJob<'_> {
         let io = context.io().clone();
         let cache = context.cache().clone();
 
-        // Immediately spawn the indirect scheduling
-        let indirect_fut = tokio::spawn(indirect_schedule_task(
+        // Immediately schedule the indirect I/O (inline future, no tokio::spawn).
+        // This ensures compatibility with both async and blocking decode paths.
+        let indirect_fut = indirect_schedule_task(
             next_offsets_decoder,
             list_reqs,
             null_offset_adjustment,
@@ -467,7 +467,8 @@ impl SchedulingJob for ListFieldSchedulingJob<'_> {
             io,
             cache,
             priority.box_clone(),
-        ));
+        )
+        .boxed();
 
         // Return a decoder
         let decoder = Box::new(ListPageDecoder {
@@ -590,9 +591,8 @@ impl FieldScheduler for ListFieldScheduler {
 /// them.
 ///
 /// TODO: Test the case where a single list page has multiple items pages
-#[derive(Debug)]
 struct ListPageDecoder {
-    unloaded: Option<JoinHandle<Result<IndirectlyLoaded>>>,
+    unloaded: Option<BoxFuture<'static, Result<IndirectlyLoaded>>>,
     // offsets and validity will have already been decoded as part of the indirect I/O
     offsets: Arc<[u64]>,
     validity: BooleanBuffer,
@@ -605,6 +605,18 @@ struct ListPageDecoder {
     data_type: DataType,
 }
 
+impl std::fmt::Debug for ListPageDecoder {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ListPageDecoder")
+            .field("unloaded", &self.unloaded.as_ref().map(|_| "..."))
+            .field("num_rows", &self.num_rows)
+            .field("rows_drained", &self.rows_drained)
+            .field("rows_loaded", &self.rows_loaded)
+            .field("data_type", &self.data_type)
+            .finish()
+    }
+}
+
 struct ListDecodeTask {
     offsets: Vec<u64>,
     validity: BooleanBuffer,
@@ -695,14 +707,7 @@ impl LogicalPageDecoder for ListPageDecoder {
             // I/O and then wait for enough items to arrive
             if self.unloaded.is_some() {
                 trace!("List scheduler needs to wait for indirect I/O to complete");
-                let indirectly_loaded = self.unloaded.take().unwrap().await;
-                if let Err(err) = indirectly_loaded {
-                    match err.try_into_panic() {
-                        Ok(err) => std::panic::resume_unwind(err),
-                        Err(err) => panic!("{:?}", err),
-                    };
-                }
-                let indirectly_loaded = indirectly_loaded.unwrap()?;
+                let indirectly_loaded = self.unloaded.take().unwrap().await?;
 
                 self.offsets = indirectly_loaded.offsets;
                 self.validity = indirectly_loaded.validity;

rust/lance-file/src/reader.rs

@@ -1916,6 +1916,112 @@ mod tests {
         );
     }
 
+    /// Reproduce: projecting a list<int32> column from a multi-column v2.0 file
+    /// triggers "drain was called on primitive field decoder … but the decoder was
+    /// never awaited".
+    #[rstest]
+    #[tokio::test]
+    async fn test_project_list_int32(
+        #[values(LanceFileVersion::V2_0, LanceFileVersion::V2_1)] version: LanceFileVersion,
+    ) {
+        use arrow_array::builder::{Int32Builder, Int64Builder, ListBuilder};
+
+        let fs = FsFixture::default();
+
+        // Schema: col_a (int64) + col_b (list<int32>)
+        let arrow_schema = Arc::new(ArrowSchema::new(vec![
+            Field::new("col_a", DataType::Int64, false),
+            Field::new(
+                "col_b",
+                DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
+                false,
+            ),
+        ]));
+
+        // Generate enough rows with large lists to exercise multi-page items
+        let num_rows = 10000_usize;
+        let items_per_row = 500_usize;
+        let mut col_a_builder = Int64Builder::with_capacity(num_rows);
+        let inner_builder = Int32Builder::new();
+        let mut list_builder = ListBuilder::new(inner_builder);
+
+        for i in 0..num_rows {
+            col_a_builder.append_value(i as i64);
+            for j in 0..items_per_row as i32 {
+                list_builder.values().append_value(i as i32 * 100 + j);
+            }
+            list_builder.append(true);
+        }
+
+        let batch = RecordBatch::try_new(
+            arrow_schema.clone(),
+            vec![
+                Arc::new(col_a_builder.finish()),
+                Arc::new(list_builder.finish()),
+            ],
+        )
+        .unwrap();
+
+        let reader =
+            arrow_array::RecordBatchIterator::new(vec![Ok(batch.clone())], arrow_schema.clone());
+        let written = write_lance_file(
+            reader,
+            &fs,
+            FileWriterOptions {
+                format_version: Some(version),
+                ..Default::default()
+            },
+        )
+        .await;
+
+        let file_scheduler = fs
+            .scheduler
+            .open_file(&fs.tmp_path, &CachedFileSize::unknown())
+            .await
+            .unwrap();
+
+        // Test: project only list column, scalar + list, and all columns
+        for col in [vec!["col_b"], vec!["col_a", "col_b"]] {
+            let file_reader = FileReader::try_open(
+                file_scheduler.clone(),
+                None,
+                Arc::<DecoderPlugins>::default(),
+                &test_cache(),
+                FileReaderOptions::default(),
+            )
+            .await
+            .unwrap();
+
+            let projection = ReaderProjection::from_column_names(
+                file_reader.metadata.version(),
+                &written.schema,
+                &col,
+            )
+            .unwrap();
+
+            let batch_stream = file_reader
+                .read_stream_projected(
+                    lance_io::ReadBatchParams::RangeFull,
+                    512,
+                    16,
+                    projection.clone(),
+                    FilterExpression::no_filter(),
+                )
+                .unwrap();
+
+            let projection_arrow = ArrowSchema::from(projection.schema.as_ref());
+            verify_expected(
+                &written.data,
+                batch_stream,
+                512,
+                Some(Box::new(move |b: &RecordBatch| {
+                    b.project_by_schema(&projection_arrow).unwrap()
+                })),
+            )
+            .await;
+        }
+    }
+
     #[test_log::test(tokio::test)]
     async fn test_compressing_buffer() {
         let fs = FsFixture::default();