perf(datafusion): push down partition-grouped COUNT(*) to manifests

crates/integrations/datafusion/src/lateral_vector_search.rs

@@ -107,6 +107,10 @@ pub(crate) fn optimizer_rules() -> Vec<Arc<dyn OptimizerRule + Send + Sync>> {
         Arc::new(RewriteLateralVectorSearch::new()),
     ];
     rules.extend(Optimizer::default().rules);
+    // After the default rules, so filters have already been pushed into the scan.
+    rules.push(Arc::new(
+        crate::partition_count_pushdown::PushDownPartitionCount,
+    ));
     rules
 }
 

crates/integrations/datafusion/src/lib.rs

@@ -52,6 +52,7 @@ mod full_text_search;
 mod hybrid_search;
 mod lateral_vector_search;
 mod merge_into;
+mod partition_count_pushdown;
 mod physical_plan;
 mod procedures;
 mod relation_planner;