Fix some issue

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

Author: AdamGS

vortex-datafusion/src/persistent/mod.rs

@@ -32,6 +32,7 @@ fn register_vortex_format_factory(
 
 #[cfg(test)]
 mod tests {
+
     use std::sync::Arc;
 
     use arrow_schema::{DataType, Field, Schema};
@@ -202,7 +203,6 @@ mod tests {
         register_vortex_format_factory(factory, &mut session_state_builder);
         let session = SessionContext::new_with_state(session_state_builder.build());
 
-        // Vortex
         session
             .sql(&format!(
                 "CREATE EXTERNAL TABLE my_tbl_vx \
@@ -238,8 +238,6 @@ mod tests {
         let (state, plan) = df.clone().into_parts();
         let physical_plan = state.create_physical_plan(&plan).await?;
 
-        dbg!(physical_plan.as_ref());
-
         insta::assert_snapshot!(DisplayableExecutionPlan::new(physical_plan.as_ref())
                 .tree_render().to_string(), @r"
         ┌───────────────────────────┐
@@ -251,9 +249,8 @@ mod tests {
         ┌─────────────┴─────────────┐
         │       DataSourceExec      │
         │    --------------------   │
-        │         files: 14         │
+        │          files: 3         │
         │       format: vortex      │
-        │      predicate: true      │
         └───────────────────────────┘
         ");
 

vortex-datafusion/src/persistent/source.rs

@@ -224,6 +224,14 @@ impl FileSource for VortexSource {
         filters: Vec<Arc<dyn PhysicalExpr>>,
         _config: &ConfigOptions,
     ) -> DFResult<FilterPushdownPropagation<Arc<dyn FileSource>>> {
+        // I'm not sure why, but this is important as it doesn't return the node to the parent
+        // which triggers a very wasteful repartition for some reason.
+        if filters.is_empty() {
+            return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+                vec![],
+            ));
+        }
+
         let Some(schema) = self.arrow_file_schema.as_ref() else {
             return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
                 vec![PushedDown::No; filters.len()],