test(compiler): From index to inner join

This patch removes the invalid translation from IndexJoin to JoinExpr,
since there is no way to capture the implicit projection of IndexJoin.
Instead we translate from IndexJoin directly to QueryExpr.

Author: joshua-spacetime

crates/core/src/vm.rs

@@ -65,11 +65,7 @@ pub fn build_query<'a>(
                     index_side: Table::MemTable(_),
                     ..
                 },
-            ) => {
-                let join: JoinExpr = join.into();
-                let iter = join_inner(ctx, stdb, tx, result, join, true)?;
-                Box::new(iter)
-            }
+            ) => build_query(ctx, stdb, tx, join.to_inner_join().into())?,
             Query::IndexJoin(IndexJoin {
                 probe_side,
                 probe_field,

crates/vm/src/expr.rs

@@ -504,6 +504,63 @@ impl IndexJoin {
             }
         }
     }
+
+    // Convert this index join to an inner join, followed by a projection.
+    // This is needed for incremental evaluation of index joins.
+    // In particular when there are updates to both the left and right tables.
+    // In other words, when an index join has two delta tables.
+    pub fn to_inner_join(self) -> QueryExpr {
+        if self.return_index_rows {
+            let col_lhs = self.index_side.head().fields[usize::from(self.index_col)].field.clone();
+            let col_rhs = self.probe_field;
+            let rhs = self.probe_side;
+
+            let fields = self
+                .index_side
+                .head()
+                .fields
+                .iter()
+                .cloned()
+                .map(|Column { field, .. }| field.into())
+                .collect();
+
+            let table = self.index_side.get_db_table().map(|t| t.table_id);
+            let source = self.index_side.into();
+            let inner_join = Query::JoinInner(JoinExpr::new(rhs, col_lhs, col_rhs));
+            let project = Query::Project(fields, table);
+            let query = if let Some(predicate) = self.index_select {
+                vec![predicate.into(), inner_join, project]
+            } else {
+                vec![inner_join, project]
+            };
+            QueryExpr { source, query }
+        } else {
+            let col_rhs = self.index_side.head().fields[usize::from(self.index_col)].field.clone();
+            let col_lhs = self.probe_field;
+            let mut rhs: QueryExpr = self.index_side.into();
+
+            if let Some(predicate) = self.index_select {
+                rhs.query.push(predicate.into());
+            }
+
+            let fields = self
+                .probe_side
+                .source
+                .head()
+                .fields
+                .iter()
+                .cloned()
+                .map(|Column { field, .. }| field.into())
+                .collect();
+
+            let table = self.probe_side.source.get_db_table().map(|t| t.table_id);
+            let source = self.probe_side.source;
+            let inner_join = Query::JoinInner(JoinExpr::new(rhs, col_lhs, col_rhs));
+            let project = Query::Project(fields, table);
+            let query = vec![inner_join, project];
+            QueryExpr { source, query }
+        }
+    }
 }
 
 #[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord)]
@@ -513,25 +570,6 @@ pub struct JoinExpr {
     pub col_rhs: FieldName,
 }
 
-//TODO: A posterior PR should fix the pass of multi-columns to the query optimization
-impl From<IndexJoin> for JoinExpr {
-    fn from(mut value: IndexJoin) -> Self {
-        if let Some(predicate) = value.index_select {
-            value.probe_side.query.push(predicate.into());
-        };
-        let pos = value.index_col;
-        if value.return_index_rows {
-            let col_lhs = value.index_side.head().fields[usize::from(pos)].field.clone();
-            let col_rhs = value.probe_field;
-            JoinExpr::new(value.probe_side, col_lhs, col_rhs)
-        } else {
-            let col_rhs = value.index_side.head().fields[usize::from(pos)].field.clone();
-            let col_lhs = value.probe_field;
-            JoinExpr::new(value.index_side.into(), col_lhs, col_rhs)
-        }
-    }
-}
-
 impl JoinExpr {
     pub fn new(rhs: QueryExpr, col_lhs: FieldName, col_rhs: FieldName) -> Self {
         Self { rhs, col_lhs, col_rhs }
@@ -1931,6 +1969,88 @@ mod tests {
         assert!(matches!(auth.check_auth(ALICE, BOB), Err(AuthError::OwnerRequired)));
     }
 
+    fn mem_table(name: &str, fields: &[(&str, AlgebraicType, bool)]) -> MemTable {
+        let table_access = StAccess::Public;
+        let data = Vec::new();
+        let head = Header::new(
+            name.into(),
+            fields
+                .iter()
+                .enumerate()
+                .map(|(i, (field, ty, _))| Column::new(FieldName::named(name, field), ty.clone(), i.into()))
+                .collect(),
+            fields
+                .iter()
+                .enumerate()
+                .filter(|(_, (_, _, indexed))| *indexed)
+                .map(|(i, _)| (ColId(i as u32).into(), Constraints::indexed()))
+                .collect(),
+        );
+        MemTable {
+            head,
+            data,
+            table_access,
+        }
+    }
+
+    #[test]
+    fn test_index_to_inner_join() {
+        let index_side = mem_table(
+            "index",
+            &[("a", AlgebraicType::U8, false), ("b", AlgebraicType::U8, true)],
+        );
+        let probe_side = mem_table(
+            "probe",
+            &[("c", AlgebraicType::U8, false), ("b", AlgebraicType::U8, true)],
+        );
+
+        let probe_field = probe_side.head.fields[1].field.clone();
+        let select_field = FieldName::Name {
+            table: "index".into(),
+            field: "a".into(),
+        };
+        let index_select = ColumnOp::cmp(select_field, OpCmp::Eq, 0.into());
+        let join = IndexJoin {
+            probe_side: probe_side.clone().into(),
+            probe_field,
+            index_side: index_side.clone().into(),
+            index_select: Some(index_select.clone()),
+            index_col: 1.into(),
+            return_index_rows: false,
+        };
+
+        let expr = join.to_inner_join();
+
+        assert_eq!(expr.source, SourceExpr::MemTable(probe_side));
+        assert_eq!(expr.query.len(), 2);
+
+        let Query::JoinInner(ref join) = expr.query[0] else {
+            panic!("expected an inner join, but got {:#?}", expr.query[0]);
+        };
+
+        assert_eq!(join.col_lhs, FieldName::named("probe", "b"));
+        assert_eq!(join.col_rhs, FieldName::named("index", "b"));
+        assert_eq!(
+            join.rhs,
+            QueryExpr {
+                source: SourceExpr::MemTable(index_side),
+                query: vec![index_select.into()]
+            }
+        );
+
+        let Query::Project(ref fields, None) = expr.query[1] else {
+            panic!("expected a projection, but got {:#?}", expr.query[1]);
+        };
+
+        assert_eq!(
+            fields,
+            &vec![
+                FieldName::named("probe", "c").into(),
+                FieldName::named("probe", "b").into()
+            ]
+        );
+    }
+
     #[test]
     fn test_auth_table() {
         tables().iter().for_each(assert_owner_private)