Support nested fields of object type (#404)

### What

This PR adds support for selecting nested fields that arise from
composite objects.

The case for nested fields which are arrays will be treated in a
followup PR.

### How

* Handling of field selection is now recursive and lives in its own
function.
* A dedicated case handling nested fields has been added. Each field at
each layer of nesting adds another lateral join

The SQL generated is of the form
```
///   SELECT
///     coalesce(json_agg(row_to_json("%6_rows")), '[]') AS "rows"
///   FROM
///     (
///       SELECT
///         "%3_nested_fields_collect"."collected" AS "result"
///       FROM
///         <current table> AS "%0_<current table>"
///         LEFT OUTER JOIN LATERAL (
///           SELECT
///             ("%0_<current table>"."<composite column>").*
///         ) AS "%2_nested_field_bound" ON ('true')
///         LEFT OUTER JOIN LATERAL (
///           SELECT
///             row_to_json("%4_nested_fields") AS "collected"
///           FROM
///             (
///               SELECT
///                 "%2_nested_field_bound"."<nested column>" AS "<nested field alias>"
///             ) AS "%4_nested_fields"
///         ) AS "%3_nested_fields_collect" ON ('true')
///     ) AS "%6_rows"
```

crates/query-engine/sql/src/sql/ast.rs

@@ -81,6 +81,7 @@ pub enum Returning {
 pub enum SelectList {
     SelectList(Vec<(ColumnAlias, Expression)>),
     SelectStar,
+    SelectStarComposite(Expression),
 }
 
 /// A FROM clause

crates/query-engine/sql/src/sql/convert.rs

@@ -92,6 +92,11 @@ impl SelectList {
             SelectList::SelectStar => {
                 sql.append_syntax("*");
             }
+            SelectList::SelectStarComposite(expr) => {
+                sql.append_syntax("(");
+                expr.to_sql(sql);
+                sql.append_syntax(").*");
+            }
         }
     }
 }

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

