implemented filter by nested collection (#573)

### What

Implementing an upcoming feature in ndc-spec:
hasura/ndc-spec#166

We want to support filtering by a field inside an array column.

For example: I want to get all institutions that where there exists a
person who's last name is "Hughes" among their staff.

This requires an update to ndc-spec and ndc-sdk-rs.

### How

When we get a request with an exists over a nested field collection, we
construct a nested field selection expression, unnest the results, and
use that as our new "current table".

Cargo.toml

@@ -31,9 +31,9 @@ similar_names = "allow"
 too_many_lines = "allow"
 
 [workspace.dependencies]
-ndc-models = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.5" }
-ndc-sdk = { git = "https://github.com/hasura/ndc-sdk-rs.git", tag = "v0.3.0" }
-ndc-test = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.5" }
+ndc-models = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.6" }
+ndc-sdk = { git = "https://github.com/hasura/ndc-sdk-rs.git", tag = "v0.4.0" }
+ndc-test = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.6" }
 
 anyhow = "1"
 async-trait = "0.1"

changelog.md

@@ -4,6 +4,9 @@
 
 ### Added
 
+- Support filtering by a nested field collection.
+  [#573](https://github.com/hasura/ndc-postgres/pull/573)
+
 ### Changed
 
 - Support setting ssl client certificate information and ssl root certificate independently.

crates/connectors/ndc-postgres/src/capabilities.rs

@@ -12,6 +12,9 @@ pub fn get_capabilities() -> models::Capabilities {
             aggregates: Some(models::LeafCapability {}),
             variables: Some(models::LeafCapability {}),
             explain: Some(models::LeafCapability {}),
+            exists: models::ExistsCapabilities {
+                nested_collections: Some(models::LeafCapability {}),
+            },
             nested_fields: models::NestedFieldCapabilities {
                 filter_by: Some(models::LeafCapability {}),
                 order_by: Some(models::LeafCapability {}),

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

@@ -95,6 +95,19 @@ impl TableSource {
             } => format!("{collection_name}.{}", field_path.0.join(".")),
         }
     }
+    /// Get collection name and field path from a source.
+    pub fn collection_name_and_field_path(&self) -> (models::CollectionName, FieldPath) {
+        match self {
+            TableSource::Collection(collection_name) => {
+                (collection_name.clone(), FieldPath(vec![]))
+            }
+            TableSource::NestedField {
+                collection_name,
+                field_path,
+                type_name: _,
+            } => (collection_name.clone(), field_path.clone()),
+        }
+    }
 }
 
 #[derive(Debug)]

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

@@ -285,48 +285,24 @@ fn translate_nested_field(
 
     // The recursive call to the next layer of fields
     let (nested_field_table_reference, nested_field_binding_alias) = {
-        match &current_table.source {
-            TableSource::Collection(collection_name) => {
-                let source = TableSource::NestedField {
-                    collection_name: collection_name.clone(),
-                    type_name: nested_field_type_name,
-                    field_path: FieldPath(vec![current_column_name.clone()]),
-                };
-                let nested_field_binding_alias = state.make_table_alias(source.name_for_alias());
-                (
-                    TableSourceAndReference {
-                        source,
-                        reference: sql::ast::TableReference::AliasedTable(
-                            nested_field_binding_alias.clone(),
-                        ),
-                    },
-                    nested_field_binding_alias,
-                )
-            }
-            TableSource::NestedField {
-                collection_name,
-                type_name: _,
-                field_path,
-            } => {
-                let mut field_path = field_path.0.clone();
-                field_path.push(current_column_name.clone());
-                let source = TableSource::NestedField {
-                    collection_name: collection_name.clone(),
-                    type_name: nested_field_type_name,
-                    field_path: FieldPath(field_path),
-                };
-                let nested_field_binding_alias = state.make_table_alias(source.name_for_alias());
-                (
-                    TableSourceAndReference {
-                        source,
-                        reference: sql::ast::TableReference::AliasedTable(
-                            nested_field_binding_alias.clone(),
-                        ),
-                    },
-                    nested_field_binding_alias,
-                )
-            }
-        }
+        let (collection_name, FieldPath(mut field_path)) =
+            current_table.source.collection_name_and_field_path();
+        field_path.push(current_column_name.clone());
+        let source = TableSource::NestedField {
+            collection_name,
+            type_name: nested_field_type_name,
+            field_path: FieldPath(field_path),
+        };
+        let nested_field_binding_alias = state.make_table_alias(source.name_for_alias());
+        (
+            TableSourceAndReference {
+                source,
+                reference: sql::ast::TableReference::AliasedTable(
+                    nested_field_binding_alias.clone(),
+                ),
+            },
+            nested_field_binding_alias,
+        )
     };
     // The FROM-clause to use for the next layer of fields returned by `translate_fields` below,
     // which brings each nested field into scope as separate columns in a sub query.

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

@@ -11,10 +11,10 @@ use super::root;
 use super::values;
 use super::variables;
 use crate::translation::error::Error;
-use crate::translation::helpers::wrap_in_field_path;
-use crate::translation::helpers::TableSource;
+use crate::translation::error::UnsupportedCapabilities;
 use crate::translation::helpers::{
-    ColumnInfo, CompositeTypeInfo, Env, RootAndCurrentTables, State, TableSourceAndReference,
+    wrap_in_field_path, ColumnInfo, CompositeTypeInfo, Env, FieldPath, RootAndCurrentTables, State,
+    TableSource, TableSourceAndReference,
 };
 use query_engine_metadata::metadata::database;
 use query_engine_sql::sql;
@@ -641,6 +641,106 @@ pub fn translate_exists_in_collection(
                 select: Box::new(select),
             })
         }
