Decision models in the query loop: engine fix, demo runner and website

.github/workflows/ci.yml

@@ -43,7 +43,7 @@ jobs:
             | jq -r '[.packages[].rust_version | select(. != null)] | unique | .[0]')" --profile minimal
           rustup default stable
 
-      - uses: taiki-e/install-action@ab6895351a88b908e96a4697860d4ffea5e57dbe # v2.87.27
+      - uses: taiki-e/install-action@8cca0918f8471c33515202b73f56ab27f2bed283 # v2.87.29
         with:
           tool: cargo-nextest,cargo-hack@0.6.45
 

crates/algeon-datafusion/src/native/expression_contract/mod.rs

@@ -14,9 +14,7 @@ pub(in crate::native) use capability::{
     is_decimal128_to_float32_cast_via_float64, is_list_carry_type, is_narrow_decimal_carry, is_numeric_or_decimal_type,
     is_supported_to_timestamp_seconds_cast, is_supported_try_cast_pair, native_aggregate_function_kind,
 };
-pub(in crate::native) use join_residual::{
-    JoinResidualCastMode, ValidatedJoinResidual, join_residual_cast_mode, validate_join_residual,
-};
+pub(in crate::native) use join_residual::{ValidatedJoinResidual, validate_join_residual};
 pub(crate) use rejection::{
     ExpressionCapability, ExpressionRejection, ExpressionRejectionKind, ExpressionRejectionSubject,
 };

crates/algeon-datafusion/src/native/expression_contract/tests.rs

