feat: integrate cypher, csr and engine work for 0.3.0

crates/lance-csr-benches/Cargo.toml

@@ -10,6 +10,7 @@ lance-csr = { path = "../lance-csr", version = "0.3.0" }
 lance = "13.0.0"
 arrow-array = "58"
 arrow-schema = "58"
+datafusion = { version = "54", default-features = false }
 criterion = { version = "0.5", features = ["async", "async_tokio", "html_reports"] }
 futures = "0.3"
 tempfile = "3"

crates/lance-csr-benches/README.md

@@ -26,3 +26,41 @@ cargo bench -p lance-csr-benches --bench graph_execution
 
 This crate is not published to crates.io and is excluded from releases.
 It exists solely for performance testing during development.
+
+
+## Query template reuse
+
+Measure fresh parsing/planning/execution against reuse of an unoptimized scalar-parameter template on the
+same 10,000-row Arrow snapshot. Both paths optimize and build a physical plan for each execution. The harness
+checks identical Arrow results, row count and ID checksum before timing; catalog construction is excluded.
+
+```bash
+cargo bench -p lance-csr-benches --bench graph_execution -- query_templates --warm-up-time 1
+```
+
+For a stage breakdown on a separate fixed 1,000-row scan, run the optional core diagnostic:
+
+```bash
+cargo test --release -p lance-csr profile_query_planning_stages -- --ignored --nocapture
+```
+
+It reports elapsed means for 100 iterations of parsing, semantic analysis, graph logical planning,
+DataFusion logical planning, optimization, physical planning, execution and cached binding. Physical timing
+calls DataFusion's query planner directly so optimization is counted only once. It verifies 990 result rows
+and their ID checksum. These local measurements do not replace the two SF1 suites required for performance
+submissions; record runtime, hardware, revision and build profile with any reported numbers. The engine's
+Python cache also includes dictionary conversion and lookup, which this Rust microbenchmark does not time.
+
+
+## Comprehension counts
+
+Compare `size([x IN list WHERE predicate])` with the equivalent identity-mapped comprehension, which keeps the
+ordinary list-building path. Both queries run against 1,000 input rows with list lengths 16 and 256; results,
+row count and summed counts must match before timing. This measures planning plus execution after parsing.
+
+```bash
+cargo bench -p lance-csr-benches --bench graph_execution -- comprehension_count --warm-up-time 1
+```
+
+The count path removes ordered list aggregation. It still numbers and unnests rows, evaluates predicates and
+groups by input row, so these measurements do not establish a general bound on planning or memory use.

crates/lance-csr-benches/benches/graph_execution.rs

@@ -336,5 +336,119 @@ fn bench_cypher_execution(c: &mut Criterion) {
     group.finish();
 }
 
