Touch up some of the interfaces and docs in vortex-datafusion (#8485)

## Summary

This PR mostly touches up some of the interfaces, and exposes a few
extra types that might be helpful for users.
1. Makes our type conversion traits public
2. Exposes `DefaultExpressionConverter`, so users can fall back to it if
they need to in their own implementations.
3. `VortexSink` if they want to explore write physical plans
4. Introduces `VortexSource::predicate`, to explore read physical plans.

Signed-off-by: Adam Gutglick <adam@spiraldb.com>

Author: AdamGS

vortex-datafusion/src/convert/exprs.rs

@@ -106,7 +106,7 @@ pub trait ExpressionConvertor: Send + Sync {
     }
 }
 
-/// The default [`ExpressionConvertor`].
+/// The default [`ExpressionConvertor`] implementation.
 #[derive(Default)]
 pub struct DefaultExpressionConvertor {}
 

vortex-datafusion/src/convert/mod.rs

@@ -1,19 +1,32 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+//! Utilities and interface to convert DataFusion types to Vortex types.
+//!
+//! Currently includes:
+//! [`ExpressionConvertor`] - Controls the rewrite of DataFusion expressions to Vortex expressions, and whether they can
+//! be pushed into the underlying scan. A default implementation is provided.
+//! [`FromDataFusion`] - Converts a DataFusion type into a Vortex type infallible.
+//! [TryToDataFusion] - Fallibly converts a Vortex type to a DataFusion type.
+
 use vortex::error::VortexResult;
 
 pub(crate) mod exprs;
 mod scalars;
 pub(crate) mod schema;
 pub(crate) mod stats;
 
+pub use exprs::DefaultExpressionConvertor;
+pub use exprs::ExpressionConvertor;
+
 /// First-party trait for implementing conversion from DataFusion types to Vortex types.
-pub(crate) trait FromDataFusion<D: ?Sized>: Sized {
+pub trait FromDataFusion<D: ?Sized>: Sized {
+    /// Convert to this Vortex type from the input DataFusion type.
     fn from_df(df: &D) -> Self;
 }
 
-/// First-party trait for implementing conversion from Vortex to DataFusion types.
-pub(crate) trait TryToDataFusion<D> {
+/// First-party trait for implementing fallible conversions from Vortex to DataFusion types.
+pub trait TryToDataFusion<D> {
+    /// Try to convert this Vortex type from the input DataFusion type.
     fn try_to_df(&self) -> VortexResult<D>;
 }

vortex-datafusion/src/convert/scalars.rs

@@ -119,10 +119,16 @@ impl TryToDataFusion<ScalarValue> for Scalar {
                     .cloned()
                     .map(|b| Vec::<u8>::from(b.into_inner())),
             ),
-            DType::List(..) => todo!("list scalar conversion"),
-            DType::FixedSizeList(..) => todo!("fixed-size list scalar conversion"),
+            dtype @ DType::List(..) => vortex_bail!(
+                "cannot convert Vortex scalar dtype {dtype} to DataFusion ScalarValue: unsupported scalar type"
+            ),
+            dtype @ DType::FixedSizeList(..) => vortex_bail!(
+                "cannot convert Vortex scalar dtype {dtype} to DataFusion ScalarValue: unsupported scalar type"
+            ),
             DType::Struct(..) => struct_to_df(self)?,
-            DType::Union(..) => todo!("union scalar conversion"),
+            dtype @ DType::Union(..) => vortex_bail!(
+                "cannot convert Vortex scalar dtype {dtype} to DataFusion ScalarValue: unsupported scalar type"
+            ),
             DType::Variant(_) => vortex_bail!("Variant scalars aren't supported with DF"),
             DType::Extension(ext) => {
                 let storage_scalar = self.as_extension().to_storage_scalar();
@@ -809,4 +815,21 @@ mod tests {
         assert!(Scalar::from_df(&df).is_null());
         Ok(())
     }
+
+    #[rstest]
+    #[case::list(Scalar::null(DType::List(
+        Arc::new(DType::Primitive(PType::I32, Nullability::Nullable)),
+        Nullability::Nullable
+    )))]
+    #[case::fixed_size_list(Scalar::null(DType::FixedSizeList(
+        Arc::new(DType::Primitive(PType::I32, Nullability::Nullable)),
+        2,
+        Nullability::Nullable
+    )))]
+    #[case::union(Scalar::null(DType::Union(Nullability::Nullable)))]
+    fn unsupported_vortex_scalars_return_errors(#[case] scalar: Scalar) {
+        let err = scalar.try_to_df().unwrap_err();
+
+        assert!(err.to_string().contains("unsupported scalar type"), "{err}");
+    }
 }

vortex-datafusion/src/lib.rs

@@ -88,14 +88,13 @@ use std::fmt::Debug;
 use datafusion_common::stats::Precision as DFPrecision;
 use vortex::expr::stats::Precision;
 
-mod convert;
+pub mod convert;
 mod persistent;
 pub mod v2;
 
 #[cfg(test)]
 mod tests;
 
-pub use convert::exprs::ExpressionConvertor;
 pub use persistent::*;
 
 /// Extension trait to convert our [`Precision`] to DataFusion's

vortex-datafusion/src/persistent/mod.rs

@@ -36,6 +36,7 @@ pub use access_plan::VortexAccessPlan;
 pub use format::VortexFormat;
 pub use format::VortexFormatFactory;
 pub use format::VortexTableOptions;
+pub use sink::VortexSink;
 pub use source::VortexSource;
 
 #[cfg(test)]

vortex-datafusion/src/persistent/reader.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-//! Factory for creating [`VortexReadAt`] instances from [`PartitionedFile`]s.
+//! Factory for creating [`VortexReadAt`] instances for [`PartitionedFile`]s.
 
 use std::fmt::Debug;
 use std::sync::Arc;

vortex-datafusion/src/persistent/sink.rs

@@ -32,13 +32,15 @@ use vortex::io::VortexWrite;
 use vortex::io::object_store::ObjectStoreWrite;
 use vortex::session::VortexSession;
 
+/// Implements [`DataSink`] for writing Vortex files.
 pub struct VortexSink {
     config: FileSinkConfig,
     schema: SchemaRef,
     session: VortexSession,
 }
 
 impl VortexSink {
+    /// Creates a new [`VortexSink`] instance.
     pub fn new(config: FileSinkConfig, schema: SchemaRef, session: VortexSession) -> Self {
         Self {
             config,

vortex-datafusion/src/persistent/source.rs

@@ -306,6 +306,11 @@ impl VortexSource {
         self
     }
 
+    /// Returns the predicate this source is going to push down
+    pub fn predicate(&self) -> Option<&Arc<dyn PhysicalExpr>> {
+        self.vortex_predicate.as_ref()
+    }
+
     fn create_vortex_opener(
         &self,
         object_store: Arc<dyn ObjectStore>,