Feature/refactor planning (#726)

This PR replaces `ChaseRule` and `RuleAnalysis` with `NormalizedRule`s.
It also reworks the planning code.

Author: aannleax

nemo-physical/src/management/database.rs

@@ -546,6 +546,7 @@ impl DatabaseInstance {
             let (result_tree, results_dependent) = self
                 .execute_tree(&temporary_storage, tree, dependent_reorderings)
                 .await;
+
             TimedCode::instance().sub(&timed_string).stop();
 
             if tree_used > 0 {

nemo-physical/src/management/execution_plan.rs

@@ -155,7 +155,8 @@ impl ExecutionNodeRef {
             | ExecutionOperation::Function(subnode, _)
             | ExecutionOperation::Null(subnode)
             | ExecutionOperation::Aggregate(subnode, _)
-            | ExecutionOperation::IncrementalImport(subnode, _) => vec![subnode.clone()],
+            | ExecutionOperation::IncrementalImport(subnode, _)
+            | ExecutionOperation::Rename(subnode) => vec![subnode.clone()],
         }
     }
 }
@@ -197,6 +198,8 @@ pub(crate) enum ExecutionOperation {
     Aggregate(ExecutionNodeRef, AggregateAssignment),
     /// Perform an incremental import
     IncrementalImport(ExecutionNodeRef, Rc<Box<dyn TableProvider>>),
+    /// Renaming of the columns of the table
+    Rename(ExecutionNodeRef),
 }
 
 /// Declares whether the resulting table form executing a plan should be kept temporarily or permamently
@@ -421,6 +424,16 @@ impl ExecutionPlan {
 
         import_node
     }
+
+    /// Return an [ExecutionNodeRef] with renamed column markers.
+    pub fn rename(
+        &mut self,
+        subnode: ExecutionNodeRef,
+        markers: OperationTable,
+    ) -> ExecutionNodeRef {
+        let operation = ExecutionOperation::Rename(subnode);
+        self.push_and_return_reference(operation, markers)
+    }
 }
 
 impl ExecutionPlan {
@@ -732,6 +745,7 @@ impl ExecutionPlan {
             }
             ExecutionOperation::Subtract(subnode_main, subnodes_subtract) => {
                 let markers_main = node_markers;
+
                 let (markers_subtract, subtract_reorderings): (
                     Vec<OperationTable>,
                     Vec<Permutation>,
@@ -905,6 +919,15 @@ impl ExecutionPlan {
                     subnode: subtree,
                 }
             }
+            ExecutionOperation::Rename(subnode) => Self::execution_node(
+                root_node_id,
+                subnode.clone(),
+                order,
+                output_nodes,
+                computed_trees,
+                computed_trees_map,
+                loaded_tables,
+            ),
         }
     }
 

nemo-physical/src/tabular/buffer/tuple_buffer.rs

@@ -477,18 +477,20 @@ impl TupleBuffer {
             return true;
         }
 
-        for pattern in &self.patterns {
+        'pattern: for pattern in &self.patterns {
             if !self.pattern_matches(pattern) {
                 continue;
             }
 
             let data_values = self.current_tuple_data_values.clone();
             for transformation in &pattern.transformations {
-                let value = transformation
-                    .program
-                    .evaluate_data(&data_values)
-                    .expect("should evaluate to a value");
-                self.current_tuple_data_values[transformation.position] = value;
+                if let Some(value) = transformation.program.evaluate_data(&data_values) {
+                    self.current_tuple_data_values[transformation.position] = value;
+                } else {
+                    // We skip this pattern if the operation did not return a result,
+                    // e.g. when using a builtin with a value of an incorrect type
+                    continue 'pattern;
+                }
             }
 
             // first matching pattern wins

nemo-physical/src/tabular/operations.rs

@@ -196,7 +196,7 @@ impl Deref for OperationTable {
 /// mainly by providing a translation between
 /// user defined "markers" (supplied as a generic parameter to this object)
 /// and [OperationColumnMarker]s.
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct OperationTableGenerator<ExternalMarker>
 where
     ExternalMarker: Clone + PartialEq + Eq + Hash,

nemo-python/src/lib.rs

@@ -2,7 +2,6 @@ use std::{collections::HashSet, fs::read_to_string, time::Duration};
 
 use nemo::{
     api::load_program,
-    chase_model::ChaseAtom,
     datavalues::{AnyDataValue, DataValue},
     error::Error,
     execution::{ExecutionEngine, tracing::trace::ExecutionTraceTree},
@@ -201,7 +200,7 @@ fn datavalue_to_python(py: Python<'_>, v: AnyDataValue) -> PyResult<Bound<'_, Py
 }
 
 #[pyclass]
-struct NemoFact(nemo::chase_model::GroundAtom);
+struct NemoFact(nemo::execution::planning::normalization::atom::ground::GroundAtom);
 
 #[pymethods]
 impl NemoFact {

nemo-python/tests/test_example.py

@@ -41,8 +41,8 @@ def setUp(self):
         self.engine.reason()
 
         self.expected_api_result = [
-            ["<world>", "_:0"],
-            ["<circle>", "_:1"],
+            ["<circle>", "_:0"],
+            ["<world>", "_:1"],
             [
                 NemoLiteral("hello world", lang="en"),
                 "_:2",
@@ -53,8 +53,8 @@ def setUp(self):
         ]
 
         self.expected_serialized_result = [
-            ["world", "_:0"],
-            ["circle", "_:1"],
+            ["circle", "_:0"],
+            ["world", "_:1"],
             [
                 '"hello world"@en',
                 "_:2",

nemo/Cargo.toml

@@ -65,5 +65,5 @@ assert_fs = "1.0"
 test-log = "0.2"
 quickcheck = "1"
 quickcheck_macros = "1"
-tokio = "1.47.1"
+tokio = { version = "1.47.1", features = [ "rt", "macros" ] }
 

nemo/src/api.rs

@@ -125,8 +125,7 @@ pub fn output_predicates(engine: &Engine) -> Vec<Tag> {
         .exports()
         .iter()
         .map(|export| export.predicate())
-        .cloned()
-        .collect()
+        .collect::<Vec<_>>()
 }
 
 // TODO: Disabled write API. This API is designed in a way that does not fit how Nemo controls exporting.