Replace RootAndCurrentTables with TableScope, which keeps track of any tables in scope for an exists, instead of only root and current. (#674)

<!-- The PR description should answer 2 (maybe 3) important questions:
-->

### What

<!-- What is this PR trying to accomplish (and why, if it's not
obvious)? -->

`ComparisonTarget::RootCollectionColumn` was removed, to be replaced by
[named
scopes](https://github.com/hasura/ndc-spec/blob/36855ff20dcbd7d129427794aee9746b895390af/rfcs/0015-named-scopes.md).

This PR implements the replacement functionality.

<!-- Consider: do we need to add a changelog entry? -->

### How

<!-- How is it trying to accomplish it (what are the implementation
steps)? -->

This PR replaces RootAndCurrentTables, with TableScope, a struct that
keeps track of the current table and any tables in scope for exists
expression.

See the accompanying review for details on the code itself.

Author: BenoitRanque

crates/query-engine/translation/src/translation/error.rs

@@ -63,6 +63,11 @@ pub enum Error {
         scalar: models::ScalarTypeName,
         function: models::AggregateFunctionName,
     },
+    ScopeOutOfBounds {
+        current_collection_name: String,
+        tables_in_scope_names: Vec<String>,
+        scope: usize,
+    },
 }
 
 /// Capabilities we don't currently support.
@@ -237,6 +242,26 @@ impl std::fmt::Display for Error {
                 f,
                 "Missing single column aggregate function {function:?} for scalar type {scalar:?}"
             ),
+            Error::ScopeOutOfBounds {
+                current_collection_name,
+                tables_in_scope_names,
+                scope,
+            } => {
+                write!(
+                    f,
+                    "Scope {scope} out of bounds. Current collection is {current_collection_name}. Collections in scope: ["
+                )?;
+                let mut first = true;
+                for collection in tables_in_scope_names {
+                    if first {
+                        first = false;
+                    } else {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{collection}")?;
+                }
+                write!(f, "].")
+            }
         }
     }
 }

crates/query-engine/translation/src/translation/helpers.rs

@@ -1,6 +1,6 @@
 //! Helpers for processing requests and building SQL.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, VecDeque};
 
 use ndc_models as models;
 
@@ -47,16 +47,68 @@ pub struct NativeQueryInfo {
     pub alias: sql::ast::TableAlias,
 }
 
-/// For the root table in the query, and for the current table we are processing,
+/// For the current table we are processing, and all ancestor table up to the closest Query,
 /// We'd like to track what is their reference in the query (the name we can use to address them,
 /// an alias we generate), and what is their name in the metadata (so we can get
 /// their information such as which columns are available for that table).
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct RootAndCurrentTables {
-    /// The root (top-most) table in the query.
-    pub root_table: TableSourceAndReference,
+pub struct TableScope {
     /// The current table we are processing.
-    pub current_table: TableSourceAndReference,
+    current_table: TableSourceAndReference,
+    /// Tables in scope. Index 0 corresponds to scope 1, which is the table immediately above the current table in the exists chain.
+    tables_in_scope: VecDeque<TableSourceAndReference>,
+}
+
+impl TableScope {
+    /// Create a scope from a Query. There will be no tables available through scopes
+    pub fn new(current_table: TableSourceAndReference) -> Self {
+        Self {
+            current_table,
+            tables_in_scope: VecDeque::new(),
+        }
+    }
+    /// Create a scope from an exists expression or path. The ancestor tables up until the closest query will stay in scope
+    #[must_use]
+    pub fn new_from_scope(&self, current_table: TableSourceAndReference) -> Self {
+        let TableScope {
+            current_table: parent_table,
+            tables_in_scope,
+        } = self;
+        let mut tables_in_scope = tables_in_scope.clone();
+        tables_in_scope.push_front(parent_table.clone());
+        Self {
+            current_table,
+            tables_in_scope,
+        }
+    }
+    /// Get the table source and reference for the current table
+    pub fn current_table(&self) -> &TableSourceAndReference {
+        &self.current_table
+    }
+    /// Get the table source and reference for a table in scope.
+    /// The scope is an index, where 0 is the current table, 1 is the parent, and so on
+    /// Errors if the scope is out of bounds
+    pub fn scoped_table(&self, scope: &Option<usize>) -> Result<&TableSourceAndReference, Error> {
+        if let Some(scope) = scope {
+            if *scope > 0 && *scope <= self.tables_in_scope.len() {
+                Ok(&self.tables_in_scope[scope - 1])
+            } else if *scope == 0 {
+                Ok(&self.current_table)
+            } else {
+                Err(Error::ScopeOutOfBounds {
+                    current_collection_name: self.current_table.source.name_for_alias(),
+                    tables_in_scope_names: self
+                        .tables_in_scope
+                        .iter()
+                        .map(|c| c.source.name_for_alias())
+                        .collect(),
+                    scope: *scope,
+                })
+            }
+        } else {
+            Ok(&self.current_table)
+        }
+    }
 }
 
 /// For a table in the query, We'd like to track what is its reference in the query

crates/query-engine/translation/src/translation/mutation/v2/delete.rs

@@ -156,10 +156,7 @@ pub fn translate(
             let predicate_expression = filtering::translate(
                 env,
                 state,
-                &helpers::RootAndCurrentTables {
-                    root_table: table_name_and_reference.clone(),
-                    current_table: table_name_and_reference,
-                },
+                &helpers::TableScope::new(table_name_and_reference),
                 &predicate,
             )?;
 

crates/query-engine/translation/src/translation/mutation/v2/insert.rs

@@ -245,10 +245,7 @@ pub fn translate(
     let predicate_expression = filtering::translate(
         env,
         state,
-        &helpers::RootAndCurrentTables {
-            root_table: table_name_and_reference.clone(),
-            current_table: table_name_and_reference,
-        },
+        &helpers::TableScope::new(table_name_and_reference),
         &predicate,
     )?;
 

crates/query-engine/translation/src/translation/mutation/v2/update.rs

@@ -151,10 +151,7 @@ pub fn translate(
                 })
                 .collect::<Result<Vec<sql::ast::Expression>, Error>>()?;
 
-            let root_and_current_tables = helpers::RootAndCurrentTables {
-                root_table: table_name_and_reference.clone(),
-                current_table: table_name_and_reference,
-            };
+            let root_and_current_tables = helpers::TableScope::new(table_name_and_reference);
 
             // Set default constrainst
             let default_constraint = default_constraint();