Have vortex-datafusion depend only on vortex crate (#3438)

Also removes the in-memory DataFusion execution plans.

---------

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

Author: gatesn

Cargo.lock

@@ -11,7 +11,6 @@ use indicatif::ProgressBar;
 use itertools::Itertools;
 use tokio::runtime::Runtime;
 use vortex::buffer::{Buffer, buffer};
-use vortex::error::VortexUnwrap;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -75,11 +74,7 @@ fn random_access(
             &runtime,
             iterations,
             || indices.clone(),
-            |indices| async {
-                take_vortex_tokio(&taxi_vortex, indices)
-                    .await
-                    .vortex_unwrap()
-            },
+            |indices| async { take_vortex_tokio(&taxi_vortex, indices).await },
         ),
     });
     progress.inc(1);
@@ -93,7 +88,7 @@ fn random_access(
                 &runtime,
                 iterations,
                 || indices.clone(),
-                |indices| async { take_parquet(&taxi_parquet, indices).await.vortex_unwrap() },
+                |indices| async { take_parquet(&taxi_parquet, indices).await },
             ),
         });
         progress.inc(1);

bench-vortex/src/bin/random_access.rs

@@ -222,10 +222,7 @@ pub async fn register_vortex_files(
     let config = if let Some(schema) = schema {
         config.with_schema(schema.into())
     } else {
-        config
-            .infer_schema(&session.state())
-            .await
-            .vortex_expect("cannot infer schema")
+        config.infer_schema(&session.state()).await?
     };
 
     let listing_table = Arc::new(ListingTable::try_new(config)?);

bench-vortex/src/clickbench.rs

@@ -11,7 +11,7 @@ use vortex::error::VortexResult;
 use vortex::file::{VortexLayoutStrategy, VortexOpenOptions, VortexWriteOptions};
 
 #[inline(never)]
-pub async fn vortex_compress_write(array: &dyn Array, buf: &mut Vec<u8>) -> VortexResult<u64> {
+pub async fn vortex_compress_write(array: &dyn Array, buf: &mut Vec<u8>) -> anyhow::Result<u64> {
     Ok(VortexWriteOptions::default()
         .with_strategy(VortexLayoutStrategy::with_executor(Arc::new(
             Handle::current(),
@@ -22,8 +22,8 @@ pub async fn vortex_compress_write(array: &dyn Array, buf: &mut Vec<u8>) -> Vort
 }
 
 #[inline(never)]
-pub async fn vortex_decompress_read(buf: Bytes) -> VortexResult<Vec<ArrayRef>> {
-    VortexOpenOptions::in_memory()
+pub async fn vortex_decompress_read(buf: Bytes) -> anyhow::Result<Vec<ArrayRef>> {
+    Ok(VortexOpenOptions::in_memory()
         .open(buf)
         .await?
         .scan()?
@@ -33,5 +33,5 @@ pub async fn vortex_decompress_read(buf: Bytes) -> VortexResult<Vec<ArrayRef>> {
         .await?
         .into_iter()
         .map(|a| a.into_arrow_preferred())
-        .collect::<VortexResult<Vec<_>>>()
+        .collect::<VortexResult<Vec<_>>>()?)
 }

bench-vortex/src/compress/vortex.rs

@@ -8,11 +8,11 @@ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use vortex::TryIntoArray;
 use vortex::dtype::DType;
 use vortex::dtype::arrow::FromArrowType;
-use vortex::error::{VortexError, VortexResult};
+use vortex::error::VortexError;
 use vortex::iter::{ArrayIteratorAdapter, ArrayIteratorExt};
 use vortex::stream::ArrayStream;
 
-pub fn parquet_to_vortex(parquet_path: PathBuf) -> VortexResult<impl ArrayStream> {
+pub fn parquet_to_vortex(parquet_path: PathBuf) -> anyhow::Result<impl ArrayStream> {
     let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(parquet_path)?)?.build()?;
 
     let array_iter = ArrayIteratorAdapter::new(
@@ -31,7 +31,7 @@ pub async fn csv_to_parquet_file(
     options: CsvReadOptions<'_>,
     csv_path: &str,
     parquet_path: &str,
-) -> VortexResult<()> {
+) -> anyhow::Result<()> {
     let df = session.read_csv(csv_path, options).await?;
 
     df.write_parquet(parquet_path, DataFrameWriteOptions::default(), None)

bench-vortex/src/conversions.rs

@@ -13,7 +13,6 @@ use object_store::path::Path as ObjectStorePath;
 use tokio::fs::OpenOptions;
 use tracing::info;
 use url::Url;
-use vortex::error::VortexExpect;
 use vortex::file::VortexWriteOptions;
 use vortex_datafusion::persistent::VortexFormat;
 
@@ -73,10 +72,7 @@ pub async fn register_parquet_files(
             let config = if let Some(schema) = schema {
                 config.with_schema(schema.into())
             } else {
-                config
-                    .infer_schema(&session.state())
-                    .await
-                    .vortex_expect("cannot infer schema")
+                config.infer_schema(&session.state()).await?
             };
 
             let listing_table = Arc::new(ListingTable::try_new(config)?);

bench-vortex/src/datasets/file.rs

@@ -18,7 +18,6 @@ use object_store::aws::AmazonS3Builder;
 use object_store::gcp::GoogleCloudStorageBuilder;
 use object_store::local::LocalFileSystem;
 use url::Url;
-use vortex::error::VortexResult;
 use vortex_datafusion::persistent::VortexFormatFactory;
 
 pub static GIT_COMMIT_ID: LazyLock<String> = LazyLock::new(|| {
@@ -50,7 +49,7 @@ pub fn get_session_context(disable_datafusion_cache: bool) -> SessionContext {
         .build_arc()
         .expect("could not build runtime environment");
 
-    let factory = VortexFormatFactory::default_config();
+    let factory = VortexFormatFactory::default();
 
     let mut session_state_builder = SessionStateBuilder::new()
         .with_config(SessionConfig::default())
@@ -109,7 +108,7 @@ pub fn make_object_store(
 pub async fn execute_query(
     ctx: &SessionContext,
     query: &str,
-) -> VortexResult<(Vec<RecordBatch>, Arc<dyn ExecutionPlan>)> {
+) -> anyhow::Result<(Vec<RecordBatch>, Arc<dyn ExecutionPlan>)> {
     let plan = ctx.sql(query).await?;
     let (state, plan) = plan.into_parts();
     let physical_plan = state.create_physical_plan(&plan).await?;

bench-vortex/src/engines/df/mod.rs

@@ -102,9 +102,6 @@ pub enum Format {
     Arrow,
     #[clap(name = "parquet")]
     Parquet,
-    #[clap(name = "in-memory-vortex")]
-    #[serde(rename = "in-memory-vortex")]
-    InMemoryVortex,
     #[clap(name = "vortex")]
     #[serde(rename = "vortex")]
     OnDiskVortex,
@@ -125,7 +122,6 @@ impl Format {
             Format::Csv => "csv",
             Format::Arrow => "arrow",
             Format::Parquet => "parquet",
-            Format::InMemoryVortex => "vortex-in-memory",
             Format::OnDiskVortex => "vortex-file-compressed",
             Format::OnDiskDuckDB => "duckdb",
         }
@@ -136,7 +132,6 @@ impl Format {
             Format::Csv => "csv",
             Format::Arrow => "arrow",
             Format::Parquet => "parquet",
-            Format::InMemoryVortex => unreachable!("no extension"),
             Format::OnDiskVortex => "vortex",
             Format::OnDiskDuckDB => "duckdb",
         }