Network execution

engine/src/element.rs

@@ -6,6 +6,7 @@ pub enum Element {
     Raw(String),
     View(String, String),
     Spec(String, String),
+    Array(Vec<Self>),
 }
 
 impl Element {
@@ -21,6 +22,10 @@ impl Element {
         Self::Spec(source.to_string(), target.to_string())
     }
 
+    pub(crate) fn array(values: Vec<Self>) -> Self {
+        Self::Array(values)
+    }
+
     pub fn as_raw(&self) -> DbResult<&str> {
         match self {
             Self::Raw(s) => Ok(s),
@@ -33,6 +38,13 @@ impl Element {
             Self::Raw(s) => s.len(),
             Self::View(source, name) => source.len() + 4 + name.len(),
             Self::Spec(source, target) => source.len() + 4 + target.len(),
+            Self::Array(values) => {
+                let mut len = 0;
+                for value in values {
+                    len += value.len();
+                }
+                len
+            }
         }
     }
 
@@ -47,6 +59,17 @@ impl std::fmt::Display for Element {
             Self::Raw(value) => write!(f, "{}", value),
             Self::View(source, name) => write!(f, "{} {}", source, name),
             Self::Spec(source, target) => write!(f, "{}.{}", source, target),
+            Self::Array(values) => {
+                write!(f, "(")?;
+                for (i, value) in values.iter().enumerate() {
+                    if i == 0 {
+                        write!(f, "{}", value)?;
+                    } else {
+                        write!(f, "{},", value)?;
+                    }
+                }
+                write!(f, ")")
+            }
         }
     }
 }
@@ -66,4 +89,10 @@ mod tests {
         let spec = Element::spec("t1", "f1");
         assert_eq!("t1.f1", spec.to_string());
     }
+
+    #[test]
+    fn array() {
+        let spec = Element::array(vec![Element::view("table", "t")]);
+        assert_eq!("(table t)", spec.to_string());
+    }
 }

engine/src/index.rs

