merge

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

vortex-array/src/arrays/struct_/array.rs

@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::fmt::Debug;
 use std::iter::once;
 use std::sync::Arc;
 

vortex-array/src/arrays/variant/vtable/mod.rs

@@ -90,8 +90,8 @@ impl VTable for Variant {
     }
 
     fn serialize(
-        array: ArrayView<'_, Self>,
-        session: &VortexSession,
+        _array: ArrayView<'_, Self>,
+        _session: &VortexSession,
     ) -> VortexResult<Option<Vec<u8>>> {
         Ok(Some(vec![]))
     }

vortex-array/src/arrow/executor/run_end.rs

@@ -24,7 +24,6 @@ use crate::IntoArray;
 use crate::arrays::Constant;
 use crate::arrays::ConstantArray;
 use crate::arrow::ArrowArrayExecutor;
-use crate::session::ArraySessionExt;
 
 /// The encoding ID used by `vortex-runend`. We match on this string to avoid a crate dependency.
 const VORTEX_RUNEND_ID: &str = "vortex.runend";

vortex-layout/src/layouts/chunked/reader.rs

@@ -364,23 +364,26 @@ mod test {
         let segments = Arc::new(TestSegments::default());
         let strategy = ChunkedLayoutStrategy::new(FlatLayoutStrategy::default());
         let (mut sequence_id, eof) = SequenceId::root().split();
-        let layout = block_on(|handle| {
+        let segments2 = Arc::<TestSegments>::clone(&segments);
+        let layout = block_on(|handle| async move {
             let session = SESSION.clone().with_handle(handle);
-            strategy.write_stream(
-                ctx,
-                Arc::<TestSegments>::clone(&segments),
-                SequentialStreamAdapter::new(
-                    DType::Primitive(PType::I32, NonNullable),
-                    stream::iter([
-                        Ok((sequence_id.advance(), buffer![1, 2, 3].into_array())),
-                        Ok((sequence_id.advance(), buffer![4, 5, 6].into_array())),
-                        Ok((sequence_id.advance(), buffer![7, 8, 9].into_array())),
-                    ]),
+            strategy
+                .write_stream(
+                    ctx,
+                    segments2,
+                    SequentialStreamAdapter::new(
+                        DType::Primitive(PType::I32, NonNullable),
+                        stream::iter([
+                            Ok((sequence_id.advance(), buffer![1, 2, 3].into_array())),
+                            Ok((sequence_id.advance(), buffer![4, 5, 6].into_array())),
+                            Ok((sequence_id.advance(), buffer![7, 8, 9].into_array())),
+                        ]),
+                    )
+                    .sendable(),
+                    eof,
+                    &session,
                 )
-                .sendable(),
-                eof,
-                &session,
-            )
+                .await
         })
         .unwrap();
 

vortex-layout/src/layouts/dict/reader.rs

@@ -275,10 +275,15 @@ mod tests {
     use vortex_array::expr::not;
     use vortex_array::expr::pack;
     use vortex_array::expr::root;
+    use vortex_array::scalar_fn::session::ScalarFnSession;
+    use vortex_array::session::ArraySession;
     use vortex_array::validity::Validity;
     use vortex_error::VortexExpect;
+    use vortex_io::runtime::Handle;
     use vortex_io::runtime::single::block_on;
+    use vortex_io::session::RuntimeSession;
     use vortex_io::session::RuntimeSessionExt;
+    use vortex_session::VortexSession;
 
     use crate::LayoutId;
     use crate::LayoutRef;
@@ -291,12 +296,23 @@ mod tests {
     use crate::sequence::SequentialArrayStreamExt;
     use crate::sequence::SequentialStreamAdapter;
     use crate::sequence::SequentialStreamExt;
-    use crate::test::SESSION;
+    use crate::session::LayoutSession;
+
+    // FIXME(ngates): Deprecate the global `runtime::single::block_on` helper and require tests
+    // to call `block_on` on an explicit runtime instance.
+    fn session_with_handle(handle: Handle) -> VortexSession {
+        VortexSession::empty()
+            .with::<ArraySession>()
+            .with::<LayoutSession>()
+            .with::<ScalarFnSession>()
+            .with::<RuntimeSession>()
+            .with_handle(handle)
+    }
 
     #[test]
     fn reading_nested_packs_works() {
         block_on(|handle| async move {
-            let session = SESSION.clone().with_handle(handle);
+            let session = session_with_handle(handle);
             let strategy = DictStrategy::new(
                 FlatLayoutStrategy::default(),
                 FlatLayoutStrategy::default(),
@@ -347,7 +363,7 @@ mod tests {
             );
             assert!(layout.encoding_id() == LayoutId::new_ref("vortex.dict"));
             let actual = layout
-                .new_reader("".into(), segments, &SESSION)
+                .new_reader("".into(), segments, &session)
                 .unwrap()
                 .projection_evaluation(
                     &(0..layout.row_count()),
@@ -395,7 +411,7 @@ mod tests {
         #[case] expected: Vec<bool>,
     ) {
         block_on(|handle| async move {
-            let session = SESSION.clone().with_handle(handle);
+            let session = session_with_handle(handle);
             let strategy = DictStrategy::new(
                 FlatLayoutStrategy::default(),
                 FlatLayoutStrategy::default(),
@@ -431,7 +447,7 @@ mod tests {
                 )),
             );
             let mask = layout
-                .new_reader("".into(), segments, &SESSION)
+                .new_reader("".into(), segments, &session)
                 .unwrap()
                 .filter_evaluation(&(0..3), &filter, MaskFuture::new_true(3))
                 .unwrap()
@@ -445,7 +461,7 @@ mod tests {
     #[test]
     fn reading_is_null_works() {
         block_on(|handle| async move {
-            let session = SESSION.clone().with_handle(handle);
+            let session = session_with_handle(handle);
             let strategy = DictStrategy::new(
                 FlatLayoutStrategy::default(),
                 FlatLayoutStrategy::default(),
@@ -491,7 +507,7 @@ mod tests {
             let expression = not(is_null(root())); // easier to test not_is_null b/c that's the validity array
             assert_eq!(layout.encoding_id(), LayoutId::new_ref("vortex.dict"));
             let actual = layout
-                .new_reader("".into(), segments, &SESSION)
+                .new_reader("".into(), segments, &session)
                 .unwrap()
                 .projection_evaluation(
                     &(0..layout.row_count()),

vortex-layout/src/layouts/flat/writer.rs

@@ -161,7 +161,7 @@ impl LayoutStrategy for FlatLayoutStrategy {
 
         let buffers = chunk.serialize(
             &ctx,
-            &session,
+            session,
             &SerializeOptions {
                 offset: 0,
                 include_padding: self.include_padding,

vortex-layout/src/layouts/repartition.rs

@@ -386,15 +386,17 @@ mod tests {
         );
 
         let stream = fsl.into_array().to_array_stream().sequenced(ptr);
-        let layout = block_on(|handle| {
+        let layout = block_on(|handle| async move {
             let session = SESSION.clone().with_handle(handle);
-            strategy.write_stream(
-                ctx,
-                Arc::<TestSegments>::clone(&segments),
-                stream,
-                eof,
-                &session,
-            )
+            strategy
+                .write_stream(
+                    ctx,
+                    Arc::<TestSegments>::clone(&segments),
+                    stream,
+                    eof,
+                    &session,
+                )
+                .await
         })?;
 
         // The layout should be a ChunkedLayout with multiple children.
@@ -449,15 +451,17 @@ mod tests {
         );
 
         let stream = elements.into_array().to_array_stream().sequenced(ptr);
-        let layout = block_on(|handle| {
+        let layout = block_on(|handle| async move {
             let session = SESSION.clone().with_handle(handle);
-            strategy.write_stream(
-                ctx,
-                Arc::<TestSegments>::clone(&segments),
-                stream,
-                eof,
-                &session,
-            )
+            strategy
+                .write_stream(
+                    ctx,
+                    Arc::<TestSegments>::clone(&segments),
+                    stream,
+                    eof,
+                    &session,
+                )
+                .await
         })?;
 
         assert_eq!(layout.row_count(), num_elements as u64);