feat: exec pushdown with global aggregates (#124)

* Exec pushdown: internal __lance_exec for global aggregates

Add a small ExecIR container (reusing existing FilterIR bytes) and an optimizer rewrite that replaces ungrouped aggregates over Lance scans with the internal __lance_exec table function.

The Rust side validates the IR and executes it via DataFusion on top of the Lance scanner, returning a (typically tiny) Arrow stream back to DuckDB.

The rewrite is always enabled but remains conservative: on any encode/validate/type mismatch it falls back to the original plan.

Tests: GEN=ninja make test_debug

* Fix ci

* Fix exec pushdown on table_filters-keyed scans

LogicalGet.table_filters uses column IDs as map keys, not scan column positions. Exec pushdown previously fed these filters into BuildLanceTableFilterIRParts with a column_ids vector sized to the projected columns, causing all_filters_pushed=false and a silent fallback for TPC-H Q6.

Fix by using an identity column_ids mapping sized by the max filter column id and collecting extra scan columns by column id.

Add a sqllogictest that asserts Q6 rewrites to __lance_exec.

Tests: GEN=ninja make test_debug

* Enable always-on exec pushdown for group/order aggregates

* Fix sqllogictest expectations under always-on exec pushdown

* ExecIR: encode CAST to preserve GROUP BY semantics

* Add formatting requirement to AGENTS

Author: Xuanwo

AGENTS.md

@@ -60,6 +60,8 @@ This repository is configured with `uv`, so you can run formatting via:
 uv run make format
 ```
 
+PR requirement: Before submitting a PR, run `uv run make format` and commit any resulting formatting changes (as a separate commit if it helps review).
+
 ### Testing
 
 The `release` build can be slow. For fast iteration, prefer `test_debug` when available.

CMakeLists.txt

@@ -13,6 +13,8 @@ set(EXTENSION_SOURCES src/lance_extension.cpp src/lance_scan.cpp
                       src/lance_search.cpp src/lance_common.cpp
                       src/lance_secrets.cpp
                       src/lance_filter_ir.cpp
+                      src/lance_exec_ir.cpp
+                      src/lance_logical_exec.cpp
                       src/lance_storage.cpp
                       src/lance_metadata.cpp
                       src/lance_replacement.cpp
@@ -117,10 +119,13 @@ set(RUST_FFI_DEPENDS
     ${CMAKE_CURRENT_LIST_DIR}/Cargo.toml
     ${CMAKE_CURRENT_LIST_DIR}/Cargo.lock
     ${CMAKE_CURRENT_LIST_DIR}/rust/constants.rs
+    ${CMAKE_CURRENT_LIST_DIR}/rust/datafusion_stream.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/error.rs
+    ${CMAKE_CURRENT_LIST_DIR}/rust/exec_ir.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/filter_ir.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/arrow_export.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/dataset.rs
+    ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/exec.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/index.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/knn.rs
     ${CMAKE_CURRENT_LIST_DIR}/rust/ffi/mod.rs

rust/datafusion_stream.rs

@@ -0,0 +1,32 @@
+use arrow::array::RecordBatch;
+use arrow::datatypes::SchemaRef;
+use datafusion::physical_plan::SendableRecordBatchStream;
+use futures::StreamExt;
+use tokio::runtime::Handle;
+
+pub struct DataFusionStream {
+    handle: Handle,
+    stream: SendableRecordBatchStream,
+}
+
+impl DataFusionStream {
+    pub fn try_new(stream: SendableRecordBatchStream) -> Result<Self, Box<dyn std::error::Error>> {
+        let handle = crate::runtime::handle()?;
+        Ok(Self { handle, stream })
+    }
+
+    pub fn schema(&self) -> SchemaRef {
+        self.stream.schema()
+    }
+
+    pub fn next(&mut self) -> Result<Option<RecordBatch>, datafusion_common::DataFusionError> {
+        self.handle.block_on(async {
+            let next = self.stream.next().await;
+            match next {
+                Some(Ok(batch)) => Ok(Some(batch)),
+                Some(Err(err)) => Err(err),
+                None => Ok(None),
+            }
+        })
+    }
+}

rust/error.rs

@@ -65,6 +65,7 @@ pub enum ErrorCode {
     NamespaceDescribeTableInfo = 49,
     NamespaceCreateEmptyTable = 50,
     NamespaceDropTable = 51,
+    Exec = 52,
 }
 
 struct LastError {