-criterion_group!(benches, bench_cypher_execution);
+// A fixed snapshot isolates template reuse from dataset/catalog construction.
+// The result count and checksum are checked outside the timed loops.
+fn bench_query_templates(c: &mut Criterion) {
+    use datafusion::datasource::{DefaultTableSource, MemTable};
+    use datafusion::execution::context::SessionContext;
+    use lance_csr::{GraphSourceCatalog, InMemoryCatalog};
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let context = SessionContext::new();
+    let batch = make_people_batch(10_000);
+    let provider = Arc::new(MemTable::try_new(batch.schema(), vec![vec![batch]]).unwrap());
+    let catalog: Arc<dyn GraphSourceCatalog> = Arc::new(
+        InMemoryCatalog::new()
+            .with_node_source("Person", Arc::new(DefaultTableSource::new(provider))),
+    );
+    let config = GraphConfig::builder()
+        .node_table("Person", "person_id")
+        .build()
+        .unwrap();
+    let text = "MATCH (p:Person) WHERE p.age >= $minimum RETURN p.person_id AS id ORDER BY id";
+    let fresh = || {
+        CypherQuery::new(text)
+            .unwrap()
+            .with_config(config.clone())
+            .with_parameter("minimum", 50)
+    };
+    let mut template = fresh();
+    assert!(template.prepare_cached_plan(catalog.clone()).unwrap());
+    let expected = rt
+        .block_on(fresh().execute_with_catalog_and_context(catalog.clone(), context.clone()))
+        .unwrap();
+    let cached = rt
+        .block_on(template.execute_with_catalog_and_context(catalog.clone(), context.clone()))
+        .unwrap();
+    assert_eq!(cached, expected);
+    let expected_ids: Vec<i32> = (0..10_000).filter(|id| 20 + id % 60 >= 50).collect();
+    assert_eq!(cached.num_rows(), expected_ids.len());
+    let ids = cached
+        .column(0)
+        .as_any()
+        .downcast_ref::<Int32Array>()
+        .unwrap();
+    assert_eq!(
+        ids.values().iter().map(|id| i64::from(*id)).sum::<i64>(),
+        expected_ids.iter().map(|id| i64::from(*id)).sum::<i64>()
+    );
+    let mut group = c.benchmark_group("query_templates");
+    group.bench_function("cold_parse_plan_execute", |b| {
+        b.iter(|| {
+            black_box(
+                rt.block_on(
+                    fresh().execute_with_catalog_and_context(catalog.clone(), context.clone()),
+                )
+                .unwrap(),
+            );
+        })
+    });
+    group.bench_function("cached_bind_plan_execute", |b| {
+        b.iter(|| {
+            let query = template.clone().with_parameter("minimum", black_box(50));
+            black_box(
+                rt.block_on(
+                    query.execute_with_catalog_and_context(catalog.clone(), context.clone()),
+                )
+                .unwrap(),
+            );
+        })
+    });
+    group.finish();
+}
+
+fn bench_comprehension_counts(c: &mut Criterion) {
+    use arrow_array::Int64Array;
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let config = GraphConfig::builder()
+        .node_table("Person", "person_id")
+        .build()
+        .unwrap();
+    let data = HashMap::from([("Person".to_string(), make_people_batch(1000))]);
+    let mut group = c.benchmark_group("comprehension_count");
+    for length in [16, 256] {
+        let query = |map| {
+            CypherQuery::new(&format!(
+            "MATCH (p:Person) RETURN p.person_id AS id,              size([x IN range(0, {}) WHERE x % 2 = 0 {map}]) AS n ORDER BY id", length - 1
+        )).unwrap().with_config(config.clone())
+        };
+        let count = query("");
+        let reference = query("| x");
+        let actual = execute_cypher_query(&rt, &count, data.clone());
+        let expected = execute_cypher_query(&rt, &reference, data.clone());
+        assert_eq!(actual.columns(), expected.columns());
+        assert_eq!(actual.num_rows(), 1000);
+        let counts = actual
+            .column(1)
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .unwrap();
+        assert_eq!(counts.values().iter().sum::<i64>(), 1000 * length / 2);
+        for (name, query) in [
+            ("filtered_count", &count),
+            ("materialized_reference", &reference),
+        ] {
+            group.bench_with_input(BenchmarkId::new(name, length), &length, |b, _| {
+                b.iter(|| black_box(execute_cypher_query(&rt, query, data.clone())));
+            });
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(
+    benches,
+    bench_cypher_execution,
+    bench_query_templates,
+    bench_comprehension_counts
+);
 criterion_main!(benches);

crates/lance-csr-python/Cargo.toml

@@ -21,4 +21,4 @@ lance-csr = { path = "../lance-csr" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 pyo3 = { version = "0.28", features = ["extension-module", "abi3-py312", "py-clone"] }
-tokio = { version = "1.37", features = ["rt-multi-thread", "macros"] }
+tokio = { version = "1.37", features = ["rt-multi-thread", "macros", "sync", "time"] }

crates/lance-csr-python/src/executor.rs

@@ -13,7 +13,54 @@
 // limitations under the License.
 
 use futures::Future;
-use pyo3::{PyResult, Python};
+use pyo3::{
+    exceptions::{PyTimeoutError, PyValueError},
+    PyResult, Python,
+};
+use std::time::Duration;
+
+pub(crate) const QUERY_TIMEOUT_MESSAGE: &str = "Query timeout exceeded";
+
+/// One absolute deadline shared by planning, execution and all streamed reads.
+#[derive(Clone, Copy)]
+pub(crate) struct QueryDeadline(Option<tokio::time::Instant>);
+
+impl QueryDeadline {
+    pub(crate) fn new(timeout: Option<f64>) -> PyResult<Self> {
+        let Some(seconds) = timeout else {
+            return Ok(Self(None));
+        };
+        let duration = Duration::try_from_secs_f64(seconds).map_err(|_| {
+            PyValueError::new_err("timeout must be a finite nonnegative number of seconds")
+        })?;
+        let instant = tokio::time::Instant::now()
+            .checked_add(duration)
+            .ok_or_else(|| PyValueError::new_err("timeout exceeds the supported deadline range"))?;
+        Ok(Self(Some(instant)))
+    }
+
+    pub(crate) fn instant(self) -> Option<tokio::time::Instant> {
+        self.0
+    }
+
+    pub(crate) fn check(self) -> PyResult<()> {
+        if self
+            .0
+            .is_some_and(|deadline| tokio::time::Instant::now() >= deadline)
+        {
+            Err(PyTimeoutError::new_err(QUERY_TIMEOUT_MESSAGE))
+        } else {
+            Ok(())
+        }
+    }
+
+    async fn expired(self) {
+        match self.0 {
+            Some(deadline) => tokio::time::sleep_until(deadline).await,
+            None => std::future::pending::<()>().await,
+        }
+    }
+}
 
 pub const SIGNAL_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);
 
@@ -56,7 +103,22 @@ impl BackgroundExecutor {
     where
         F::Output: Send,
     {
-        let future = Self::result_or_interrupt(future);
+        self.block_on_until(py, QueryDeadline(None), future)
+    }
+
+    /// Apply a query deadline without changing unrelated executor callers.
+    pub(crate) fn block_on_until<F: Future + Send>(
+        &self,
+        py: Option<Python<'_>>,
+        deadline: QueryDeadline,
+        future: F,
+    ) -> PyResult<F::Output>
+    where
+        F::Output: Send,
+    {
+        deadline.check()?;
+        let future = lance_csr::query::with_execution_deadline(deadline.instant(), future);
+        let future = Self::result_or_interrupt(future, deadline);
         if let Some(py) = py {
             py.detach(move || self.runtime.block_on(future))
         } else {
@@ -65,7 +127,7 @@ impl BackgroundExecutor {
         }
     }
 
-    async fn result_or_interrupt<F>(future: F) -> PyResult<F::Output>
+    async fn result_or_interrupt<F>(future: F, deadline: QueryDeadline) -> PyResult<F::Output>
     where
         F: Future + Send,
         F::Output: Send,
@@ -84,7 +146,9 @@ impl BackgroundExecutor {
         };
 
         tokio::select! {
-            result = future => Ok(result),
+            biased;
+            _ = deadline.expired() => Err(PyTimeoutError::new_err(QUERY_TIMEOUT_MESSAGE)),
+            result = future => { deadline.check()?; Ok(result) },
             err = interrupt_future => err,
         }
     }