fix: Expression handling improvements for DataFusion integration

Ported from spiceai-51 patches:
- Skip unsupported expressions in make_vortex_predicate instead of erroring (#8)
- Handle empty IN lists: x IN () returns false, x NOT IN () returns true (#8)
- Allow DynamicFilterPhysicalExpr pushdown (#8)
- Remove is_dynamic_physical_expr check that blocked all dynamic filters
- Add Debug impl for BtrBlocksCompressor to satisfy CompressorPlugin bounds

Author: lukekim

vortex-btrblocks/src/canonical_compressor.rs

@@ -101,6 +101,16 @@ pub struct BtrBlocksCompressor {
     pub string_schemes: Vec<&'static dyn StringScheme>,
 }
 
+impl std::fmt::Debug for BtrBlocksCompressor {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BtrBlocksCompressor")
+            .field("int_schemes", &self.int_schemes.len())
+            .field("float_schemes", &self.float_schemes.len())
+            .field("string_schemes", &self.string_schemes.len())
+            .finish()
+    }
+}
+
 impl Default for BtrBlocksCompressor {
     fn default() -> Self {
         BtrBlocksCompressorBuilder::default().build()

vortex-datafusion/src/convert/exprs.rs

@@ -16,7 +16,6 @@ use datafusion_physical_expr::ScalarFunctionExpr;
 use datafusion_physical_expr::projection::ProjectionExpr;
 use datafusion_physical_expr::projection::ProjectionExprs;
 use datafusion_physical_expr::utils::collect_columns;
-use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_plan::expressions as df_expr;
 use itertools::Itertools;
 use vortex::dtype::DType;
@@ -52,10 +51,13 @@ pub(crate) fn make_vortex_predicate(
     expr_convertor: &dyn ExpressionConvertor,
     predicate: &[Arc<dyn PhysicalExpr>],
 ) -> DFResult<Option<Expression>> {
-    let exprs = predicate
+    let exprs: Vec<_> = predicate
         .iter()
-        .map(|e| expr_convertor.convert(e.as_ref()))
-        .collect::<DFResult<Vec<_>>>()?;
+        .filter_map(|e| {
+            // If conversion fails, skip this expression (equivalent to lit(true) in AND conjunction)
+            expr_convertor.convert(e.as_ref()).ok()
+        })
+        .collect();
 
     Ok(and_collect(exprs))
 }
@@ -221,8 +223,13 @@ impl ExpressionConvertor for DefaultExpressionConvertor {
                 })
                 .try_collect()?;
 
+            // Handle empty IN list: `x IN ()` is always false, `x NOT IN ()` is always true
+            let Some(first_element) = list_elements.first() else {
+                return Ok(lit(in_list.negated()));
+            };
+
             let list = Scalar::list(
-                list_elements[0].dtype().clone(),
+                first_element.dtype().clone(),
                 list_elements,
                 Nullability::Nullable,
             );
@@ -361,12 +368,6 @@ fn try_operator_from_df(value: &DFOperator) -> DFResult<Operator> {
 }
 
 fn can_be_pushed_down_impl(df_expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
-    // We currently do not support pushdown of dynamic expressions in DF.
-    // See issue: https://github.com/vortex-data/vortex/issues/4034
-    if is_dynamic_physical_expr(df_expr) {
-        return false;
-    }
-
     let expr = df_expr.as_any();
     if let Some(binary) = expr.downcast_ref::<df_expr::BinaryExpr>() {
         can_binary_be_pushed_down(binary, schema)
@@ -396,6 +397,12 @@ fn can_be_pushed_down_impl(df_expr: &Arc<dyn PhysicalExpr>, schema: &Schema) ->
                 .all(|e| can_be_pushed_down_impl(e, schema))
     } else if let Some(scalar_fn) = expr.downcast_ref::<ScalarFunctionExpr>() {
         can_scalar_fn_be_pushed_down(scalar_fn)
+    } else if expr
+        .downcast_ref::<df_expr::DynamicFilterPhysicalExpr>()
+        .is_some()
+    {
+        // Assume dynamic filters can be pushed down - the child won't be specified until execution time
+        true
     } else {
         tracing::debug!(%df_expr, "DataFusion expression can't be pushed down");
         false