@@ -42,7 +42,7 @@ mod tests {
         let majorid = Element::raw("majorid");
 
         let setup_tx = db.get_tx().unwrap();
-        let schema = SchemaBuilder::default()
+        let schema = SchemaBuilder::new(Element::raw(table))
             .add_int_field(sid.clone())
             .add_string_field(sname.clone(), 16)
             .add_int_field(majorid.clone())

engine/src/index_mgr.rs

@@ -87,7 +87,7 @@ impl IndexMgr {
         tx: &Arc<Transaction>,
     ) -> DbResult<Self> {
         if is_new {
-            let schema = SchemaBuilder::default()
+            let schema = SchemaBuilder::new(Element::raw(IDX_TABLE))
                 .add_string_field(Element::raw(IDX_NAME), MAX_LENGTH)
                 .add_string_field(Element::raw(TABLE_NAME), MAX_LENGTH)
                 .add_string_field(Element::raw(FIELD_NAME), MAX_LENGTH)

engine/src/layout.rs

@@ -84,7 +84,7 @@ mod tests {
 
     #[test]
     fn from() {
-        let schema = SchemaBuilder::default()
+        let schema = SchemaBuilder::new(Element::raw("test"))
             .add_int_field(Element::raw("id"))
             .build();
         Layout::from(schema, HashMap::new(), 16);

engine/src/lib.rs

@@ -29,6 +29,7 @@ pub mod record_page;
 pub mod rid;
 pub mod scan;
 pub mod schema;
+mod schema_mapping;
 pub mod sort_by;
 pub mod stat_mgr;
 pub mod table_mgr;
@@ -196,7 +197,7 @@ mod tests {
             .unwrap();
         }
         let result = db
-            .query(&tx, "SELECT id, name FROM test SORT BY name")
+            .query(&tx, "SELECT id, name FROM test ORDER BY name")
             .unwrap();
         for i in 0..1000 {
             assert!(result.next().unwrap());
@@ -248,30 +249,93 @@ mod tests {
         let dir = tempdir().unwrap();
         let db = SimpleDB::new(dir.path()).unwrap();
         let tx = db.get_tx().unwrap();
-        db.execute(&tx, "CREATE TABLE t1(i1 INT)").unwrap();
-        db.execute(&tx, "CREATE TABLE t2(i2 INT)").unwrap();
+        db.execute(&tx, "CREATE TABLE users(id INT, name VARCHAR(20))")
+            .unwrap();
+        db.execute(&tx, "CREATE TABLE employees(eid INT, uid INT)")
+            .unwrap();
         tx.commit().unwrap();
         for i in 0..100 {
-            db.execute(&tx, &format!("INSERT INTO t1(i1) VALUES({})", i))
-                .unwrap();
-            db.execute(&tx, &format!("INSERT INTO t2(i2) VALUES({})", i))
-                .unwrap();
+            db.execute(
+                &tx,
+                &format!("INSERT INTO users(id, name) VALUES({}, 'user{}')", i, i),
+            )
+            .unwrap();
+            db.execute(
+                &tx,
+                &format!(
+                    "INSERT INTO employees(eid, uid) VALUES({}, {})",
+                    i + 1000,
+                    i
+                ),
+            )
+            .unwrap();
         }
         tx.commit().unwrap();
-        let mut existed = HashSet::new();
-        for i in 0..10 {
-            existed.insert(i);
-            let result = db
-                .query(
-                    &tx,
-                    &format!("SELECT i1, i2 FROM t1, t2 WHERE i1 = i2 AND i1 = {}", i),
-                )
-                .unwrap();
-            while result.next().unwrap() {
-                assert_eq!(i, result.get_i32(&Element::raw("i1")).unwrap());
-                assert_eq!(i, result.get_i32(&Element::raw("i2")).unwrap());
-            }
+
+        let result = db
+            .query(
+                &tx,
+                "SELECT id, name, eid FROM users JOIN employees ON id = uid",
+            )
+            .unwrap();
+        let mut matched = HashSet::new();
+        while result.next().unwrap() {
+            let id = result.get_i32(&Element::raw("id")).unwrap();
+            let eid = result.get_i32(&Element::raw("eid")).unwrap();
+            let name = result.get_string(&Element::raw("name")).unwrap();
+            assert_eq!(eid, id + 1000);
+            assert_eq!(name, format!("user{}", id));
+            assert!(matched.insert(id), "each user must match exactly once");
         }
+        assert_eq!(matched.len(), 100);
         tx.commit().unwrap();
     }
+
+    #[test]
+    fn select_with_views_and_specs() {
+        let dir = tempdir().unwrap();
+        let db = SimpleDB::new(dir.path()).unwrap();
+        let tx = db.get_tx().unwrap();
+        db.execute(&tx, "CREATE TABLE test(id INT, name VARCHAR(100))")
+            .unwrap();
+        tx.commit().unwrap();
+
+        db.execute(&tx, "INSERT INTO test(id, name) VALUES(1, 'User')")
+            .unwrap();
+
+        let result = db
+            .query(&tx, "SELECT id i, name n FROM test t WHERE t.id=1")
+            .unwrap();
+        assert!(result.next().unwrap());
+        assert_eq!(result.get_i32(&Element::raw("i")).unwrap(), 1);
+        assert_eq!(result.get_string(&Element::raw("n")).unwrap(), "User");
+
+        let result = db
+            .query(&tx, "SELECT id i, name n FROM test t WHERE t.id=2")
+            .unwrap();
+        assert!(!result.next().unwrap());
+    }
+
+    #[test]
+    fn select_with_specs() {
+        let dir = tempdir().unwrap();
+        let db = SimpleDB::new(dir.path()).unwrap();
+        let tx = db.get_tx().unwrap();
+        db.execute(&tx, "CREATE TABLE test(id INT, name VARCHAR(100))")
+            .unwrap();
+        tx.commit().unwrap();
+
+        db.execute(&tx, "INSERT INTO test(id, name) VALUES(1, 'User')")
+            .unwrap();
+
+        let result = db
+            .query(&tx, "SELECT t.id, t.name FROM test t WHERE t.id=1")
+            .unwrap();
+        assert!(result.next().unwrap());
+        assert_eq!(result.get_i32(&Element::spec("t", "id")).unwrap(), 1);
+        assert_eq!(
+            result.get_string(&Element::spec("t", "name")).unwrap(),
+            "User"
+        );
+    }
 }

engine/src/plan.rs

@@ -1,9 +1,8 @@
 use std::rc::Rc;
 
-use common::DbResult;
-
 use crate::element::Element;
 use crate::{scan::Scan, schema::Schema};
+use common::DbResult;
 
 pub(crate) mod group;
 pub mod index;

engine/src/plan/group.rs

@@ -25,7 +25,7 @@ impl GroupByPlan {
         group_fields: Vec<Element>,
         aggregation_fn: Vec<AggregationFn>,
     ) -> DbResult<Self> {
-        let mut schema = SchemaBuilder::default();
+        let mut schema = SchemaBuilder::new(plan.schema()?.table().clone());
         let s = plan.schema()?;
         for field in &group_fields {
             schema = schema.add(field.clone(), &s);

engine/src/plan/index.rs

@@ -79,7 +79,14 @@ impl IndexJoinPlan {
     ) -> DbResult<Self> {
         let s1 = p1.schema()?;
         let s2 = p2.schema()?;
-        let schema = SchemaBuilder::default().add_all(&s1).add_all(&s2).build();
+        let schema = SchemaBuilder::new(Element::Raw(format!(
+            "index_join_{}_{}",
+            p1.schema()?.table(),
+            p2.schema()?.table()
+        )))
+        .add_all(&s1)
+        .add_all(&s2)
+        .build();
         Ok(Self {
             p1: Rc::clone(p1),
             p2: Rc::clone(p2),

engine/src/plan/merge.rs

@@ -30,7 +30,12 @@ impl MergeJoinPlan {
     ) -> DbResult<Self> {
         let p1 = Rc::new(SortPlan::new(tx, p1, vec![field_name1.clone()])?);
         let p2 = Rc::new(SortPlan::new(tx, p2, vec![field_name2.clone()])?);
-        let schema = SchemaBuilder::default().build();
+        let schema = SchemaBuilder::new(Element::Raw(format!(
+            "merge_join_{}_{}",
+            p1.schema()?.table(),
+            p2.schema()?.table()
+        )))
+        .build();
         Ok(Self {
             p1,
             p2,

engine/src/plan/multibuffer.rs

@@ -27,7 +27,14 @@ impl MultiBufferProductPlan {
     ) -> DbResult<Self> {
         let s1 = left.schema()?;
         let s2 = right.schema()?;
-        let schema = SchemaBuilder::default().add_all(&s1).add_all(&s2).build();
+        let schema = SchemaBuilder::new(Element::Raw(format!(
+            "multibuffer_{}_{}",
+            left.schema()?.table(),
+            right.schema()?.table()
+        )))
+        .add_all(&s1)
+        .add_all(&s2)
+        .build();
         let plan = Self {
             tx: Arc::clone(tx),
             left: Rc::new(MaterializePlan::new(left, tx)),
@@ -59,10 +66,13 @@ impl Plan for MultiBufferProductPlan {
     fn open(&self) -> DbResult<Rc<dyn Scan>> {
         let left = self.left.open()?;
         let t = self.copy_records_from(&self.right)?;
+        // `TableScan` stores a temp table's records in `<name>.tbl`, so the
+        // multibuffer scan (which reads the blocks directly) must target that
+        // physical file rather than the bare table name.
         Ok(Rc::new(MultiBufferProductScan::new(
             &self.tx,
             &left,
-            &t.table_name(),
+            &format!("{}.tbl", t.table_name()),
             t.layout(),
         )?))
     }

engine/src/plan/product.rs

@@ -20,7 +20,14 @@ impl ProductPlan {
     pub fn new(p1: Rc<dyn Plan>, p2: Rc<dyn Plan>) -> DbResult<Self> {
         let s1 = p1.schema()?;
         let s2 = p2.schema()?;
-        let schema = SchemaBuilder::default().add_all(&s1).add_all(&s2).build();
+        let schema = SchemaBuilder::new(Element::Raw(format!(
+            "product_{}_{}",
+            s1.table(),
+            s2.table()
+        )))
+        .add_all(&s1)
+        .add_all(&s2)
+        .build();
         Ok(Self { p1, p2, schema })
     }
 }

engine/src/plan/project.rs

@@ -1,30 +1,56 @@
 use std::{collections::HashSet, rc::Rc};
 
-use common::DbResult;
-
 use crate::schema::SchemaBuilder;
+use crate::schema_mapping::SchemaMapping;
 use crate::{
     element::Element,
     plan::Plan,
     scan::{Scan, project::ProjectScan},
     schema::Schema,
 };
+use common::DbResult;
 
-pub struct ProjectPlan {
+pub(crate) struct ProjectPlan {
     plan: Rc<dyn Plan>,
     schema: Schema,
+    mapping: SchemaMapping,
 }
 
 impl ProjectPlan {
-    pub fn new(plan: Rc<dyn Plan>, fields: Vec<Element>) -> DbResult<Self> {
-        let mut schema = SchemaBuilder::default();
+    pub(crate) fn new(
+        plan: Rc<dyn Plan>,
+        fields: Vec<Element>,
+        mapping: SchemaMapping,
+    ) -> DbResult<Self> {
+        let mut schema = SchemaBuilder::new(plan.schema()?.table().clone());
+        let other = plan.schema()?;
         for field in fields {
-            let other = plan.schema()?;
-            schema = schema.add(field, &other);
+            let source_field = match &field {
+                Element::Spec(table, field) => {
+                    let source_table = Element::raw(table);
+                    let table = if let Some(Element::Raw(table)) = mapping.table(&source_table) {
+                        table
+                    } else {
+                        table
+                    };
+                    Element::Spec(table.to_string(), field.to_string())
+                }
+                field => {
+                    if let Some(field) = mapping.field(field) {
+                        field.clone()
+                    } else {
+                        field.clone()
+                    }
+                }
+            };
+            if let Some(info) = other.info(&source_field) {
+                schema = schema.add_field(field, info);
+            }
         }
         Ok(Self {
             plan,
             schema: schema.build(),
+            mapping,
         })
     }
 }
@@ -33,7 +59,11 @@ impl Plan for ProjectPlan {
     fn open(&self) -> DbResult<Rc<dyn Scan>> {
         let scan = self.plan.open()?;
         let fields: HashSet<Element> = self.schema.fields().into_iter().map(|(f, _)| f).collect();
-        Ok(Rc::new(ProjectScan::new(scan, fields)))
+        Ok(Rc::new(ProjectScan::new(
+            scan,
+            fields,
+            self.mapping.clone(),
+        )))
     }
 
     fn blocks_accessed(&self) -> DbResult<i32> {