some work

Signed-off-by: Adam Gutglick <[email protected]>

Author: AdamGS

Cargo.lock

@@ -334,3 +334,20 @@ lto = false
 [profile.bench_assert]
 debug-assertions = true
 inherits = "bench"
+
+[patch.crates-io]
+datafusion = { version = "51", default-features = false, features = [
+    "sql",
+], git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-catalog = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-common = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-common-runtime = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-datasource = { version = "51", default-features = false, git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-execution = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-expr = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-functions = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-physical-expr = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-physical-expr-adapter = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-physical-expr-common = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-physical-plan = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }
+datafusion-pruning = { version = "51", git = "https://github.com/apache/datafusion", rev = "47df535d2cd5aac5ad5a92bdc837f38e05ea0f0f" }

Cargo.toml

@@ -10,7 +10,6 @@ use vortex::expr::stats::Precision;
 
 mod convert;
 mod persistent;
-pub mod vendor;
 
 pub use convert::exprs::ExpressionConvertor;
 pub use persistent::*;

vortex-datafusion/src/lib.rs

@@ -17,16 +17,17 @@ use datafusion_common::Result as DFResult;
 use datafusion_common::Statistics;
 use datafusion_common::config::ConfigField;
 use datafusion_common::config_namespace;
+use datafusion_common::internal_datafusion_err;
 use datafusion_common::not_impl_err;
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::stats::Precision;
 use datafusion_common_runtime::SpawnedTask;
+use datafusion_datasource::TableSchema;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_compression_type::FileCompressionType;
 use datafusion_datasource::file_format::FileFormat;
 use datafusion_datasource::file_format::FileFormatFactory;
 use datafusion_datasource::file_scan_config::FileScanConfig;
-use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
 use datafusion_datasource::file_sink_config::FileSinkConfig;
 use datafusion_datasource::sink::DataSinkExec;
 use datafusion_datasource::source::DataSourceExec;
@@ -363,6 +364,7 @@ impl FileFormat for VortexFormat {
                                 is_constant.as_exact().map(|_| Precision::Exact(1))
                             })
                             .unwrap_or(Precision::Absent),
+                        byte_size: Precision::Absent,
                     }
                 })
                 .collect::<Vec<_>>();
@@ -386,14 +388,12 @@ impl FileFormat for VortexFormat {
         _state: &dyn Session,
         file_scan_config: FileScanConfig,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
-        let source = VortexSource::new(self.session.clone(), self.file_cache.clone());
-        let source = Arc::new(source);
+        // We make sure the scan's source is the right type, but we don't have anything else to do here.
+        if !file_scan_config.file_source().as_any().is::<VortexSource>() {
+            return Err(internal_datafusion_err!("Expected VortexSource"));
+        }
 
-        Ok(DataSourceExec::from_data_source(
-            FileScanConfigBuilder::from(file_scan_config)
-                .with_source(source)
-                .build(),
-        ))
+        Ok(DataSourceExec::from_data_source(file_scan_config))
     }
 
     async fn create_writer_physical_plan(
@@ -413,8 +413,9 @@ impl FileFormat for VortexFormat {
         Ok(Arc::new(DataSinkExec::new(input, sink, order_requirements)) as _)
     }
 
-    fn file_source(&self) -> Arc<dyn FileSource> {
+    fn file_source(&self, table_schema: TableSchema) -> Arc<dyn FileSource> {
         Arc::new(VortexSource::new(
+            table_schema,
             self.session.clone(),
             self.file_cache.clone(),
         ))