refactor: build known scan expressions as bound expressions directly

Scan callers that know their projection or filter statically were writing
`expr.optimize_recursive(dtype)?.bind(dtype)?`. For these shapes the optimizer
pass is a no-op, so the round trip only costs a tree walk and produces a fresh
tree identity that defeats the identity-keyed caches the parent PR introduces.

Build them with the `bound::*` constructors instead, at the crate doc example,
the `vortex` and `vortex-file` tests, and the compress/TPC-H benchmarks. The
one exception is `and(gt, lt_eq)` over a single column, which the optimizer
folds into a `between`; that site now constructs the `between` directly, which
is the form the scan was already receiving.

Callers whose expression arrives from outside — Python, DataFusion, scan
requests and the fuzz target — still optimize and bind, as do the `vortex-file`
tests that deliberately exercise expressions which bind but fail during
execution.

Add `bound_constructors_match_optimize_then_bind`, which asserts each converted
shape equals its `optimize_recursive(..).bind(..)` result so the two cannot
drift apart.

Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

benchmarks/compress-bench/src/vortex.rs

@@ -14,8 +14,7 @@ use futures::StreamExt;
 use futures::pin_mut;
 use vortex::array::IntoArray;
 use vortex::dtype::FieldNames;
-use vortex::expr::root;
-use vortex::expr::select;
+use vortex::expr::bound;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::WriteOptionsSessionExt;
 use vortex_arrow::ToArrowType;
@@ -71,9 +70,7 @@ impl Compressor for VortexCompressor {
         if let Some(cols) = read_projection(root_columns) {
             // Columns are named "0".."num_columns-1"; project the given subset.
             let names: FieldNames = cols.iter().map(|i| i.to_string()).collect();
-            let projection = select(names, root())
-                .optimize_recursive(&source_dtype)?
-                .bind(&source_dtype)?;
+            let projection = bound::select(names, bound::root(source_dtype.clone()));
             scan = scan.with_projection(projection);
         }
         let schema = Arc::new(scan.dtype()?.to_arrow_schema()?);

vortex-array/src/expr/mod.rs

@@ -225,9 +225,12 @@ mod tests {
     use crate::expr::not;
     use crate::expr::not_eq;
     use crate::expr::or;
+    use crate::expr::pack;
     use crate::expr::select;
     use crate::expr::select_exclude;
     use crate::scalar::Scalar;
+    use crate::scalar_fn::fns::between::BetweenOptions;
+    use crate::scalar_fn::fns::between::StrictComparison;
     use crate::scalar_fn::fns::literal::Literal;
 
     #[test]
@@ -410,4 +413,93 @@ mod tests {
         let expression = root();
         assert!(!expression.contains::<Literal>().unwrap());
     }
+
+    /// Scan callers that know their expression statically build it with the `bound::*`
+    /// constructors instead of `optimize_recursive(..).bind(..)`. Both must agree, otherwise
+    /// those call sites would silently ship a different expression to the scan.
+    #[test]
+    fn bound_constructors_match_optimize_then_bind() -> vortex_error::VortexResult<()> {
+        let u64_scope = DType::Primitive(PType::U64, Nullability::NonNullable);
+        let struct_scope = DType::Struct(
+            StructFields::from_iter([
+                ("name", DType::Utf8(Nullability::Nullable)),
+                ("age", DType::Primitive(PType::I32, Nullability::Nullable)),
+            ]),
+            Nullability::NonNullable,
+        );
+
+        let cases: Vec<(DType, Expression, BoundExpression)> = vec![
+            (
+                u64_scope.clone(),
+                gt(root(), lit(2u64)),
+                bound::gt(bound::root(u64_scope), bound::lit(2u64)),
+            ),
+            (
+                struct_scope.clone(),
+                select(["name"], root()),
+                bound::select(["name"], bound::root(struct_scope.clone())),
+            ),
+            (
+                struct_scope.clone(),
+                pack([("name", col("name"))], Nullability::NonNullable),
+                bound::pack(
+                    [("name", bound::col("name", struct_scope.clone()))],
+                    Nullability::NonNullable,
+                ),
+            ),
+            (
+                struct_scope.clone(),
+                eq(get_item("name", root()), lit("Joseph")),
+                bound::eq(
+                    bound::col("name", struct_scope.clone()),
+                    bound::lit("Joseph"),
+                ),
+            ),
+            (
+                struct_scope.clone(),
+                or(
+                    eq(get_item("name", root()), lit("Angela")),
+                    and(
+                        gt_eq(get_item("age", root()), lit(20)),
+                        lt_eq(get_item("age", root()), lit(30)),
+                    ),
+                ),
+                bound::or(
+                    bound::eq(
+                        bound::col("name", struct_scope.clone()),
+                        bound::lit("Angela"),
+                    ),
+                    bound::and(
+                        bound::gt_eq(bound::col("age", struct_scope.clone()), bound::lit(20)),
+                        bound::lt_eq(bound::col("age", struct_scope.clone()), bound::lit(30)),
+                    ),
+                ),
+            ),
+            // `and(gt, lt_eq)` over a single column is folded into a `between`, so the direct
+            // form is the `between` rather than the conjunction it was written as.
+            (
+                struct_scope.clone(),
+                and(
+                    gt(get_item("age", root()), lit(21)),
+                    lt_eq(get_item("age", root()), lit(33)),
+                ),
+                bound::between(
+                    bound::col("age", struct_scope),
+                    bound::lit(21),
+                    bound::lit(33),
+                    BetweenOptions {
+                        lower_strict: StrictComparison::Strict,
+                        upper_strict: StrictComparison::NonStrict,
+                    },
+                ),
+            ),
+        ];
+
+        for (scope, unbound, direct) in cases {
+            let via_optimizer = unbound.optimize_recursive(&scope)?.bind(&scope)?;
+            assert_eq!(via_optimizer, direct, "mismatch for {unbound}");
+        }
+
+        Ok(())
+    }
 }

vortex-bench/src/datasets/tpch_l_comment.rs

@@ -13,8 +13,7 @@ use vortex::array::IntoArray;
 use vortex::array::arrays::ChunkedArray;
 use vortex::array::arrays::StructArray;
 use vortex::dtype::Nullability::NonNullable;
-use vortex::expr::col;
-use vortex::expr::pack;
+use vortex::expr::bound;
 use vortex::file::OpenOptionsSessionExt;
 
 use crate::Format;
@@ -66,9 +65,10 @@ impl Dataset for TPCHLCommentChunked {
 
         let path = data_dir.join("lineitem.vortex");
         let file = SESSION.open_options().open_path(path).await?;
-        let projection = pack(vec![("l_comment", col("l_comment"))], NonNullable)
-            .optimize_recursive(file.dtype())?
-            .bind(file.dtype())?;
+        let projection = bound::pack(
+            vec![("l_comment", bound::col("l_comment", file.dtype().clone()))],
+            NonNullable,
+        );
         let chunks: Vec<_> = file
             .scan()?
             .with_projection(projection)