Clarify docs in session accessors

Signed-off-by: Nicholas Gates <[email protected]>

Author: gatesn

vortex-array/src/arrays/filter/kernel.rs

@@ -48,8 +48,20 @@ impl Kernel for FilterKernel {
 }
 
 fn cost_for_dtype(dtype: &DType, selection: &Mask) -> f64 {
+    if selection.true_count() == 0 {
+        return 0.0;
+    }
+
     match dtype {
         DType::Null => 0.0,
+        DType::Primitive(ptype, _) => {
+            // Some estimate based on PType width and true-count or number of slices??
+            // Maybe nullable types are more expensive?
+            // ... No idea how to tune this yet.
+            let width = ptype.byte_width() as f64;
+            let selectivity = selection.true_count() as f64 / selection.len() as f64;
+            width * selectivity
+        }
         DType::Extension(ext) => cost_for_dtype(ext.storage_dtype(), selection),
         _ => f64::INFINITY,
     }

vortex-array/src/arrays/scalar_fn/rules.rs

@@ -3,19 +3,19 @@
 
 use vortex_error::VortexResult;
 use vortex_scalar::Scalar;
+use vortex_vector::scalar_matches_dtype;
 use vortex_vector::Datum;
 use vortex_vector::VectorOps;
-use vortex_vector::scalar_matches_dtype;
 
-use crate::Array;
-use crate::ArrayRef;
-use crate::IntoArray;
 use crate::arrays::AnyScalarFn;
 use crate::arrays::ConstantArray;
 use crate::arrays::ConstantVTable;
 use crate::arrays::ScalarFnArray;
 use crate::expr::ExecutionArgs;
 use crate::optimizer::rules::ArrayReduceRule;
+use crate::Array;
+use crate::ArrayRef;
+use crate::IntoArray;
 
 #[derive(Debug)]
 pub(crate) struct ScalarFnConstantRule;
@@ -56,6 +56,7 @@ impl ArrayReduceRule<AnyScalarFn> for ScalarFnConstantRule {
         };
         assert!(scalar_matches_dtype(&result, &array.dtype));
 
+        let _fn = format!("{}", array.scalar_fn);
         Ok(Some(
             ConstantArray::new(Scalar::from_vector_scalar(result, &array.dtype)?, array.len)
                 .into_array(),

vortex-array/src/arrays/scalar_fn/vtable/canonical.rs

@@ -4,15 +4,15 @@
 use itertools::Itertools;
 use vortex_error::VortexExpect;
 
-use crate::Array;
-use crate::Canonical;
-use crate::arrays::LEGACY_SESSION;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::vtable::ScalarFnVTable;
+use crate::arrays::LEGACY_SESSION;
 use crate::executor::VectorExecutor;
 use crate::expr::ExecutionArgs;
 use crate::vectors::VectorIntoArray;
 use crate::vtable::CanonicalVTable;
+use crate::Array;
+use crate::Canonical;
 
 impl CanonicalVTable<ScalarFnVTable> for ScalarFnVTable {
     fn canonicalize(array: &ScalarFnArray) -> Canonical {

vortex-array/src/executor.rs

@@ -1,18 +1,18 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
+use vortex_error::VortexResult;
 use vortex_session::VortexSession;
 use vortex_vector::Datum;
 use vortex_vector::Vector;
 use vortex_vector::VectorOps;
 
-use crate::Array;
-use crate::ArrayRef;
 use crate::arrays::ConstantVTable;
 use crate::kernel::BindCtx;
 use crate::session::ArraySessionExt;
+use crate::Array;
+use crate::ArrayRef;
 
 /// Executor for exporting a Vortex [`Vector`] or [`Datum`] from an [`ArrayRef`].
 pub trait VectorExecutor {

vortex-array/src/expr/exprs/get_item.rs

@@ -9,18 +9,19 @@ use vortex_dtype::DType;
 use vortex_dtype::FieldName;
 use vortex_dtype::FieldPath;
 use vortex_dtype::Nullability;
+use vortex_error::vortex_err;
+use vortex_error::vortex_panic;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_error::vortex_err;
 use vortex_mask::Mask;
 use vortex_proto::expr as pb;
 use vortex_vector::Datum;
 use vortex_vector::ScalarOps;
 use vortex_vector::VectorOps;
 
-use crate::ArrayRef;
-use crate::ToCanonical;
 use crate::compute::mask;
+use crate::expr::exprs::root::root;
+use crate::expr::stats::Stat;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
@@ -31,8 +32,9 @@ use crate::expr::SimplifyCtx;
 use crate::expr::StatsCatalog;
 use crate::expr::VTable;
 use crate::expr::VTableExt;
-use crate::expr::exprs::root::root;
-use crate::expr::stats::Stat;
+use crate::scalar_fns::ExprBuiltins;
+use crate::ArrayRef;
+use crate::ToCanonical;
 
 pub struct GetItem;
 
@@ -139,9 +141,10 @@ impl VTable for GetItem {
         &self,
         field_name: &FieldName,
         expr: &Expression,
-        _ctx: &dyn SimplifyCtx,
+        ctx: &dyn SimplifyCtx,
     ) -> VortexResult<Option<Expression>> {
         let child = expr.child(0);
+        let child_dtype = ctx.return_dtype(child)?;
 
         // If the child is a Pack expression, we can directly return the corresponding child.
         if let Some(pack) = child.as_opt::<Pack>() {
@@ -157,7 +160,15 @@ impl VTable for GetItem {
                     )
                 })?;
 
-            return Ok(Some(child.child(idx).clone()));
+            let mut field = child.child(idx).clone();
+
+            // If the pack expression is nullable but the child field is not, we need to
+            // adjust the nullability of the resulting expression.
+            if pack.nullability.is_nullable() && !child_dtype.is_nullable() {
+                field = field.cast(child_dtype.as_nullable())?;
+            }
+
+            return Ok(Some(field));
         }
 
         Ok(None)
@@ -232,15 +243,15 @@ mod tests {
     use vortex_dtype::StructFields;
     use vortex_scalar::Scalar;
 
-    use crate::Array;
-    use crate::IntoArray;
     use crate::arrays::StructArray;
     use crate::expr::exprs::binary::checked_add;
     use crate::expr::exprs::get_item::get_item;
     use crate::expr::exprs::literal::lit;
     use crate::expr::exprs::pack::pack;
     use crate::expr::exprs::root::root;
     use crate::validity::Validity;
+    use crate::Array;
+    use crate::IntoArray;
 
     fn test_array() -> StructArray {
         StructArray::from_fields(&[

vortex-array/src/vtable/mod.rs

@@ -28,14 +28,14 @@ use vortex_dtype::DType;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
 
+use crate::kernel::kernel;
+use crate::kernel::BindCtx;
+use crate::kernel::KernelRef;
+use crate::serde::ArrayChildren;
 use crate::Array;
 use crate::ArrayRef;
 use crate::IntoArray;
 use crate::VectorExecutor;
-use crate::kernel::BindCtx;
-use crate::kernel::KernelRef;
-use crate::kernel::kernel;
-use crate::serde::ArrayChildren;
 
 /// The array [`VTable`] encapsulates logic for an Array type within Vortex.
 ///
@@ -149,7 +149,7 @@ pub trait VTable: 'static + Sized + Send + Sync + Debug {
         let session = ctx.session().clone();
         Ok(kernel(move || {
             let canonical = Self::CanonicalVTable::canonicalize(&array);
-            canonical.into_array().execute_vector_optimized(&session)
+            canonical.into_array().execute_vector(&session)
         }))
     }
 

vortex-compute/src/arrow/primitive.rs

@@ -3,30 +3,31 @@
 
 use std::sync::Arc;
 
-use arrow_array::Array;
-use arrow_array::ArrayRef;
-use arrow_array::PrimitiveArray;
 use arrow_array::types::Float16Type;
 use arrow_array::types::Float32Type;
 use arrow_array::types::Float64Type;
-use arrow_array::types::Int8Type;
 use arrow_array::types::Int16Type;
 use arrow_array::types::Int32Type;
 use arrow_array::types::Int64Type;
-use arrow_array::types::UInt8Type;
+use arrow_array::types::Int8Type;
 use arrow_array::types::UInt16Type;
 use arrow_array::types::UInt32Type;
 use arrow_array::types::UInt64Type;
+use arrow_array::types::UInt8Type;
+use arrow_array::Array;
+use arrow_array::ArrayRef;
+use arrow_array::PrimitiveArray;
 use vortex_buffer::Buffer;
 use vortex_dtype::half::f16;
+use vortex_error::vortex_panic;
 use vortex_error::VortexResult;
 use vortex_vector::match_each_pvector;
 use vortex_vector::primitive::PVector;
 use vortex_vector::primitive::PrimitiveVector;
 
+use crate::arrow::nulls_to_mask;
 use crate::arrow::IntoArrow;
 use crate::arrow::IntoVector;
-use crate::arrow::nulls_to_mask;
 
 impl IntoArrow for PrimitiveVector {
     type Output = ArrayRef;

vortex-datafusion/src/persistent/opener.rs

@@ -9,28 +9,28 @@ use arrow_schema::ArrowError;
 use arrow_schema::DataType;
 use arrow_schema::Field;
 use arrow_schema::SchemaRef;
+use datafusion_common::arrow::array::RecordBatch;
 use datafusion_common::DataFusionError;
 use datafusion_common::Result as DFResult;
-use datafusion_common::arrow::array::RecordBatch;
-use datafusion_datasource::FileRange;
-use datafusion_datasource::PartitionedFile;
 use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::FileOpenFuture;
 use datafusion_datasource::file_stream::FileOpener;
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
-use datafusion_physical_expr::PhysicalExprRef;
+use datafusion_datasource::FileRange;
+use datafusion_datasource::PartitionedFile;
 use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
 use datafusion_physical_expr::split_conjunction;
+use datafusion_physical_expr::PhysicalExprRef;
 use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_plan::metrics::Count;
 use datafusion_pruning::FilePruner;
+use futures::stream;
 use futures::FutureExt;
 use futures::StreamExt;
 use futures::TryStreamExt;
-use futures::stream;
-use object_store::ObjectStore;
 use object_store::path::Path;
+use object_store::ObjectStore;
 use tracing::Instrument;
 use vortex::array::ArrayRef;
 use vortex::dtype::FieldName;
@@ -334,8 +334,8 @@ impl FileOpener for VortexOpener {
 
             let stream = scan_builder
                 .with_metrics(metrics)
-                .with_projection(projection_expr)
-                .with_some_filter(filter)
+                .with_projection(projection_expr.clone())
+                .with_some_filter(filter.clone())
                 .with_ordered(has_output_ordering)
                 .map(|chunk| RecordBatch::try_from(chunk.as_ref()))
                 .into_stream()
@@ -428,15 +428,15 @@ mod tests {
     use datafusion::scalar::ScalarValue;
     use insta::assert_snapshot;
     use itertools::Itertools;
-    use object_store::ObjectMeta;
     use object_store::memory::InMemory;
+    use object_store::ObjectMeta;
     use rstest::rstest;
-    use vortex::VortexSessionDefault;
     use vortex::array::arrow::FromArrowArray;
     use vortex::file::WriteOptionsSessionExt;
     use vortex::io::ObjectStoreWriter;
     use vortex::io::VortexWrite;
     use vortex::session::VortexSession;
+    use vortex::VortexSessionDefault;
 
     use super::*;
 

vortex-io/src/runtime/handle.rs

@@ -4,15 +4,15 @@
 use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::Weak;
+use std::task::ready;
 use std::task::Context;
 use std::task::Poll;
-use std::task::ready;
 
+use futures::channel::mpsc;
 use futures::FutureExt;
 use futures::StreamExt;
-use futures::channel::mpsc;
-use vortex_error::VortexResult;
 use vortex_error::vortex_panic;
+use vortex_error::VortexResult;
 use vortex_metrics::VortexMetrics;
 
 use crate::file::FileRead;
@@ -186,18 +186,19 @@ impl<T> Task<T> {
 impl<T> Future for Task<T> {
     type Output = T;
 
+    #[allow(clippy::panic)]
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         match ready!(self.recv.poll_unpin(cx)) {
             Ok(result) => Poll::Ready(result),
-            Err(recv_err) => {
+            Err(_recv_err) => {
                 // If the other end of the channel was dropped, it means the runtime dropped
                 // the future without ever completing it. If the caller aborted this task by
                 // dropping it, then they wouldn't be able to poll it anymore.
                 // So we consider a closed channel to be a Runtime programming error and therefore
                 // we panic.
-                vortex_panic!(
-                    "Runtime dropped task without completing it, likely it panicked: {recv_err}"
-                )
+
+                // NOTE(ngates): we don't use vortex_panic to avoid printing a useless backtrace.
+                panic!("Runtime dropped task without completing it, likely it panicked")
             }
         }
     }