+        // Filter by a predicate related to columns inside an array field.
+        models::ExistsInCollection::NestedCollection {
+            column_name,
+            arguments,
+            field_path,
+        } => {
+            if !arguments.is_empty() {
+                Err(Error::CapabilityNotSupported(
+                    UnsupportedCapabilities::FieldArguments,
+                ))?;
+            }
+            let table = &root_and_current_tables.current_table;
+
+            // Get the table information from the metadata.
+            let collection_fields_info = env.lookup_fields_info(&table.source)?;
+
+            // The initial column we start with, it's reference, and it's source.
+            let column_info = collection_fields_info.lookup_column(&column_name)?;
+            let column_expr =
+                sql::ast::Expression::ColumnReference(sql::ast::ColumnReference::AliasedColumn {
+                    table: table.reference.clone(),
+                    column: sql::helpers::make_column_alias(column_info.name.0),
+                });
+
+            let source = {
+                let (collection_name, FieldPath(mut field_path)) =
+                    table.source.collection_name_and_field_path();
+                field_path.push(column_name.clone());
+                TableSource::NestedField {
+                    collection_name,
+                    type_name: underlying_type_name(column_info.r#type),
+                    field_path: FieldPath(field_path),
+                }
+            };
+            // Walk over the fields and construct a select expression over the fields
+            // E.g. `institution.staff.last_name`
+            // And the source, to be used to fetch information about the fields it contains.
+            let (select_expression, source) = field_path.iter().try_fold(
+                (column_expr, source),
+                |(expression, source), nested_field| {
+                    let collection_fields_info = env.lookup_fields_info(&source)?;
+
+                    let column_info = collection_fields_info.lookup_column(nested_field)?;
+
+                    let source = {
+                        let (collection_name, FieldPath(mut field_path)) =
+                            table.source.collection_name_and_field_path();
+                        field_path.push(column_name.clone());
+                        TableSource::NestedField {
+                            collection_name,
+                            type_name: underlying_type_name(column_info.r#type),
+                            field_path: FieldPath(field_path),
+                        }
+                    };
+
+                    Ok((
+                        sql::ast::Expression::NestedFieldSelect {
+                            expression: Box::new(expression),
+                            nested_field: sql::ast::NestedField(column_info.name.0),
+                        },
+                        source,
+                    ))
+                },
+            )?;
+
+            // Create an alias for the source which we will use as a reference.
+            let alias = state.make_table_alias(source.name_for_alias());
+
+            // Define a from clause from the field selection expression.
+            let from_source = sql::ast::From::Unnest {
+                expression: select_expression,
+                alias: alias.clone(),
+                columns: vec![],
+            };
+
+            // Define a new root and current table structure pointing the current table
+            // at the nested field.
+            let new_root_and_current_tables = RootAndCurrentTables {
+                root_table: root_and_current_tables.root_table.clone(),
+                current_table: TableSourceAndReference {
+                    reference: sql::ast::TableReference::AliasedTable(alias),
+                    source,
+                },
+            };
+
+            // Translate the predicate inside the exists.
+            let (exists_cond, exists_joins) = translate_expression_with_joins(
+                env,
+                state,
+                &new_root_and_current_tables,
+                predicate,
+            )?;
+
+            // Construct the `where exists` expression.
+            Ok(sql::helpers::where_exists_select(
+                from_source,
+                exists_joins,
+                sql::ast::Where(exists_cond),
+            ))
+        }
     }
 }
 
