Fix compilation errors in vortex-datafusion for DF52

- Add missing imports: VortexResult, case_when, case_when_no_else, vortex_bail
- Fix can_be_pushed_down -> can_be_pushed_down_impl in non-trait contexts
- Remove duplicate can_case_be_pushed_down function (keep searched CASE version)
- Remove duplicate projection/try_pushdown_projection methods (keep non-Option version)
- Re-add FileScanConfigBuilder import removed in previous commit

Author: lukekim

vortex-array/src/scalar_fn/fns/mod.rs

@@ -3,6 +3,7 @@
 
 pub mod between;
 pub mod binary;
+pub mod case_when;
 pub mod cast;
 pub mod dynamic;
 pub mod fill_null;

vortex-datafusion/src/convert/exprs.rs

@@ -37,7 +37,11 @@ use vortex::scalar_fn::ScalarFnVTableExt;
 use vortex::scalar_fn::fns::binary::Binary;
 use vortex::scalar_fn::fns::like::Like;
 use vortex::scalar_fn::fns::like::LikeOptions;
+use vortex::scalar_fn::fns::case_when::case_when;
+use vortex::scalar_fn::fns::case_when::case_when_no_else;
 use vortex::scalar_fn::fns::operators::Operator;
+use vortex::error::VortexResult;
+use vortex::error::vortex_bail;
 
 use crate::convert::FromDataFusion;
 
@@ -511,22 +515,6 @@ fn can_binary_be_pushed_down(binary: &df_expr::BinaryExpr, schema: &Schema) -> b
         && can_be_pushed_down_impl(binary.right(), schema)
 }
 
-fn can_case_be_pushed_down(case_expr: &df_expr::CaseExpr, schema: &Schema) -> bool {
-    case_expr
-        .expr()
-        .is_none_or(|base_expr| can_be_pushed_down_impl(base_expr, schema))
-        && case_expr
-            .when_then_expr()
-            .iter()
-            .all(|(when_expr, then_expr)| {
-                can_be_pushed_down_impl(when_expr, schema)
-                    && can_be_pushed_down_impl(then_expr, schema)
-            })
-        && case_expr
-            .else_expr()
-            .is_some_and(|else_expr| can_be_pushed_down_impl(else_expr, schema))
-}
-
 fn can_case_be_pushed_down(case_expr: &df_expr::CaseExpr, schema: &Schema) -> bool {
     // We only support the "searched CASE" form (CASE WHEN cond THEN result ...)
     // not the "simple CASE" form (CASE expr WHEN value THEN result ...)
@@ -536,14 +524,14 @@ fn can_case_be_pushed_down(case_expr: &df_expr::CaseExpr, schema: &Schema) -> bo
 
     // Check all when/then pairs
     for (when_expr, then_expr) in case_expr.when_then_expr() {
-        if !can_be_pushed_down(when_expr, schema) || !can_be_pushed_down(then_expr, schema) {
+        if !can_be_pushed_down_impl(when_expr, schema) || !can_be_pushed_down_impl(then_expr, schema) {
             return false;
         }
     }
 
     // Check the optional else clause
     if let Some(else_expr) = case_expr.else_expr()
-        && !can_be_pushed_down(else_expr, schema)
+        && !can_be_pushed_down_impl(else_expr, schema)
     {
         return false;
     }
@@ -591,7 +579,7 @@ fn can_scalar_fn_be_pushed_down(scalar_fn: &ScalarFunctionExpr, schema: &Schema)
         && scalar_fn
             .args()
             .iter()
-            .all(|arg| can_be_pushed_down(arg, schema))
+            .all(|arg| can_be_pushed_down_impl(arg, schema))
 }
 
 // TODO(adam): Replace with `DataType::is_decimal` once its released.

vortex-datafusion/src/persistent/format.rs

@@ -28,6 +28,7 @@ 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;

vortex-datafusion/src/persistent/source.rs

@@ -222,22 +222,6 @@ impl FileSource for VortexSource {
         VORTEX_FILE_EXTENSION
     }
 
-    fn projection(&self) -> Option<&ProjectionExprs> {
-        self.projection.as_ref()
-    }
-
-    fn try_pushdown_projection(
-        &self,
-        projection: &ProjectionExprs,
-    ) -> DFResult<Option<Arc<dyn FileSource>>> {
-        let mut source = self.clone();
-        source.projection = match &self.projection {
-            Some(existing) => Some(existing.try_merge(projection)?),
-            None => Some(projection.clone()),
-        };
-        Ok(Some(Arc::new(source)))
-    }
-
     fn fmt_extra(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
         match t {
             DisplayFormatType::Default | DisplayFormatType::Verbose => {