@@ -26,8 +26,7 @@ use query_planning::expr::{
 
 use super::{
     ExpressionCapability, ExpressionRejectionSubject, ScalarFunctionKind, SupportedBinaryOperator, fallible_operand,
-    join_residual::validate_join_residual_scalar_function, join_residual_cast_mode, native_aggregate_function_kind,
-    validate_join_residual, validate_physical_expr_contract,
+    native_aggregate_function_kind, validate_join_residual, validate_physical_expr_contract,
 };
 use crate::{capability::CapabilityReason, native::lowering::lower_physical_expr_for_test};
 
@@ -59,22 +58,227 @@ fn binary_capability_rejection_is_structured_and_preserves_public_reason() {
     assert_eq!(rejection.capability_reason(), CapabilityReason::UnsupportedExpression);
 }
 
+/// The left input reads `name`, `n` and `t`, the right input `r` and `label`;
+/// the filter schema lists them in that order.
+fn join_residual_schemas() -> (Arc<Schema>, Arc<Schema>, Arc<Schema>) {
+    let timestamp = DataType::Timestamp(TimeUnit::Millisecond, None);
+    let left = Arc::new(Schema::new(vec![
+        Field::new("name", DataType::Utf8, true),
+        Field::new("n", DataType::Int32, true),
+        Field::new("t", timestamp.clone(), true),
+    ]));
+    let right =
+        Arc::new(Schema::new(vec![Field::new("r", DataType::Int32, true), Field::new("label", DataType::Utf8, true)]));
+    let filter = Arc::new(Schema::new(vec![
+        Field::new("name", DataType::Utf8, true),
+        Field::new("n", DataType::Int32, true),
+        Field::new("t", timestamp, true),
+        Field::new("r", DataType::Int32, true),
+        Field::new("label", DataType::Utf8, true),
+    ]));
+    (left, right, filter)
+}
+
+fn join_residual_filter(expr: Arc<dyn PhysicalExpr>, filter_schema: &Arc<Schema>) -> JoinFilter {
+    JoinFilter::new(
+        expr,
+        vec![
+            ColumnIndex { index: 0, side: JoinSide::Left },
+            ColumnIndex { index: 1, side: JoinSide::Left },
+            ColumnIndex { index: 2, side: JoinSide::Left },
+            ColumnIndex { index: 0, side: JoinSide::Right },
+            ColumnIndex { index: 1, side: JoinSide::Right },
+        ],
+        Arc::clone(filter_schema),
+    )
+}
+
 #[test]
-fn join_residual_cast_targets_are_owned_by_the_contract() {
-    assert_eq!(join_residual_cast_mode(&DataType::Int64).unwrap(), super::JoinResidualCastMode::Ast);
-    assert_eq!(
-        join_residual_cast_mode(&DataType::Decimal128(12, 2)).unwrap(),
-        super::JoinResidualCastMode::Materialize
+fn join_residual_moves_single_input_expressions_outside_the_ast_before_the_join() {
+    let (left_schema, right_schema, schema) = join_residual_schemas();
+    let column = |name: &str| col(name, &schema).unwrap();
+    let text = |value: &str| lit(ScalarValue::Utf8(Some(value.to_string())));
+    let int = |value: i32| lit(ScalarValue::Int32(Some(value)));
+    let like = |negated: bool, expr: Arc<dyn PhysicalExpr>| -> Arc<dyn PhysicalExpr> {
+        Arc::new(LikeExpr::new(negated, false, expr, text("%bar%")))
+    };
+    let abs = scalar_function("abs", math::abs(), vec![column("n")], DataType::Int32);
+    let single_input_case: Arc<dyn PhysicalExpr> = Arc::new(
+        CaseExpr::try_new(
+            None,
+            vec![(binary(column("n"), Operator::Gt, int(0), &schema).unwrap(), column("n"))],
+            Some(int(0)),
+        )
+        .unwrap(),
     );
-    let rejection = join_residual_cast_mode(&DataType::Date32).unwrap_err();
+    let or_right_positive = |site: Arc<dyn PhysicalExpr>| {
+        let right_positive = binary(column("r"), Operator::Gt, int(0), &schema).unwrap();
+        (Arc::clone(&site), binary(site, Operator::Or, right_positive, &schema).unwrap())
+    };
+    let compared = |site: Arc<dyn PhysicalExpr>, operator: Operator, other: Arc<dyn PhysicalExpr>| {
+        (Arc::clone(&site), binary(site, operator, other, &schema).unwrap())
+    };
+    let cases = vec![
+        ("like", or_right_positive(like(false, column("name")))),
+        ("not like", or_right_positive(like(true, column("name")))),
+        (
+            "like over lower",
+            or_right_positive(like(
+                false,
+                scalar_function("lower", string::lower(), vec![column("name")], DataType::Utf8),
+            )),
+        ),
+        (
+            "is distinct from",
+            or_right_positive(binary(column("n"), Operator::IsDistinctFrom, int(5), &schema).unwrap()),
+        ),
+        ("abs", compared(abs, Operator::Gt, column("r"))),
+        (
+            "concatenation",
+            compared(
+                binary(column("name"), Operator::StringConcat, text("!"), &schema).unwrap(),
+                Operator::Eq,
+                column("label"),
+            ),
+        ),
+        ("try cast", compared(Arc::new(TryCastExpr::new(column("name"), DataType::Int32)), Operator::Eq, column("r"))),
+        (
+            "timestamp to bigint",
+            compared(
+                Arc::new(CastExpr::new(column("t"), DataType::Int64, None)),
+                Operator::Gt,
+                Arc::new(CastExpr::new(column("r"), DataType::Int64, None)),
+            ),
+        ),
+        (
+            "integer division by a literal",
+            compared(binary(column("n"), Operator::Divide, int(2), &schema).unwrap(), Operator::Eq, column("r")),
+        ),
+        ("single-input case", compared(single_input_case, Operator::Eq, column("r"))),
+    ];
+    for (label, (site, residual)) in cases {
+        let filter = join_residual_filter(residual, &schema);
+        let validated = validate_join_residual(&filter, &left_schema, &right_schema)
+            .unwrap_or_else(|error| panic!("{label} must validate: {error:?}"));
+        let sites = validated.materializations();
+        assert_eq!(sites.len(), 1, "{label}: {sites:?}");
+        assert!(Arc::ptr_eq(&sites[0].expr, &site), "{label} materializes {:?}", sites[0].expr);
+        assert_eq!(sites[0].side, JoinSide::Left, "{label}");
+    }
+
+    let ast_only = binary(
+        binary(binary(column("n"), Operator::Plus, int(1), &schema).unwrap(), Operator::Gt, column("r"), &schema)
+            .unwrap(),
+        Operator::And,
+        Arc::new(NotExpr::new(Arc::new(IsNullExpr::new(column("n"))))),
+        &schema,
+    )
+    .unwrap();
+    let filter = join_residual_filter(ast_only, &schema);
+    let validated = validate_join_residual(&filter, &left_schema, &right_schema).unwrap();
+    assert!(validated.materializations().is_empty(), "the AST evaluates arithmetic, comparison and NOT");
+}
+
+#[test]
+fn join_residual_keeps_operations_that_can_fail_out_of_pre_join_columns() {
+    let (left_schema, right_schema, schema) = join_residual_schemas();
+    let column = |name: &str| col(name, &schema).unwrap();
+    let to_decimal = |name: &str, precision: u8| -> Arc<dyn PhysicalExpr> {
+        Arc::new(CastExpr::new(column(name), DataType::Decimal128(precision, 0), None))
+    };
+    let narrowing_subject =
+        ExpressionRejectionSubject::Cast { source: DataType::Int32, target: DataType::Decimal128(4, 0) };
+
+    // An Int32 does not always fit four digits, and a pre-join column would
+    // also cast rows that match no key.
+    let narrowing = binary(to_decimal("n", 4), Operator::Eq, to_decimal("r", 10), &schema).unwrap();
+    let rejection =
+        validate_join_residual(&join_residual_filter(narrowing, &schema), &left_schema, &right_schema).unwrap_err();
     assert_eq!(rejection.capability(), ExpressionCapability::JoinResidualCastTarget);
+    assert_eq!(rejection.subject(), &narrowing_subject);
+    assert_eq!(rejection.sql(), Some("CAST(n AS Decimal128(4, 0))"));
+
+    let guarded: Arc<dyn PhysicalExpr> = Arc::new(
+        CaseExpr::try_new(
+            None,
+            vec![(
+                binary(column("n"), Operator::Gt, lit(ScalarValue::Int32(Some(0))), &schema).unwrap(),
+                to_decimal("n", 4),
+            )],
+            None,
+        )
+        .unwrap(),
+    );
+    let guarded = binary(guarded, Operator::Eq, to_decimal("r", 10), &schema).unwrap();
+    let rejection =
+        validate_join_residual(&join_residual_filter(guarded, &schema), &left_schema, &right_schema).unwrap_err();
+    assert_eq!(rejection.subject(), &narrowing_subject, "the rejection names the cast inside the CASE");
+
+    // The checked post-join filter of a decimal division evaluates matched
+    // pairs only, so the narrowing cast stays in the residual there.
+    let left_cast = to_decimal("n", 4);
+    let right_cast = to_decimal("r", 10);
+    let quotient = binary(Arc::clone(&left_cast), Operator::Divide, Arc::clone(&right_cast), &schema).unwrap();
+    let post_join = binary(quotient, Operator::Gt, lit(ScalarValue::Decimal128(Some(1), 10, 0)), &schema).unwrap();
+    let filter = join_residual_filter(post_join, &schema);
+    let validated = validate_join_residual(&filter, &left_schema, &right_schema).unwrap();
+    let sites = validated.materializations();
+    assert_eq!(sites.len(), 1, "{sites:?}");
+    assert!(Arc::ptr_eq(&sites[0].expr, &right_cast), "only the cast that always fits moves before the join");
 }
 
 #[test]
-fn join_residual_scalar_functions_add_a_stricter_contract_layer() {
-    validate_join_residual_scalar_function(ScalarFunctionKind::Substring).unwrap();
-    let rejection = validate_join_residual_scalar_function(ScalarFunctionKind::Abs).unwrap_err();
+fn join_residual_rejects_cross_input_expressions_neither_evaluator_runs() {
+    let (left_schema, right_schema, schema) = join_residual_schemas();
+    let column = |name: &str| col(name, &schema).unwrap();
+    let validate = |expr: Arc<dyn PhysicalExpr>| {
+        validate_join_residual(&join_residual_filter(expr, &schema), &left_schema, &right_schema).map(|_| ())
+    };
+
+    let concatenation = binary(
+        binary(column("name"), Operator::StringConcat, column("label"), &schema).unwrap(),
+        Operator::Eq,
+        lit(ScalarValue::Utf8(Some("x".to_string()))),
+        &schema,
+    )
+    .unwrap();
+    let rejection = validate(concatenation).unwrap_err();
+    assert_eq!(rejection.capability(), ExpressionCapability::JoinResidualShape);
+    assert!(matches!(
+        rejection.subject(),
+        ExpressionRejectionSubject::BinaryOperator { operator: Operator::StringConcat, .. }
+    ));
+
+    let sum = binary(column("n"), Operator::Plus, column("r"), &schema).unwrap();
+    let widened = binary(
+        Arc::new(CastExpr::new(sum, DataType::Decimal128(12, 0), None)),
+        Operator::Gt,
+        lit(ScalarValue::Decimal128(Some(5), 12, 0)),
+        &schema,
+    )
+    .unwrap();
+    let rejection = validate(widened).unwrap_err();
+    assert_eq!(rejection.capability(), ExpressionCapability::JoinResidualCastTarget);
+    assert_eq!(
+        rejection.subject(),
+        &ExpressionRejectionSubject::Cast { source: DataType::Int32, target: DataType::Decimal128(12, 0) }
+    );
+
+    let difference = binary(column("n"), Operator::Minus, column("r"), &schema).unwrap();
+    let distance = binary(
+        scalar_function("abs", math::abs(), vec![difference], DataType::Int32),
+        Operator::Lt,
+        lit(ScalarValue::Int32(Some(5))),
+        &schema,
+    )
+    .unwrap();
+    let rejection = validate(distance).unwrap_err();
     assert_eq!(rejection.capability(), ExpressionCapability::JoinResidualScalarFunction);
+    assert_eq!(rejection.subject(), &ExpressionRejectionSubject::ScalarFunction("abs".to_string()));
+
+    // An inner join runs a null-aware comparison as a post-join filter.
+    validate(binary(column("n"), Operator::IsDistinctFrom, column("r"), &schema).unwrap())
+        .expect("a cross-input IS DISTINCT FROM stays in the residual");
 }
 
 #[test]
@@ -665,11 +869,6 @@ fn join_residual_ast_rejects_integer_division_and_casts_it_cannot_evaluate_or_ch
         binary(elapsed, Operator::Gt, lit(ScalarValue::DurationMillisecond(Some(5))), &filter_schema).unwrap();
     assert_eq!(validate(elapsed_bound).unwrap_err().capability(), ExpressionCapability::JoinResidualShape);
 
-    // The cast cannot fail, but the AST has no cast over temporal values.
-    let temporal_cast =
-        binary(cast("t", DataType::Int64), Operator::Eq, col("r", &filter_schema).unwrap(), &filter_schema).unwrap();
-    assert_eq!(validate(temporal_cast).unwrap_err().capability(), ExpressionCapability::CastPair);
-
     let failing_cast =
         binary(cast("l", DataType::UInt64), Operator::Eq, cast("r", DataType::UInt64), &filter_schema).unwrap();
     assert_eq!(validate(failing_cast).unwrap_err().capability(), ExpressionCapability::CastPair);

crates/algeon-datafusion/src/native/lowering/expr.rs

@@ -81,20 +81,31 @@ pub(super) fn lower_validated_physical_expr_with_bindings(
     let activation_nodes = Cell::new(0);
     lower_validated_physical_expr(
         validated,
-        LoweringScope { bindings, join_columns: None, activation: None, activation_nodes: &activation_nodes },
+        LoweringScope {
+            bindings,
+            join_columns: None,
+            join_substitutions: &[],
+            activation: None,
+            activation_nodes: &activation_nodes,
+        },
     )
 }
 
+/// Lowers a join filter expression whose column `i` reads `columns[&i]`.
+/// Each subexpression in `substitutions`, matched by pointer identity, lowers
+/// to its paired expression instead.
 pub(super) fn lower_validated_join_residual_expr(
     validated: &ValidatedPhysicalExpr,
     columns: &BTreeMap<usize, ColumnRef>,
+    substitutions: &[(Arc<dyn PhysicalExpr>, NativeExpr)],
 ) -> Result<NativeExpr, LoweringRejection> {
     let activation_nodes = Cell::new(0);
     lower_validated_physical_expr(
         validated,
         LoweringScope {
             bindings: None,
             join_columns: Some(columns),
+            join_substitutions: substitutions,
             activation: None,
             activation_nodes: &activation_nodes,
         },
@@ -106,6 +117,7 @@ pub(super) fn lower_validated_join_residual_expr(
 struct LoweringScope<'a> {
     bindings: Option<&'a [FieldBinding]>,
     join_columns: Option<&'a BTreeMap<usize, ColumnRef>>,
+    join_substitutions: &'a [(Arc<dyn PhysicalExpr>, NativeExpr)],
     /// Set inside a `CASE` branch or the right operand of an `AND` or `OR`,
     /// which DataFusion may evaluate on a subset of rows: the rows where this
     /// boolean is TRUE. Native evaluation visits every row, so an operand that
@@ -150,6 +162,9 @@ fn lower_validated_physical_expr(
 ) -> Result<NativeExpr, LoweringRejection> {
     let expr = validated.expr();
     let input_schema = validated.input_schema();
+    if let Some((_, substitute)) = scope.join_substitutions.iter().find(|(site, _)| Arc::ptr_eq(site, expr)) {
+        return Ok(substitute.clone());
+    }
     if let Some(column) = expr.downcast_ref::<Column>() {
         if let Some(columns) = scope.join_columns {
             let resolved = resolve_column(column, input_schema)?;

crates/algeon-datafusion/src/native/lowering/join.rs

@@ -157,9 +157,9 @@ fn add_join_residual_materializations(
     input: LoweredNativeInput,
     materializations: &[JoinResidualMaterialization],
 ) -> Result<LoweredNativeInput, LoweringFailure> {
-    // Residual casts that cannot live in the predicate AST are appended as
-    // synthetic pre-join projection columns on the original DataFusion side.
-    // Canonical input ordering happens after this step.
+    // Residual subexpressions that the contract moves before the join are
+    // appended as synthetic projection columns on the original DataFusion
+    // side. Canonical input ordering happens after this step.
     if materializations.is_empty() {
         return Ok(input);
     }