@@ -768,3 +868,12 @@ fn make_unnest_subquery(
     subquery.from = Some(subquery_from);
     sql::ast::Expression::CorrelatedSubSelect(Box::new(subquery))
 }
+
+/// Fetch the ndc-spec model type name referenced by this database type.
+fn underlying_type_name(typ: database::Type) -> models::TypeName {
+    match typ {
+        database::Type::ScalarType(scalar_type) => scalar_type.into(),
+        database::Type::CompositeType(typ) => typ,
+        database::Type::ArrayType(typ) => underlying_type_name(*typ),
+    }
+}

crates/tests/databases-tests/src/postgres/explain_tests.rs

@@ -73,6 +73,16 @@ mod query {
         insta::assert_snapshot!(result.details.query);
     }
 
+    #[tokio::test]
+    async fn filter_institution_by_nested_field_collection() {
+        let result = run_query_explain(
+            create_router().await,
+            "filter_institution_by_nested_field_collection",
+        )
+        .await;
+        insta::assert_snapshot!(result.details.query);
+    }
+
     mod native_queries {
         use super::super::super::common::create_router;
         use tests_common::assert::is_contained_in_lines;

crates/tests/databases-tests/src/postgres/query_tests.rs

@@ -297,6 +297,22 @@ mod predicates {
         insta::assert_json_snapshot!(result);
     }
 
+    #[tokio::test]
+    async fn filter_by_nested_field_collection() {
+        let result = run_query(create_router().await, "filter_by_nested_field_collection").await;
+        insta::assert_json_snapshot!(result);
+    }
+
+    #[tokio::test]
+    async fn filter_institution_by_nested_field_collection() {
+        let result = run_query(
+            create_router().await,
+            "filter_institution_by_nested_field_collection",
+        )
+        .await;
+        insta::assert_json_snapshot!(result);
+    }
+
     #[tokio::test]
     async fn select_where_array_relationship() {
         let result = run_query(create_router().await, "select_where_array_relationship").await;

crates/tests/databases-tests/src/postgres/snapshots/databases_tests__postgres__explain_tests__query__filter_institution_by_nested_field_collection.snap

@@ -0,0 +1,53 @@
+---
+source: crates/tests/databases-tests/src/postgres/explain_tests.rs
+expression: result.details.query
+---
+EXPLAIN
+SELECT
+  coalesce(json_agg(row_to_json("%5_universe")), '[]') AS "universe"
+FROM
+  (
+    SELECT
+      *
+    FROM
+      (
+        SELECT
+          coalesce(json_agg(row_to_json("%6_rows")), '[]') AS "rows"
+        FROM
+          (
+            SELECT
+              "%0_institution_institution"."id" AS "id",
+              "%0_institution_institution"."name" AS "name",
+              "%3_nested_fields_collect"."collected" AS "staff"
+            FROM
+              "institution"."institution" AS "%0_institution_institution"
+              LEFT OUTER JOIN LATERAL (
+                SELECT
+                  json_agg(row_to_json("%1_nested_fields")) AS "collected"
+                FROM
+                  (
+                    SELECT
+                      "%2_institution_institution.staff"."first_name" AS "first_name",
+                      "%2_institution_institution.staff"."last_name" AS "last_name",
+                      "%2_institution_institution.staff"."specialities" AS "specialities"
+                    FROM
+                      (
+                        SELECT
+                          (unnest("%0_institution_institution"."staff")).*
+                      ) AS "%2_institution_institution.staff"
+                  ) AS "%1_nested_fields"
+              ) AS "%3_nested_fields_collect" ON ('true')
+            WHERE
+              EXISTS (
+                SELECT
+                  1
+                FROM
+                  UNNEST("%0_institution_institution"."staff") AS "%4_institution_institution.staff"
+                WHERE
+                  (
+                    "%4_institution_institution.staff"."last_name" = cast($1 as "pg_catalog"."text")
+                  )
+              )
+          ) AS "%6_rows"
+      ) AS "%6_rows"
+  ) AS "%5_universe"