fix: plan prompt_jev in sort, window, group by, and joins; keep filters first

README.md

@@ -145,6 +145,11 @@ each row must be judged in complete isolation.
   `prompt_jev('hello', ...)` costs about one request per 256 rows scanned.
 - **Repeated calls** with identical arguments in one query are evaluated once.
   Reading several fields from one result does not repeat the request.
+- **Anywhere in a query.** `prompt_jev` works in `SELECT`, `WHERE`, `ORDER BY`,
+  `GROUP BY`, `HAVING`, window `OVER (...)` clauses, and join conditions. A call
+  in a join condition must use columns from one side of the join only.
+- **Filters run first.** Rows removed by `WHERE` are never sent to the service,
+  including when the call sits inside a CTE or subquery.
 - **Limits.** Each row's text may be up to 64 KiB after JSON encoding. Longer text
   fails the query. Requests are capped at 256 KiB.
 - **Outages.** Each request is tried three times with a 30-second timeout. If all

src/lib.rs

@@ -7,6 +7,7 @@
 //!
 //! See the README for the SQL syntax and result types.
 mod optimizer;
+mod planner;
 pub mod sql;
 mod udf;
 use async_trait::async_trait;
@@ -43,7 +44,23 @@ pub fn register(ctx: &SessionContext, client: Arc<dyn JevClient>) {
     ctx.register_udf(udf::function(client));
     let state_ref = ctx.state_ref();
     let mut state = state_ref.write();
+    // Keep DataFusion's optimizer list as is, except that the two leaf-pushdown
+    // rules are skipped for plans calling prompt_jev (see `planner`).
+    let optimizer_rules = state
+        .optimizer()
+        .rules
+        .iter()
+        .map(|rule| {
+            if planner::LeafPushdownGuard::GUARDED.contains(&rule.name()) {
+                planner::LeafPushdownGuard::wrap(Arc::clone(rule))
+            } else {
+                Arc::clone(rule)
+            }
+        })
+        .collect();
     *state = datafusion::execution::SessionStateBuilder::new_from_existing(state.clone())
+        .with_analyzer_rule(Arc::new(planner::HoistJev))
+        .with_optimizer_rules(optimizer_rules)
         .with_physical_optimizer_rule(Arc::new(optimizer::DeduplicateJev))
         .build();
 }