@@ -80,6 +80,20 @@ pub fn make_column_alias(name: String) -> ColumnAlias {
 
 // SELECTs //
 
+/// Build a simple 'SELECT (exp).*'
+pub fn select_composite(exp: Expression) -> Select {
+    Select {
+        with: empty_with(),
+        select_list: SelectList::SelectStarComposite(exp),
+        from: None,
+        joins: vec![],
+        where_: Where(empty_where()),
+        group_by: empty_group_by(),
+        order_by: empty_order_by(),
+        limit: empty_limit(),
+    }
+}
+
 /// Build a simple select with a select list and the rest are empty.
 pub fn simple_select(select_list: Vec<(ColumnAlias, Expression)>) -> Select {
     Select {
@@ -471,6 +485,8 @@ pub fn select_mutation_rowset(
     final_select
 }
 
+/// Turn all rows of a query result into a single json array of objects.
+///
 /// Wrap a query that returns multiple rows in the following format:
 ///
 /// ```sql
@@ -507,6 +523,26 @@ pub fn select_rows_as_json(
     select
 }
 
+/// Turn each row of a query result into a json object.
+///
+/// ```sql
+/// SELECT row_to_json(<table_alias>) AS <column_alias>
+/// FROM <query> as <table_alias>
+/// ```
+pub fn select_row_as_json(
+    row_select: Select,
+    column_alias: ColumnAlias,
+    table_alias: TableAlias,
+) -> Select {
+    let expression = Expression::RowToJson(TableReference::AliasedTable(table_alias.clone()));
+    let mut select = simple_select(vec![(column_alias, expression)]);
+    select.from = Some(From::Select {
+        select: Box::new(row_select),
+        alias: table_alias,
+    });
+    select
+}
+
 /// Wrap a query that returns multiple rows in the following format:
 ///
 /// ```sql

crates/query-engine/sql/src/sql/rewrites/constant_folding.rs

@@ -16,6 +16,9 @@ pub fn normalize_select(mut select: Select) -> Select {
     // select list
     select.select_list = match select.select_list {
         SelectList::SelectStar => SelectList::SelectStar,
+        SelectList::SelectStarComposite(exp) => {
+            SelectList::SelectStarComposite(normalize_expr(exp))
+        }
         SelectList::SelectList(vec) => SelectList::SelectList(
             vec.into_iter()
                 .map(|(alias, expr)| (alias, normalize_expr(expr)))

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

@@ -1,6 +1,6 @@
 //! Errors for translation.
 
-use query_engine_metadata::metadata::database;
+use query_engine_metadata::metadata::{database, Type};
 
 /// A type for translation errors.
 #[derive(Debug, Clone)]
@@ -31,6 +31,10 @@ pub enum Error {
     NoProcedureResultFieldsRequested,
     UnexpectedStructure(String),
     InternalError(String),
+    NestedFieldNotOfCompositeType {
+        field_name: String,
+        actual_type: Type,
+    },
 }
 
 /// Capabilities we don't currently support.
@@ -132,6 +136,16 @@ impl std::fmt::Display for Error {
             Error::NonScalarTypeUsedInOperator { r#type } => {
                 write!(f, "Non-scalar-type used in operator: {:?}", r#type)
             }
+            Error::NestedFieldNotOfCompositeType {
+                field_name,
+                actual_type,
+            } => {
+                write!(
+                    f,
+                    "Nested field '{}' not of composite type. Actual type: {:?}",
+                    field_name, actual_type
+                )
+            }
         }
     }
 }

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

@@ -88,6 +88,16 @@ pub enum CollectionInfo<'env> {
     },
 }
 
+#[derive(Debug)]
+/// Metadata information about a specific collection.
+pub enum CompositeTypeInfo<'env> {
+    CollectionInfo(CollectionInfo<'env>),
+    CompositeTypeInfo {
+        name: String,
+        info: metadata::CompositeType,
+    },
+}
+
 impl<'request> Env<'request> {
     /// Create a new Env by supplying the metadata and relationships.
     pub fn new(
@@ -105,6 +115,29 @@ impl<'request> Env<'request> {
     }
 
     /// Lookup a collection's information in the metadata.
+
+    pub fn lookup_composite_type(
+        &self,
+        type_name: &'request str,
+    ) -> Result<CompositeTypeInfo<'request>, Error> {
+        let it_is_a_collection = self.lookup_collection(type_name);
+
+        match it_is_a_collection {
+            Ok(collection_info) => Ok(CompositeTypeInfo::CollectionInfo(collection_info)),
+            Err(Error::CollectionNotFound(_)) => {
+                let its_a_type = self.metadata.composite_types.0.get(type_name).map(|t| {
+                    CompositeTypeInfo::CompositeTypeInfo {
+                        name: t.name.clone(),
+                        info: t.clone(),
+                    }
+                });
+
+                its_a_type.ok_or(Error::CollectionNotFound(type_name.to_string()))
+            }
+            Err(err) => Err(err),
+        }
+    }
+
     pub fn lookup_collection(
         &self,
         collection_name: &'request str,
@@ -209,6 +242,28 @@ impl CollectionInfo<'_> {
     }
 }
 
+impl CompositeTypeInfo<'_> {
+    /// Lookup a column in a collection.
+    pub fn lookup_column(&self, column_name: &str) -> Result<ColumnInfo, Error> {
+        match self {
+            CompositeTypeInfo::CollectionInfo(collection_info) => {
+                collection_info.lookup_column(column_name)
+            }
+            CompositeTypeInfo::CompositeTypeInfo { name, info } => info
+                .fields
+                .get(column_name)
+                .map(|field_info| ColumnInfo {
+                    name: sql::ast::ColumnName(field_info.name.clone()),
+                    r#type: field_info.r#type.clone(),
+                })
+                .ok_or(Error::ColumnNotFoundInCollection(
+                    column_name.to_string(),
+                    name.clone(),
+                )),
+        }
+    }
+}
+
 impl Default for State {
     fn default() -> State {
         State {