Evaluate LIKE over filtered inputs filter-first

vortex-array/src/arrays/filter/kernel.rs

@@ -26,11 +26,14 @@ use crate::kernel::ExecuteParentKernel;
 use crate::matcher::Matcher;
 use crate::optimizer::kernels::ArrayKernelsExt;
 use crate::optimizer::rules::ArrayParentReduceRule;
+use crate::scalar_fn::ScalarFnVTable;
+use crate::scalar_fn::fns::like::Like;
+use crate::scalar_fn::fns::like::LikeFilterExecuteAdaptor;
 
 pub(crate) fn initialize(session: &VortexSession) {
-    session
-        .kernels()
-        .register_execute_parent_kernel(Dict.id(), Filter, TakeExecuteAdaptor(Filter));
+    let kernels = session.kernels();
+    kernels.register_execute_parent_kernel(Dict.id(), Filter, TakeExecuteAdaptor(Filter));
+    kernels.register_execute_parent_kernel(Like.id(), Filter, LikeFilterExecuteAdaptor);
 }
 
 pub trait FilterReduce: VTable {

vortex-array/src/scalar_fn/fns/like/kernel.rs

@@ -6,9 +6,12 @@ use vortex_error::VortexResult;
 
 use crate::ArrayRef;
 use crate::ExecutionCtx;
+use crate::IntoArray;
 use crate::array::ArrayView;
 use crate::array::VTable;
+use crate::arrays::Filter;
 use crate::arrays::ScalarFn;
+use crate::arrays::ScalarFnArray;
 use crate::arrays::scalar_fn::ExactScalarFn;
 use crate::arrays::scalar_fn::ScalarFnArrayExt;
 use crate::arrays::scalar_fn::ScalarFnArrayView;
@@ -105,3 +108,45 @@ where
         <V as LikeKernel>::like(array, pattern, options, ctx)
     }
 }
+
+/// Adaptor that executes a filtered input before evaluating LIKE.
+///
+/// This preserves sparse row demand for `LIKE(Filter(child), constant)`: the filter executes into a
+/// filtered child first, then the regular child-specific LIKE execute-parent kernel can run over
+/// only the selected rows.
+#[derive(Default, Debug)]
+pub struct LikeFilterExecuteAdaptor;
+
+impl ExecuteParentKernel<Filter> for LikeFilterExecuteAdaptor {
+    type Parent = ExactScalarFn<LikeExpr>;
+
+    fn execute_parent(
+        &self,
+        array: ArrayView<'_, Filter>,
+        parent: ScalarFnArrayView<'_, LikeExpr>,
+        child_idx: usize,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<ArrayRef>> {
+        if child_idx != 0 {
+            return Ok(None);
+        }
+        let scalar_fn_array = parent
+            .as_opt::<ScalarFn>()
+            .vortex_expect("ExactScalarFn matcher confirmed ScalarFnArray");
+        let filtered = array.array().clone().execute::<ArrayRef>(ctx)?;
+        let children = scalar_fn_array
+            .iter_children()
+            .enumerate()
+            .map(|(idx, child)| {
+                if idx == child_idx {
+                    filtered.clone()
+                } else {
+                    child.clone()
+                }
+            })
+            .collect();
+        Ok(Some(
+            ScalarFnArray::try_new(scalar_fn_array.scalar_fn().clone(), children)?.into_array(),
+        ))
+    }
+}